Skip to content

Overview

fabric-data-pipelines

Get started API reference

fabric-data-pipelines is a Python library for building Microsoft Fabric data pipelines as code.

Fabric pipelines are JSON definitions under the hood. This library lets you work at a higher level: compose typed Python activities, wire dependencies in code, validate the graph, and export either raw pipeline JSON or Fabric Git item folders.

Why teams use it

  • Avoid hand-authoring nested Fabric pipeline JSON.
  • Keep pipeline definitions in Git as code.
  • Reuse patterns for orchestration, ETL, and scheduling.
  • Generate the folder structure Fabric expects for Git integration.
  • Catch dependency and graph mistakes before export.

Before / after

{
  "properties": {
    "activities": [
      {
        "name": "Truncate_Landing_customers",
        "type": "Script",
        "dependsOn": [],
        "policy": {
          "timeout": "0.12:00:00",
          "retry": 0,
          "retryIntervalInSeconds": 30,
          "secureOutput": false,
          "secureInput": false
        },
        "externalReferences": {
          "connection": "@pipeline().libraryVariables.Demo_ETL_Library_Landing"
        },
        "typeProperties": {
          "database": "Landing",
          "scripts": [
            {
              "text": {
                "value": "TRUNCATE TABLE sales.customers",
                "type": "Expression"
              },
              "type": "Query"
            }
          ],
          "scriptBlockExecutionTimeout": "02:00:00"
        }
      },
      {
        "name": "Copy_customers_to_Landing",
        "type": "Copy",
        "dependsOn": [
          {
            "activity": "Truncate_Landing_customers",
            "dependencyConditions": [
              "Succeeded"
            ]
          }
        ],
        "policy": {
          "timeout": "0.12:00:00",
          "retry": 0,
          "retryIntervalInSeconds": 30,
          "secureOutput": false,
          "secureInput": false
        },
        "typeProperties": {
          "source": {
            "type": "SqlMISource",
            "datasetSettings": {
              "type": "AzureSqlMITable",
              "annotations": [],
              "schema": [],
              "typeProperties": {
                "database": "SourceDb"
              },
              "externalReferences": {
                "connection": "@pipeline().libraryVariables.Demo_ETL_Library_SourceDb"
              }
            },
            "sqlReaderQuery": "SELECT\n   CAST([customer_id] AS NVARCHAR (12)) AS [customer_id]\n   , [region]\n   , [segment]\n   , [updated_at]\nFROM [dbo].[customers];\n",
            "partitionOption": "None"
          },
          "sink": {
            "type": "SqlMISink",
            "datasetSettings": {
              "type": "AzureSqlMITable",
              "annotations": [],
              "schema": [],
              "typeProperties": {
                "database": "Landing",
                "table": "customers",
                "schema": "sales"
              },
              "externalReferences": {
                "connection": "@pipeline().libraryVariables.Demo_ETL_Library_Landing"
              }
            },
            "writeBehavior": "insert",
            "sqlWriterUseTableLock": true
          },
          "translator": {
            "type": "TabularTranslator",
            "mappings": [
              {
                "source": {
                  "name": "customer_id",
                  "type": "String",
                  "physicalType": "nvarchar"
                },
                "sink": {
                  "name": "customer_id",
                  "type": "String",
                  "physicalType": "nvarchar"
                }
              },
              {
                "source": {
                  "name": "region",
                  "type": "String",
                  "physicalType": "nvarchar"
                },
                "sink": {
                  "name": "region",
                  "type": "String",
                  "physicalType": "nvarchar"
                }
              },
              {
                "source": {
                  "name": "segment",
                  "type": "String",
                  "physicalType": "nvarchar"
                },
                "sink": {
                  "name": "segment",
                  "type": "String",
                  "physicalType": "nvarchar"
                }
              },
              {
                "source": {
                  "name": "updated_at",
                  "type": "String",
                  "physicalType": "nvarchar"
                },
                "sink": {
                  "name": "updated_at",
                  "type": "String",
                  "physicalType": "nvarchar"
                }
              }
            ],
            "typeConversion": true,
            "typeConversionSettings": {
              "allowDataTruncation": true,
              "treatBooleanAsNumber": false
            }
          },
          "enableStaging": false
        }
      }
    ],
    "description": "Truncate Landing.sales.customers, then load a typed subset from SourceDb.",
    "libraryVariables": {
      "Demo_ETL_Library_Landing": {
        "type": "String",
        "variableName": "Landing",
        "libraryName": "Demo_ETL_Library"
      },
      "Demo_ETL_Library_SourceDb": {
        "type": "String",
        "variableName": "SourceDb",
        "libraryName": "Demo_ETL_Library"
      }
    }
  }
}

from fabric_data_pipelines import (
    AzureSqlMITable, ColumnMapping, ColumnRef, Copy, ExternalReferences,
    LibraryVariable, Pipeline, Script, ScriptBlock, SqlMISink, SqlMISource,
    TabularTranslator, TypeConversionSettings, expr,
)

LANDING = expr.library_variable("Demo_ETL_Library_Landing")
SOURCE = expr.library_variable("Demo_ETL_Library_SourceDb")
COLUMNS = ["customer_id", "region", "segment", "updated_at"]

truncate = Script(
    name="Truncate_Landing_customers",
    database="Landing",
    scripts=[ScriptBlock(
        text={"value": "TRUNCATE TABLE sales.customers", "type": "Expression"},
        type="Query",
    )],
    external_references=ExternalReferences(connection=LANDING),
)
copy = Copy(
    name="Copy_customers_to_Landing",
    source=SqlMISource(
        sql_reader_query=SELECT_SQL,
        dataset_settings=AzureSqlMITable(database="SourceDb", connection=SOURCE),
    ),
    sink=SqlMISink(
        write_behavior="insert",
        dataset_settings=AzureSqlMITable(
            database="Landing", schema_name="sales", table="customers", connection=LANDING,
        ),
    ),
    translator=TabularTranslator(
        mappings=[
            ColumnMapping(
                source=ColumnRef(name=col, type="String", physical_type="nvarchar"),
                sink=ColumnRef(name=col, type="String", physical_type="nvarchar"),
            )
            for col in COLUMNS
        ],
        type_conversion=True,
        type_conversion_settings=TypeConversionSettings(allow_data_truncation=True),
    ),
)
truncate.then(copy)
pipeline = Pipeline(name="Landing_customers", activities=[truncate, copy], ...)

Full runnable source: examples/landing_truncate_copy.py.

What you can build

This package supports both simple and production-style workflows:

  • quick pipelines such as Wait -> Notebook
  • ETL and ELT flows using Copy, Script, Lookup, and ExecutePipeline
  • control flow with IfCondition, ForEach, Switch, and Until
  • scheduled pipeline items with .schedules
  • Git-backed Fabric items with stable .platform metadata

Two output modes

Raw definition output

Use Pipeline.to_json() or Pipeline.save() when you need the pipeline definition itself.

Fabric Git item output

Use Pipeline.save_item() or save_workspace() when your repository is the source of truth and Fabric should consume *.DataPipeline/ folders.

Start here

Sponsored by datalyft