🌬️ Apache Airflow Job - Managed Orchestration in Fabric¶

# Set in Airflow Job configuration, NOT in DAG code
# These are available to all DAGs in the job

# Fabric workspace settings
FABRIC_WORKSPACE_ID = "your-workspace-id"
FABRIC_LAKEHOUSE_NAME = "lh_bronze"

# Business configuration
CTR_THRESHOLD = "10000"
RETENTION_DAYS = "365"
ENVIRONMENT = "prod"

"""
Casino Medallion Pipeline
Orchestrates the full bronze → silver → gold data flow for casino data.
"""
from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.providers.microsoft.fabric.operators.notebook import FabricNotebookOperator
from airflow.providers.microsoft.fabric.operators.pipeline import FabricPipelineOperator
from airflow.providers.microsoft.fabric.sensors.lakehouse import FabricLakehouseSensor

default_args = {
    "owner": "data-engineering",
    "depends_on_past": False,
    "retries": 2,
    "retry_delay": timedelta(minutes=5),
    "email_on_failure": True,
    "email": ["data-team@contoso.com"],
}

@dag(
    dag_id="casino_medallion_pipeline",
    description="Daily medallion pipeline for casino data",
    schedule="0 6 * * *",  # 6 AM daily
    start_date=datetime(2026, 1, 1),
    catchup=False,
    default_args=default_args,
    tags=["casino", "medallion", "daily"],
)
def casino_medallion():

    # ---- BRONZE LAYER ----

    ingest_slot_telemetry = FabricNotebookOperator(
        task_id="bronze_slot_telemetry",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="01_bronze_slot_telemetry",
        parameters={"date": "{{ ds }}", "env": "{{ var.value.environment }}"},
    )

    ingest_table_games = FabricNotebookOperator(
        task_id="bronze_table_games",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="02_bronze_table_game_results",
        parameters={"date": "{{ ds }}"},
    )

    ingest_player_tracking = FabricNotebookOperator(
        task_id="bronze_player_tracking",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="03_bronze_player_tracking",
        parameters={"date": "{{ ds }}"},
    )

    # ---- DATA QUALITY GATE ----

    @task
    def validate_bronze(ds=None):
        """Run Great Expectations on bronze tables."""
        import great_expectations as gx

        context = gx.get_context()
        results = context.run_checkpoint(checkpoint_name="bronze_checkpoint")

        if not results.success:
            raise ValueError(f"Bronze quality check failed for {ds}")
        return {"status": "passed", "date": ds}

    bronze_quality = validate_bronze()

    # ---- SILVER LAYER ----

    cleanse_slots = FabricNotebookOperator(
        task_id="silver_slot_cleansed",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="01_silver_slot_cleansed",
        parameters={"date": "{{ ds }}"},
    )

    cleanse_table_games = FabricNotebookOperator(
        task_id="silver_table_games_cleansed",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="02_silver_table_games_cleansed",
        parameters={"date": "{{ ds }}"},
    )

    cleanse_players = FabricNotebookOperator(
        task_id="silver_player_profiles",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="03_silver_player_profiles",
        parameters={"date": "{{ ds }}"},
    )

    # ---- GOLD LAYER ----

    gold_slot_kpis = FabricNotebookOperator(
        task_id="gold_slot_performance",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="01_gold_slot_performance",
        parameters={"date": "{{ ds }}"},
    )

    gold_compliance = FabricNotebookOperator(
        task_id="gold_compliance_report",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="05_gold_compliance_monitoring",
        parameters={"date": "{{ ds }}"},
    )

    gold_revenue = FabricNotebookOperator(
        task_id="gold_revenue_dashboard",
        workspace_id="{{ var.value.workspace_id }}",
        notebook_name="03_gold_revenue_analysis",
        parameters={"date": "{{ ds }}"},
    )

    # ---- DEPENDENCIES ----

    # Bronze tasks run in parallel
    [ingest_slot_telemetry, ingest_table_games, ingest_player_tracking] >> bronze_quality

    # Silver depends on bronze quality gate
    bronze_quality >> [cleanse_slots, cleanse_table_games, cleanse_players]

    # Gold depends on all silver
    [cleanse_slots, cleanse_table_games, cleanse_players] >> [gold_slot_kpis, gold_compliance, gold_revenue]

casino_medallion()

"""
Federal Agency Dynamic Pipeline
Generates tasks dynamically based on configured agencies.
"""
from airflow.decorators import dag, task
from airflow.providers.microsoft.fabric.operators.notebook import FabricNotebookOperator

AGENCIES = ["usda", "sba", "noaa", "epa", "doi", "doj"]
LAYERS = ["bronze", "silver", "gold"]

@dag(
    dag_id="federal_multi_agency_pipeline",
    schedule="0 4 * * 1",  # Weekly on Monday at 4 AM
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["federal", "multi-agency"],
)
def federal_pipeline():

    @task
    def check_data_freshness(agency: str) -> dict:
        """Check if source data has been updated since last run."""
        from fabric.onelake import OneLakeClient
        client = OneLakeClient()
        stats = client.get_table_stats(f"lh_bronze.Lakehouse/Tables/{agency}")
        return {"agency": agency, "needs_refresh": stats["last_modified_hours_ago"] < 168}

    for agency in AGENCIES:
        freshness = check_data_freshness(agency)

        bronze = FabricNotebookOperator(
            task_id=f"bronze_{agency}",
            notebook_name=f"10_bronze_{agency}_ingest",
            parameters={"agency": agency, "date": "{{ ds }}"},
        )

        silver = FabricNotebookOperator(
            task_id=f"silver_{agency}",
            notebook_name=f"10_silver_{agency}_cleansed",
            parameters={"agency": agency},
        )

        gold = FabricNotebookOperator(
            task_id=f"gold_{agency}",
            notebook_name=f"10_gold_{agency}_analytics",
            parameters={"agency": agency},
        )

        freshness >> bronze >> silver >> gold

federal_pipeline()

from airflow.providers.microsoft.fabric.sensors.lakehouse import FabricLakehouseSensor

# Wait for upstream data to arrive before processing
wait_for_slot_data = FabricLakehouseSensor(
    task_id="wait_for_slot_data",
    workspace_id="{{ var.value.workspace_id }}",
    lakehouse_name="lh_bronze",
    table_name="slot_telemetry",
    partition_filter={"date": "{{ ds }}"},
    min_row_count=1000,
    poke_interval=300,  # Check every 5 minutes
    timeout=3600,  # Give up after 1 hour
    mode="reschedule",  # Free up worker while waiting
)

from airflow.providers.microsoft.fabric.operators.rest import FabricRestOperator

refresh_dataset = FabricRestOperator(
    task_id="refresh_semantic_model",
    method="POST",
    endpoint=f"/v1/workspaces/{workspace_id}/semanticModels/{model_id}/refresh",
    json_body={"type": "full"},
    expected_status_code=202,
)

# In your DAG, emit custom metrics to Fabric
@task
def emit_metrics(run_stats: dict):
    """Push DAG run metrics to Eventhouse for unified monitoring."""
    from fabric.eventhouse import EventhouseClient

    client = EventhouseClient(database="platform_monitoring")
    client.ingest({
        "dag_id": "casino_medallion_pipeline",
        "run_date": run_stats["date"],
        "duration_seconds": run_stats["duration"],
        "task_count": run_stats["tasks"],
        "failed_tasks": run_stats["failures"],
        "timestamp": datetime.utcnow().isoformat()
    }, table="airflow_runs")

# tests/test_casino_medallion.py
import pytest
from airflow.models import DagBag

@pytest.fixture
def dagbag():
    return DagBag(dag_folder="dags/", include_examples=False)

def test_dag_loads(dagbag):
    assert "casino_medallion_pipeline" in dagbag.dags
    assert dagbag.import_errors == {}

def test_dag_structure(dagbag):
    dag = dagbag.get_dag("casino_medallion_pipeline")
    assert len(dag.tasks) >= 9  # 3 bronze + 1 quality + 3 silver + 2 gold

    # Verify bronze tasks have no upstream dependencies
    bronze_tasks = [t for t in dag.tasks if t.task_id.startswith("bronze_")]
    for task in bronze_tasks:
        assert len(task.upstream_list) == 0

def test_no_cycles(dagbag):
    dag = dagbag.get_dag("casino_medallion_pipeline")
    # Airflow will raise if there's a cycle during parsing
    assert dag is not None

@task
def check_ctr_backlog(ds=None):
    """Verify all CTR-eligible transactions have been reported."""
    from fabric.sqlendpoint import SqlClient

    sql = SqlClient(lakehouse="lh_gold")
    result = sql.query(f"""
        SELECT COUNT(*) as unreported
        FROM gold_compliance_monitoring
        WHERE transaction_date = '{ds}'
          AND ctr_required = true
          AND ctr_filed = false
    """)

    unreported = result[0]["unreported"]
    if unreported > 0:
        raise ValueError(f"{unreported} unfiled CTRs for {ds}")
    return {"status": "compliant", "date": ds}

@task
def cross_agency_quality_report(agency_results: list):
    """Aggregate quality metrics across all federal agencies."""
    report = {
        "report_date": datetime.utcnow().isoformat(),
        "agencies": {},
        "overall_pass_rate": 0
    }

    passed = 0
    for result in agency_results:
        agency = result["agency"]
        report["agencies"][agency] = {
            "status": result["status"],
            "row_count": result.get("row_count", 0),
            "quality_score": result.get("quality_score", 0)
        }
        if result["status"] == "passed":
            passed += 1

    report["overall_pass_rate"] = passed / len(agency_results)
    return report