🔧 Variable Libraries - Parameterized Pipelines & Environments¶

import requests

base_url = "https://api.fabric.microsoft.com/v1"
workspace_id = "your-workspace-id"
token = "your-bearer-token"

headers = {
    "Authorization": f"Bearer {token}",
    "Content-Type": "application/json"
}

# Create a Variable Library
payload = {
    "displayName": "casino-config-dev",
    "description": "Casino POC configuration for development environment",
    "definition": {
        "parts": [
            {
                "path": "variableLibrary.json",
                "payload": {
                    "variables": {
                        "storage_account": {
                            "type": "String",
                            "value": "fabricpocdev"
                        },
                        "bronze_path": {
                            "type": "String",
                            "value": "abfss://bronze@onelake.dfs.fabric.microsoft.com/lh_bronze.Lakehouse/Tables"
                        },
                        "retention_days": {
                            "type": "Int",
                            "value": 90
                        },
                        "enable_pii_masking": {
                            "type": "Bool",
                            "value": True
                        },
                        "ctr_threshold": {
                            "type": "Int",
                            "value": 10000
                        },
                        "db_connection_string": {
                            "type": "Secret",
                            "keyVaultUrl": "https://kv-fabric-poc-dev.vault.azure.net/",
                            "secretName": "sql-connection-string"
                        }
                    }
                }
            }
        ]
    }
}

response = requests.post(
    f"{base_url}/workspaces/{workspace_id}/variableLibraries",
    headers=headers,
    json=payload
)
print(response.status_code, response.json())

# Update a specific variable value
library_id = "your-library-id"

update_payload = {
    "variables": {
        "retention_days": {
            "type": "Int",
            "value": 365  # Production: longer retention
        }
    }
}

response = requests.patch(
    f"{base_url}/workspaces/{workspace_id}/variableLibraries/{library_id}",
    headers=headers,
    json=update_payload
)

"""
Promote Variable Library values across environments.
Used in GitHub Actions or Azure DevOps pipelines.
"""
import json
import sys

def promote_variables(source_env: str, target_env: str, overrides: dict):
    """Read source env config, apply target overrides, push to target workspace."""

    with open(f"config/variables-{source_env}.json") as f:
        variables = json.load(f)

    # Apply environment-specific overrides
    for key, value in overrides.items():
        if key in variables:
            variables[key]["value"] = value

    # Update target workspace Variable Library via REST API
    update_variable_library(
        workspace_id=WORKSPACE_IDS[target_env],
        library_id=LIBRARY_IDS[target_env],
        variables=variables
    )
    print(f"Promoted {len(variables)} variables from {source_env} to {target_env}")

# Usage: python promote.py staging prod
if __name__ == "__main__":
    OVERRIDES = {
        "prod": {
            "retention_days": 365,
            "log_level": "WARNING",
            "max_parallelism": 8
        }
    }
    promote_variables(sys.argv[1], sys.argv[2], OVERRIDES.get(sys.argv[2], {}))

# In a Fabric notebook, secret variables are resolved automatically
# when accessed through the Variable Library binding

# Option 1: Via mssparkutils (recommended)
db_conn = mssparkutils.credentials.getSecret(
    "https://kv-fabric-poc-prod.vault.azure.net/",
    "casino-db-connection"
)

# Option 2: Via pipeline parameter passthrough
# Pipeline passes secret variable as a notebook parameter
db_conn = dbutils.widgets.get("db_connection_string")  # Already resolved

# Read variable library values passed as notebook parameters
env = spark.conf.get("spark.fabric.variable.env", "dev")
bronze_path = spark.conf.get("spark.fabric.variable.bronze_path")
retention_days = int(spark.conf.get("spark.fabric.variable.retention_days", "90"))
ctr_threshold = int(spark.conf.get("spark.fabric.variable.ctr_threshold", "10000"))

print(f"Environment: {env}")
print(f"Bronze path: {bronze_path}")
print(f"Retention: {retention_days} days")

# Use in transformations
from pyspark.sql import functions as F
from datetime import datetime, timedelta

cutoff_date = datetime.now() - timedelta(days=retention_days)

df = spark.read.format("delta").load(bronze_path)
df_filtered = df.filter(F.col("event_timestamp") >= cutoff_date)

# Compliance check using variable threshold
df_ctr = df_filtered.filter(F.col("transaction_amount") >= ctr_threshold)
print(f"CTR candidates: {df_ctr.count()}")

# main.py for Spark Job Definition
import argparse
from pyspark.sql import SparkSession

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--env", required=True)
    parser.add_argument("--bronze-path", required=True)
    parser.add_argument("--retention-days", type=int, default=90)
    args = parser.parse_args()

    spark = SparkSession.builder.appName(f"bronze-ingest-{args.env}").getOrCreate()

    df = spark.read.format("delta").load(args.bronze_path)
    print(f"Loaded {df.count()} records from {args.bronze_path}")

    spark.stop()

if __name__ == "__main__":
    main()