⚡ User Data Functions - Serverless Compute in Fabric¶

# function_app.py
import json
import logging
from datetime import datetime

import fabric.functions as fn

app = fn.FabricFunctionsApp()

@app.route("health", methods=["GET"])
def health_check(req: fn.HttpRequest) -> fn.HttpResponse:
    """Simple health check endpoint."""
    return fn.HttpResponse(
        json.dumps({"status": "healthy", "timestamp": datetime.utcnow().isoformat()}),
        status_code=200,
        mimetype="application/json"
    )

@app.route("validate-transaction", methods=["POST"])
def validate_transaction(req: fn.HttpRequest) -> fn.HttpResponse:
    """Validate a casino transaction against compliance thresholds."""
    try:
        body = req.get_json()
        amount = body.get("amount", 0)
        transaction_type = body.get("type", "unknown")

        result = {
            "transaction_id": body.get("id"),
            "amount": amount,
            "ctr_flag": amount >= 10000,
            "sar_flag": 8000 <= amount < 10000,
            "w2g_flag": False,
            "validated_at": datetime.utcnow().isoformat()
        }

        # W-2G threshold depends on game type
        if transaction_type == "slot_jackpot" and amount >= 1200:
            result["w2g_flag"] = True
        elif transaction_type == "keno" and amount >= 600:
            result["w2g_flag"] = True
        elif transaction_type == "poker" and amount >= 5000:
            result["w2g_flag"] = True

        return fn.HttpResponse(
            json.dumps(result),
            status_code=200,
            mimetype="application/json"
        )
    except Exception as e:
        logging.error(f"Validation error: {e}")
        return fn.HttpResponse(
            json.dumps({"error": str(e)}),
            status_code=400,
            mimetype="application/json"
        )

@app.schedule("daily-quality-check", cron="0 6 * * *")
def daily_quality_check(timer: fn.TimerRequest) -> None:
    """Run daily data quality checks on bronze tables."""
    from fabric.onelake import OneLakeClient

    client = OneLakeClient()
    # Check row counts, null rates, freshness
    tables = ["slot_telemetry", "table_game_results", "player_tracking"]

    for table in tables:
        path = f"lh_bronze.Lakehouse/Tables/{table}"
        stats = client.get_table_stats(path)

        if stats["last_modified_hours_ago"] > 24:
            logging.warning(f"STALE DATA: {table} not updated in {stats['last_modified_hours_ago']}h")
        if stats["null_rate"] > 0.05:
            logging.warning(f"HIGH NULL RATE: {table} at {stats['null_rate']:.1%}")

        logging.info(f"Quality check passed: {table} ({stats['row_count']} rows)")

import requests

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

# Create a User Data Function
payload = {
    "displayName": "casino-compliance-udf",
    "description": "Transaction compliance validation endpoints",
    "definition": {
        "parts": [
            {
                "path": "function_app.py",
                "payloadType": "InlineBase64",
                "payload": "<base64-encoded-python-code>"
            },
            {
                "path": "requirements.txt",
                "payloadType": "InlineBase64",
                "payload": "<base64-encoded-requirements>"
            }
        ]
    }
}

response = requests.post(
    f"{base_url}/workspaces/{workspace_id}/userDataFunctions",
    headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
    json=payload
)

# Read from OneLake
@app.route("get-player/{player_id}", methods=["GET"])
@app.input_binding(type="onelake", path="lh_silver.Lakehouse/Tables/player_profiles")
def get_player(req: fn.HttpRequest, onelake_data) -> fn.HttpResponse:
    player_id = req.route_params.get("player_id")
    # onelake_data is a DataFrame-like object filtered by binding config
    player = onelake_data.filter(f"player_id = '{player_id}'").first()
    return fn.HttpResponse(json.dumps(player), mimetype="application/json")

# Write to Eventhouse
@app.eventstream_trigger("process-slot-event", stream="slot-telemetry-stream")
@app.output_binding(type="eventhouse", database="casino_rt", table="enriched_events")
def enrich_slot_event(event: fn.EventstreamEvent, eventhouse_out) -> None:
    enriched = {
        "original_event": event.body,
        "enriched_at": datetime.utcnow().isoformat(),
        "machine_zone": lookup_zone(event.body["machine_id"]),
        "player_tier": lookup_tier(event.body.get("player_id"))
    }
    eventhouse_out.write(enriched)

from fabric.onelake import OneLakeClient

client = OneLakeClient()

# Read a Delta table
df = client.read_table("lh_bronze.Lakehouse/Tables/slot_telemetry")

# Write a file
client.write_file(
    "lh_bronze.Lakehouse/Files/webhooks/incoming.json",
    content=json.dumps(payload)
)

from fabric.eventhouse import EventhouseClient

kql_client = EventhouseClient(database="casino_rt")

# Execute a KQL query
results = kql_client.query("""
    slot_events
    | where ingestion_time() > ago(5m)
    | summarize AvgBet = avg(bet_amount) by machine_id
    | top 10 by AvgBet desc
""")

from fabric.sqlendpoint import SqlClient

sql = SqlClient(lakehouse="lh_gold")
result = sql.query("SELECT TOP 10 * FROM gold_slot_performance ORDER BY total_revenue DESC")

@app.route("admin/purge", methods=["DELETE"], auth_level="admin")
def purge_stale_data(req: fn.HttpRequest) -> fn.HttpResponse:
    """Only Fabric admins can call this endpoint."""
    # auth_level options: anonymous, function, admin
    # "admin" requires Fabric Admin role
    pass

@app.eventstream_trigger("compliance-check", stream="financial-transactions")
@app.output_binding(type="eventhouse", database="casino_rt", table="compliance_flags")
def compliance_check(event: fn.EventstreamEvent, compliance_out) -> None:
    """Real-time compliance flagging for casino transactions."""
    txn = event.body

    flags = {
        "transaction_id": txn["id"],
        "timestamp": datetime.utcnow().isoformat(),
        "amount": txn["amount"],
        "ctr_flag": txn["amount"] >= 10000,
        "sar_structuring_flag": False,
        "w2g_flag": False
    }

    # SAR structuring detection: multiple transactions just under CTR
    # (In production, this would query recent history from Eventhouse)
    if 8000 <= txn["amount"] < 10000:
        flags["sar_structuring_flag"] = True

    # W-2G by game type
    thresholds = {"slot": 1200, "keno": 600, "table": 600, "poker": 5000}
    game_type = txn.get("game_type", "unknown")
    if txn["amount"] >= thresholds.get(game_type, 99999):
        flags["w2g_flag"] = True

    compliance_out.write(flags)

@app.route("webhook/usda-crop-report", methods=["POST"], auth_level="function")
@app.output_binding(type="onelake", path="lh_bronze.Lakehouse/Files/webhooks/usda/")
def receive_usda_report(req: fn.HttpRequest, onelake_out) -> fn.HttpResponse:
    """Receive USDA crop report webhook and store in OneLake."""
    payload = req.get_json()

    filename = f"crop_report_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json"
    onelake_out.write(filename, json.dumps(payload))

    logging.info(f"Received USDA crop report: {len(payload.get('data', []))} records")

    return fn.HttpResponse(
        json.dumps({"status": "received", "filename": filename}),
        status_code=202,
        mimetype="application/json"
    )