microsoft/airflow-providers-microsoft-fabric

microsoft/airflow-providers-microsoft-fabric

Description: Repository containing the source code for the Microsoft Fabric provider for Airflow.

Language: Python

License: MIT

Stars: 13

Forks: 6

Open issues: 7

Created: 2025-06-24T02:20:52Z

Pushed: 2026-06-09T04:45:12Z

Default branch: main

Fork: no

Archived: no

README:

Apache Airflow Provider for Microsoft Fabric

Introduction

An Apache Airflow provider package for Microsoft Fabric. It enables Data and Analytics engineers to trigger, monitor, and orchestrate Microsoft Fabric items from Airflow DAGs.

Microsoft Fabric is an end-to-end analytics and data platform designed for enterprises that require a unified solution. It encompasses data movement, processing, ingestion, transformation, real-time event routing, and report building.

Installation

pip install apache-airflow-providers-microsoft-fabric

Authentication

The provider supports the following authentication methods:

| Method | Description | |--------|-------------| | Service Principal (SPN) | Uses client_id and client_secret for automated pipelines. | | User Token | Uses a refresh token obtained via Microsoft OAuth. |

Connection setup

Create a connection in Airflow with the following settings:

| Field | Value | |-------|-------| | Connection Id | Your connection name | | Connection Type | microsoft-fabric | | Login | Client ID of your service principal or Entra ID app | | Password | Client secret (SPN) or refresh token (User Token) | | Extra | {"tenantId": "", "auth_type": "spn"} |

> For user token auth, set auth_type to token and provide a refresh token as the password.

Operators

MSFabricRunJobOperator

Triggers and monitors a Fabric job item (notebook, pipeline, Spark job definition).

from airflow.providers.microsoft.fabric.operators.run_item import MSFabricRunJobOperator

run_notebook = MSFabricRunJobOperator(
task_id="run_fabric_notebook",
workspace_id="",
item_id="",
fabric_conn_id="fabric_conn_id",
job_type="RunNotebook",
wait_for_termination=True,
deferrable=True,
)

| Parameter | Type | Default | Description | |-----------|------|---------|-------------| | workspace_id | str | — | The workspace ID. | | item_id | str | — | The item ID (notebook, pipeline, etc.). | | fabric_conn_id | str | — | Airflow connection ID for Fabric. | | job_type | str | — | The item type to run (see table below). | | wait_for_termination | bool | True | Wait for the item run to complete. | | timeout | int | 604800 | Timeout in seconds (default: 7 days). | | check_interval | int | 60 | Polling interval in seconds. | | max_retries | int | 5 | Max polling retries after job start. | | retry_delay | int | 1 | Delay between polling retries (seconds). | | deferrable | bool | False | Run the operator in deferrable mode. | | job_params | dict | None | Parameters passed to the job. |

Supported `job_type` values:

| Fabric item | job_type values | API | Required permission | |-------------|-------------------|-----|---------------------| | Notebook | RunNotebook, Notebook | Fabric Job Scheduler | Item.Execute.All or Notebook.Execute.All | | Data Pipeline | RunPipeline, Pipeline | Fabric Job Scheduler | Item.Execute.All or DataPipeline.Execute.All | | Spark Job Definition | RunSparkJob, SparkJob | Fabric Job Scheduler | Item.Execute.All or SparkJobDefinition.Execute.All | | DBT Job | DataBuildToolJob, DBT | Fabric Job Scheduler | Item.Execute.All | | Copy Job | CopyJobs | Fabric Job Scheduler | Item.Execute.All | | Materialized Lake Views | RefreshMaterializedLakeViews | Fabric Job Scheduler | Item.Execute.All |

Sending parameters

The job_params parameter accepts a JSON string. Use the built-in helper classes to construct it.

Notebook parameters — use MSFabricNotebookJobParameters:

from airflow.providers.microsoft.fabric.operators.run_item.notebook_parameters import (
MSFabricNotebookJobParameters,
)

params = (
MSFabricNotebookJobParameters()
.set_parameter("input_path", "/data/raw")
.set_parameter("threshold", 0.9) # auto-inferred as float
.set_parameter("debug_mode", True) # auto-inferred as bool
.set_conf("spark.executor.memory", "4g")
.set_environment(environment_name="my-env")
.set_default_lakehouse(name="bronze", id="", workspace_id="")
.set_use_starter_pool(False)
.set_use_workspace_pool("my-pool")
)

run_notebook = MSFabricRunJobOperator(
task_id="run_notebook",
workspace_id="",
item_id="",
fabric_conn_id="fabric_conn_id",
job_type="RunNotebook",
job_params=params.to_json(),
deferrable=True,
)

Pipeline parameters — use MSFabricPipelineJobParameters:

from airflow.providers.microsoft.fabric.operators.run_item.pipeline_parameters import (
MSFabricPipelineJobParameters,
)

params = (
MSFabricPipelineJobParameters()
.set_parameter("source_table", "orders")
.set_parameter("batch_size", 1000)
.set_parameter("full_refresh", True)
)

run_pipeline = MSFabricRunJobOperator(
task_id="run_pipeline",
workspace_id="",
item_id="",
fabric_conn_id="fabric_conn_id",
job_type="Pipeline",
job_params=params.to_json(),
deferrable=True,
)

Spark Job / DBT / Copy Job / Materialized Lake Views — pass a raw JSON string if the API expects executionData:

run_spark = MSFabricRunJobOperator(
task_id="run_spark_job",
workspace_id="",
item_id="",
fabric_conn_id="fabric_conn_id",
job_type="SparkJob",
job_params='{"executionData": {}}',
deferrable=True,
)

MSFabricRunSemanticModelRefreshOperator

Triggers a Power BI semantic model refresh via the Power BI REST API.

| | | |---|---| | API | Power BI REST API (https://api.powerbi.com) | | Required permission | Dataset.ReadWrite.All |

from airflow.providers.microsoft.fabric.operators.run_item import MSFabricRunSemanticModelRefreshOperator

refresh = MSFabricRunSemanticModelRefreshOperator(
task_id="refresh_model",
workspace_id="",
item_id="",
fabric_conn_id="fabric_conn_id",
deferrable=True,
)

MSFabricRunUserDataFunctionOperator

Invokes a User Data Function in Microsoft Fabric. This operator is synchronous only — the API returns immediately with the function output.

| | | |---|---| | API | Fabric REST API (https://api.fabric.microsoft.com) | | Required permission | Execute permission…