All posts

Author Fabric Data Pipelines as Code

I've been using Microsoft Fabric quite extensively for the past two years. In the beginning I was disappointed, to say the least. That was mainly due to the missing support for Infrastructure as Code.

featured-image.png

Shortly after I published that article, the official Terraform provider was released, and over time the platform kept developing into something workable. But while it is gradually getting better, it is still not completely where I need it to be.

One of the areas where Fabric is still lacking is with Data Pipelines. Fabric Data Pipelines are basically rebranded Azure Data Factory pipelines that live inside Fabric and integrate reasonably well with the other items in a workspace. The problem I ran into is that there are exactly two ways to build and change them: the GUI (i.e. ClickOps) or editing the underlying JSON, either by hand or through the MCP. Neither is ideal, and both get worse as complexity and dependencies grow.

I would rather use code to create and structure these pipelines. So I built fabric-data-pipelines: a Python library that lets you author Fabric Data Pipelines as code.

The actual problem

A Fabric Data Pipeline is a JSON document. Activities are objects in a list, execution order lives in nested dependsOn arrays, and every connector detail sits somewhere inside that tree.

When you build pipelines in the GUI you never see this, and it works fine until:

  • you need the same landing pattern for twelve tables and end up copy-pasting or duplicating something twelve times
  • you rename an activity and silently break three dependencies
  • you have to review a script in the JSON that is on a single line, which means you have to copy, paste and parse it before understanding what’s going on (see image below)
  • you want to promote from dev to prod and discover half the definition is environment-specific
  • you want to reuse a SQL query, a Teams message template, or a shared policy and find there is no module boundary: everything is inline in one serialization tree

fabric-data-pipelines-before-after.png Have you ever reviewed a change to a SQL script that’s on a single line, stored in a JSON?

The solution

Making Python the DAG authoring interface solves these problems. Twelve landing tables become a loop or a factory function. Renames are refactors your IDE handles. Reviews diff readable Python and scripts instead of generated JSON. Environment-specific values become parameters. Shared queries and templates become imports, because now you finally have module boundaries.

Your first pipeline

First, install the latest version of the library through pip or uv:

pip install fabric-data-pipelines
# or
uv add fabric-data-pipelines

Then, create a Python file, like below:

from fabric_data_pipelines import Notebook, Pipeline, Wait

wait = Wait(name="Wait_For_Upstream", wait_time_in_seconds=30)

transform = Notebook(
    name="Transform_Silver_Sales",
    notebook_id="f67ac10b-58cc-4372-a567-0e05b2c3d479",
    workspace_id="b81g1d9f-33c7-462d-b818-2e4908a123f3",
)
wait.then(transform)

pipeline = Pipeline(
    name="Daily_Silver_Sales_Transform",
    activities=[wait, transform],
)
pipeline.save("daily_silver_sales_transform.json")

This code emits a valid Fabric pipeline definition. The mental model stays the same for everything after this:

  1. Create typed activities for Data Movement (Copy, Lookup, etc.), Control Flow (ForEach, IfCondition, Fail, etc.) or Execution (Notebook, Script, etc.). See the full list of supported activities. If an activity is not supported yet, you always have the option to include it as a RawActivity, making sure you are never blocked.
  2. Declare execution order in code, more on that below
  3. Build a Pipeline object, where you set your defaults
  4. Export as JSON or Fabric folder

This library comes with a few features out of the box, some of which are highlighted below.

Dependencies that read like a workflow

Orchestration logic should be readable. This is the part that inspired me to build the library in the first place, as I wanted Airflow DAG like code for Fabric Data Pipelines.

lookup.then(copy).then(notebook)

# or, for linear flows
lookup >> copy >> notebook

.then() defaults to a Succeeded dependency. The other Fabric conditions are there when you need them:

copy.then(notify, on="Failed")
copy.then(cleanup, on="Completed")

Fan-in is also supported through:

join.after(copy_a, copy_b)

Validation before deployment, not after

Another benefit of using Python as interface is that everything gets validated before serialization: duplicate activity names, unknown dependency targets, dependency cycles, and cross-scope dependencies which Fabric refuses.

Control-flow activities such as IfCondition, ForEach, Switch, and Until have their own nested scopes. Wiring a dependency across scope boundaries is a classic mistake, and now it fails in your terminal or in CI instead of during a Fabric deployment.

Easier metadata-driven pipelines

The claim above was that twelve landing tables become a loop or a factory function. Here is what that actually looks like:

from fabric_data_pipelines import (
    AzureSqlMITable, Copy, ExternalReferences, Pipeline,
    Script, ScriptBlock, SqlMISink, SqlMISource, expr,
)

LANDING = expr.library_variable("Demo_ETL_Library_Landing")
SOURCE = expr.library_variable("Demo_ETL_Library_SourceDb")

def land_table(schema: str, table: str) -> tuple[Script, Copy]:
    truncate = Script(
        name=f"Truncate_Landing_{table}",
        database="Landing",
        scripts=[ScriptBlock(
            text={"value": f"TRUNCATE TABLE {schema}.{table}", "type": "Expression"},
            type="Query",
        )],
        external_references=ExternalReferences(connection=LANDING),
    )
    copy = Copy(
        name=f"Copy_{table}_to_Landing",
        source=SqlMISource(
            sql_reader_query=f"SELECT * FROM [{schema}].[{table}];",
            dataset_settings=AzureSqlMITable(database="SourceDb", connection=SOURCE),
        ),
        sink=SqlMISink(
            write_behavior="insert",
            dataset_settings=AzureSqlMITable(
                database="Landing", schema_name=schema, table=table, connection=LANDING,
            ),
        ),
    )
    truncate.then(copy)
    return truncate, copy

TABLES = ["customers", "orders", "order_lines", "products",
          "invoices", "payments", "shipments", "returns",
          "suppliers", "inventory", "promotions", "stores"]

pipeline = Pipeline(
    name="Landing_Daily_Load",
    activities=[activity for table in TABLES for activity in land_table("sales", table)],
)

pipeline.save("landing_daily_load.json")

One function, twelve tables, and a readable diff when the pattern changes.

And because the table list is just Python data, it does not have to be hardcoded. Load it from a YAML file or a metadata table in your warehouse, and you have a metadata-driven ingestion framework in about thirty lines.

This is a classic metadata-driven pattern, let’s compare it with the GUI workflow.

In Data Pipelines: a Lookup reads the control table, a ForEach loops at runtime, dynamic expressions fill in the blanks. The logic only exists while the pipeline runs. Testing means deploying, debugging means the monitoring view.

In Python: the loop runs before deployment. The logic is code you can unit test, the output is a definition you can diff, and by the time anything reaches Fabric, there is nothing left to guess.

If you do want runtime iteration, Fabric’s ForEach is fully supported, see the parameterized ELT controller example. You can find more production-style patterns, including watermark lookups and lock-guarded orchestration, in the ETL Patterns guide.

Two implementations

fabric-data-pipelines has two export modes, so you can pick the one that matches your deployment strategy. Use raw definition when you want the JSON itself, for inspection, tests, or an API call:

json_text = pipeline.to_json()
pipeline.save("daily_load.json")

Use Fabric Git item folder when your repository is the source of truth:

pipeline.save_item("out")

This method writes exactly the shape Fabric’s Git integration expects:

Gold_Finance_Metrics_Refresh.DataPipeline/
  pipeline-content.json
  .platform
  .schedules

And for a full workspace:

from fabric_data_pipelines import save_workspace

save_workspace([pipeline_a, pipeline_b], "workspace")

Schedule support

Schedules are part of the model, too. If a pipeline has schedules configured, .schedules is written for you.

One detail worth calling out: Fabric tracks item identity through logicalId in .platform. Set logical_id explicitly on your Pipeline and renames keep the same Fabric item instead of creating a new one. Omit it and the library derives one from the name. If a .platform file already exists, the existing logicalId is preserved (this is the kind of thing you only learn by breaking it once).

What this is not

This is purely an authoring and export library. It does not deploy for you, it does not replace Terraform, fabric-cli or fabric-cicd. It only produces the definitions. Your existing deployment pipeline moves them.

Migrating your pipelines

If you’re interested in this library you’re probably not starting from zero, so the library also supports importing existing pipeline JSONs and transforming them into Python code.

The import is round-trip safe: load a pipeline definition, export it again, and you get equivalent JSON back. Nothing is dropped or reinterpreted along the way, so you can import production pipelines without worrying that something changes behind your back.

Though be aware that the generated code is not pretty, but it is correct. It serves as a working starting point instead of a rewrite. Clean it up once, and from then on the Python becomes the source of truth.

See the migration guide for the full walkthrough.

Try it

If you are running Fabric pipelines at any scale, I would like to know which activities and patterns you need next. Issues are welcome, and so is your opinion on whether this solves your problem or not.

References: