import os import json import yaml from datetime import datetime from fastapi import APIRouter, status from app.core.config import settings from app.schemas.sitomat import ( BundleItem, BundleListResponse, SitomatProvisionRequest, SitomatProvisionResponse, SitomatTriggerRequest, SitomatTriggerResponse, SitomatLedgerCallback, ) router = APIRouter(prefix="/sitomat", tags=["sitomat Ingestion Governance"]) @router.get("/health") def sitomat_health(): """Health status of sitomat control plane integration bridge.""" return { "status": "online", "service": "MERICA sitomat Governance Bridge", "supported_bundles": ["dlh-core-ingestion", "bpkp-perizinan"] } @router.get("/bundles", response_model=BundleListResponse) def list_dag_bundles(): """ List configured Airflow 3 GitDagBundles available for pipeline binding. Matches AIRFLOW__DAG_PROCESSOR__DAG_BUNDLE_CONFIG_LIST specifications. """ return { "bundles": [ { "name": "dlh-core-ingestion", "tracking_ref": "development", "repo_url": "https://repo.sicabe.com/sekota/merica-beta.git", "subdir": "airflow/configs", "type": "git" }, { "name": "bpkp-perizinan", "tracking_ref": "development", "repo_url": "https://repo.sicabe.com/sekota/merica-beta.git", "subdir": "airflow/configs", "type": "git" } ] } @router.post("/pipelines/provision", response_model=SitomatProvisionResponse, status_code=status.HTTP_201_CREATED) def provision_sitomat_pipeline(req: SitomatProvisionRequest): """ Control Plane Handshake (Gate-1): Register metadata, compile initial YAML config for Airflow Git DAG Bundle, and create initial semantic version snapshot (v1.0.0). """ clean_id = req.dataset_id.lower().replace(" ", "_") bronze_dag_id = f"ingestion_{clean_id}" transform_dag_id = f"transform_{clean_id}" bundle_name = req.bundle_name or "dlh-core-ingestion" initial_version = "v1.0.0" now_iso = datetime.utcnow().isoformat() + "Z" # 1. Canonical YAML specification for Airflow DAG builder yaml_config = { "pipeline_id": bronze_dag_id, "dataset_id": req.dataset_id, "bundle_name": bundle_name, "version": initial_version, "schedule": req.schedule or "0 2 * * *", "mode": "medallion", "bronze": { "source_type": req.source_type, "connection_id": req.connection_id, "source_config": req.source_config, "target": req.bronze_target or { "s3_bucket": "dlh-landing", "s3_prefix": f"bronze/raw/{clean_id}/" } }, "silver": { "transformation": { "select_models": req.dbt_models.get("silver", f"stg_{clean_id}") if req.dbt_models else f"stg_{clean_id}" } }, "gold": { "transformation": { "select_models": req.dbt_models.get("gold", f"fct_{clean_id}") if req.dbt_models else f"fct_{clean_id}" } } } os.makedirs(settings.CONFIGS_DIR, exist_ok=True) target_meta_dir = getattr(settings, "METADATA_DIR", settings.CONFIGS_DIR) os.makedirs(target_meta_dir, exist_ok=True) yaml_filename = f"{bronze_dag_id}.yaml" yaml_path = os.path.join(settings.CONFIGS_DIR, yaml_filename) yaml_header = ( f"# ============================================================\n" f"# MERICA Studio Data Lakehouse Pipeline\n" f"# Airflow 3 DAG Bundle: {bundle_name} | Dataset: {req.dataset_id}\n" f"# Initial Version: {initial_version} | Provisioned: {now_iso}\n" f"# ============================================================\n" ) with open(yaml_path, "w", encoding="utf-8") as f: f.write(yaml_header + yaml.dump(yaml_config, sort_keys=False)) # 2. JSON metadata for UI Canvas load json_path = os.path.join(target_meta_dir, f"{bronze_dag_id}.json") canvas_metadata = { "id": bronze_dag_id, "dataset_id": req.dataset_id, "bundle_name": bundle_name, "version": initial_version, "changelog": "Initial provisioning via sitomat Control Plane", "published_at": now_iso, "is_draft": False, "status": "deployed", "nodes": [], "edges": [] } with open(json_path, "w", encoding="utf-8") as f: json.dump(canvas_metadata, f, indent=2) # 3. Create initial version snapshot file for Time-Machine snapshot_filename = f"{bronze_dag_id}.version.{initial_version}.json" snapshot_path = os.path.join(target_meta_dir, snapshot_filename) with open(snapshot_path, "w", encoding="utf-8") as f: json.dump(canvas_metadata, f, indent=2) return SitomatProvisionResponse( status="provisioned", dataset_id=req.dataset_id, bundle_name=bundle_name, bronze_dag_id=bronze_dag_id, transform_dag_id=transform_dag_id, initial_version=initial_version, git_commit_hash="local-dev", yaml_config_path=yaml_path ) @router.post("/runs/trigger", response_model=SitomatTriggerResponse) def trigger_sitomat_run(req: SitomatTriggerRequest): """ Data Plane Trigger: sitomat triggers an ingestion execution with correlated ledger run ID and pinned version. """ clean_id = req.dataset_id.lower().replace(" ", "_") dag_id = f"ingestion_{clean_id}" dag_run_id = f"sitomat_run__{req.correlation_id}" pinned_version = req.target_dag_version or "latest" return SitomatTriggerResponse( status="triggered", correlation_id=req.correlation_id, dag_id=dag_id, dag_run_id=dag_run_id, pinned_version=pinned_version ) @router.post("/runs/callback") def sitomat_run_callback(payload: SitomatLedgerCallback): """ Ingestion Run Ledger Callback: Airflow task notifies sitomat Control Plane when Bronze extraction completes. """ return { "status": "received", "correlation_id": payload.correlation_id, "dataset_id": payload.dataset_id, "message": f"Ledger updated for run '{payload.dag_run_id}' with status '{payload.status}'." }