"""Generate authored data-product contracts for the transform-layer models. Per ADR 0009/0015 the transform layer owns and generates its staging and curated contracts. This is a minimal seed of that generator: it introspects each built relation in Databend (DESCRIBE) for the column schema and emits a contract YAML beside the model under contracts/generated/, validated against schemas/data-product-contract.authored.v0.schema.json. Raw-snapshot contracts stay the load stage's; here we reference them as upstreams. Usage (after dbt run): uv run --env-file .env -- python transform/dbt/engine/generate_contracts.py """ from __future__ import annotations import os from pathlib import Path import yaml from databend_driver import BlockingDatabendClient from jsonschema import Draft202012Validator REPO = Path(__file__).resolve().parents[3] DBT = REPO / "transform" / "dbt" SCHEMA = json_schema = None # set in main STG, DWH = "transform_dbt_stg", "transform_dbt_dwh" # model -> spec. dir is relative to transform/dbt/. upstreams are (lineage_kind, ref). # A staging model's immediate upstream is the load stage's raw-snapshot *product* contract it # reads FROM (kind data_product), not the extract stage's source contract — the transform layer # references but does not author those raw-snapshot contracts (ADR 0019/0016). googlesheets reads a # dbt seed and stg_dates is generated in Databend, so neither has a raw-snapshot product upstream. MODELS = { "stg_salesforce_org": dict(id="crm.nexuscrm.stg_salesforce_org", layer="staging", db=STG, mat="view", dir="sources/nexuscrm/contracts/generated", domain="crm", cls=["customer_data"], up=[("data_product", "crm.nexuscrm.org__c_snapshot")], src=dict(kind="database", system="nexuscrm", database="nexuscrm_raw", table="nexuscrm_org__c_snapshots")), "stg_salesforce_account": dict(id="crm.nexuscrm.stg_salesforce_account", layer="staging", db=STG, mat="view", dir="sources/nexuscrm/contracts/generated", domain="crm", cls=["customer_data"], up=[("data_product", "crm.nexuscrm.account_snapshot")], src=dict(kind="database", system="nexuscrm", database="nexuscrm_raw", table="nexuscrm_account_snapshots")), "stg_metadb_billing2_invoices": dict(id="billing.metadb.stg_metadb_billing2_invoices", layer="staging", db=STG, mat="view", dir="sources/metadb/contracts/generated", domain="billing", cls=["financial", "customer_data"], up=[("data_product", "billing.metadb.billing2_invoices_snapshot")], src=dict(kind="file", system="bigquery", database="metadb_raw", table="metadb_billing2_invoices_snapshots")), "stg_cloud_tenants": dict(id="cloud_platform.bq_warehouse.stg_cloud_tenants", layer="staging", db=STG, mat="view", dir="sources/bq_warehouse/contracts/generated", domain="cloud_platform", cls=["customer_data"], up=[("data_product", "cloud_platform.cloud_tenants.cloud_tenants_snapshot")], src=dict(kind="file", system="bigquery", database="bq_warehouse_raw", table="cloud_tenants_snapshots")), "stg_data_foundation_finance_salesforce_account": dict( id="finance.data_foundation_finance.stg_salesforce_account", layer="staging", db=STG, mat="view", dir="sources/data_foundation_finance/contracts/generated", domain="finance", cls=["financial", "customer_data"], up=[("data_product", "finance.data_foundation_finance.salesforce_account_snapshot")], src=dict(kind="database", system="data_foundation_finance", database="data_foundation_finance_raw", table="data_foundation_finance_salesforce_account_snapshots")), "stg_google_sheets_org_monthly_bill_confirmation_for_revrec": dict( id="finance.googlesheets.stg_google_sheets_org_monthly_bill_confirmation_for_revrec", layer="staging", db=STG, mat="view", dir="sources/googlesheets/contracts/generated", domain="finance", cls=["financial", "customer_data"], up=[("source", "seed.googlesheets.org_monthly_bill_confirmation_for_revrec")], src=dict(kind="application", system="google_sheets", seed="org_monthly_bill_confirmation_for_revrec")), "salesforce_orgs": dict(id="crm.salesforce_orgs", layer="curated", db=DWH, mat="view", dir="products/salesforce_orgs/contracts/generated", domain="crm", cls=["customer_data"], up=[("data_product", "crm.nexuscrm.stg_salesforce_org")]), "salesforce_accounts": dict(id="crm.salesforce_accounts", layer="curated", db=DWH, mat="view", dir="products/salesforce_accounts/contracts/generated", domain="crm", cls=["financial", "customer_data"], up=[("data_product", "crm.nexuscrm.stg_salesforce_account"), ("data_product", "finance.data_foundation_finance.stg_salesforce_account")]), "customer_surrogate_keys": dict(id="core.customer_surrogate_keys", layer="curated", db=DWH, mat="table", dir="products/customer_surrogate_keys/contracts/generated", domain="core", cls=["customer_data"], up=[("data_product", "crm.salesforce_accounts"), ("data_product", "crm.salesforce_orgs"), ("data_product", "cloud_platform.bq_warehouse.stg_cloud_tenants")]), "customers": dict(id="core.customers", layer="curated", db=DWH, mat="view", dir="products/customers/contracts/generated", domain="core", cls=["customer_data"], up=[("data_product", "core.customer_surrogate_keys"), ("data_product", "crm.salesforce_accounts"), ("data_product", "cloud_platform.bq_warehouse.stg_cloud_tenants"), ("data_product", "finance.googlesheets.stg_google_sheets_org_monthly_bill_confirmation_for_revrec")]), "salesforce_org_monthly_revrec": dict(id="finance.salesforce_org_monthly_revrec", layer="curated", db=DWH, mat="view", dir="products/salesforce_org_monthly_revrec/contracts/generated", domain="finance", cls=["financial", "customer_data"], up=[("data_product", "finance.googlesheets.stg_google_sheets_org_monthly_bill_confirmation_for_revrec"), ("data_product", "crm.salesforce_orgs"), ("data_product", "core.customer_surrogate_keys"), ("data_product", "core.customers"), ("data_product", "billing.metadb.stg_metadb_billing2_invoices")]), # --- salesforce_order_daily_view chain --- "stg_dates": dict(id="core.stg_dates", layer="staging", db=STG, mat="table", dir="sources/core/contracts/generated", domain="core", cls=["internal"], up=[], src=dict(kind="other", system="databend_generated")), "stg_salesforce_order": dict(id="crm.nexuscrm.stg_salesforce_order", layer="staging", db=STG, mat="view", dir="sources/nexuscrm/contracts/generated", domain="crm", cls=["customer_data"], up=[("data_product", "crm.nexuscrm.order_snapshot")], src=dict(kind="database", system="nexuscrm", database="nexuscrm_raw", table="nexuscrm_order_snapshots")), "stg_salesforce_opportunity": dict(id="crm.nexuscrm.stg_salesforce_opportunity", layer="staging", db=STG, mat="view", dir="sources/nexuscrm/contracts/generated", domain="crm", cls=["customer_data"], up=[("data_product", "crm.nexuscrm.opportunity_snapshot")], src=dict(kind="database", system="nexuscrm", database="nexuscrm_raw", table="nexuscrm_opportunity_snapshots")), "stg_salesforce_order_item": dict(id="crm.nexuscrm.stg_salesforce_order_item", layer="staging", db=STG, mat="view", dir="sources/nexuscrm/contracts/generated", domain="crm", cls=["financial", "customer_data"], up=[("data_product", "crm.nexuscrm.order_product__c_snapshot")], src=dict(kind="database", system="nexuscrm", database="nexuscrm_raw", table="nexuscrm_order_product__c_snapshots")), "stg_salesforce_product": dict(id="crm.nexuscrm.stg_salesforce_product", layer="staging", db=STG, mat="view", dir="sources/nexuscrm/contracts/generated", domain="crm", cls=["internal"], up=[("data_product", "crm.nexuscrm.product2_snapshot")], src=dict(kind="database", system="nexuscrm", database="nexuscrm_raw", table="nexuscrm_product2_snapshots")), "stg_salesforce_revenue": dict(id="finance.salesforce.stg_salesforce_revenue", layer="staging", db=STG, mat="view", dir="sources/salesforce/contracts/generated", domain="finance", cls=["financial", "customer_data"], up=[("data_product", "finance.salesforce.revenue_c_snapshot")], src=dict(kind="database", system="salesforce", database="salesforce_raw", table="salesforce_revenue_c_snapshots")), "stg_cloud_revenue": dict(id="finance.data_foundation_finance.stg_cloud_revenue", layer="staging", db=STG, mat="view", dir="sources/data_foundation_finance/contracts/generated", domain="finance", cls=["financial", "customer_data"], up=[("data_product", "finance.data_foundation_finance.monthly_cloud_revrec_snapshot")], src=dict(kind="database", system="data_foundation_finance", database="data_foundation_finance_raw", table="data_foundation_finance_monthly_cloud_revrec_snapshots")), "date_spine": dict(id="core.date_spine", layer="curated", db=DWH, mat="table", dir="products/date_spine/contracts/generated", domain="core", cls=["internal"], up=[("data_product", "core.stg_dates")]), "salesforce_opportunities": dict(id="crm.salesforce_opportunities", layer="curated", db=DWH, mat="view", dir="products/salesforce_opportunities/contracts/generated", domain="crm", cls=["customer_data"], up=[("data_product", "crm.nexuscrm.stg_salesforce_opportunity")]), "cloud_revenue": dict(id="finance.cloud_revenue", layer="curated", db=DWH, mat="view", dir="products/cloud_revenue/contracts/generated", domain="finance", cls=["financial", "customer_data"], up=[("data_product", "finance.data_foundation_finance.stg_cloud_revenue")]), "salesforce_revrec": dict(id="finance.salesforce_revrec", layer="curated", db=DWH, mat="view", dir="products/salesforce_revrec/contracts/generated", domain="finance", cls=["financial", "customer_data"], up=[("data_product", "finance.salesforce.stg_salesforce_revenue")]), "salesforce_orders": dict(id="crm.salesforce_orders", layer="curated", db=DWH, mat="view", dir="products/salesforce_orders/contracts/generated", domain="crm", cls=["financial", "customer_data"], up=[("data_product", "crm.nexuscrm.stg_salesforce_order"), ("data_product", "crm.salesforce_opportunities"), ("data_product", "crm.nexuscrm.stg_salesforce_order_item")]), "salesforce_order_daily_view": dict(id="finance.salesforce_order_daily_view", layer="curated", db=DWH, mat="view", dir="products/salesforce_order_daily_view/contracts/generated", domain="finance", cls=["financial", "customer_data"], up=[("data_product", "core.date_spine"), ("data_product", "crm.salesforce_orders"), ("data_product", "crm.nexuscrm.stg_salesforce_order_item"), ("data_product", "crm.nexuscrm.stg_salesforce_product"), ("data_product", "finance.salesforce_revrec")]), } def _merge_dict(base: dict, override: dict) -> dict: merged = dict(base) for key, value in override.items(): if isinstance(value, dict) and isinstance(merged.get(key), dict): merged[key] = _merge_dict(merged[key], value) else: merged[key] = value return merged def contract_fragment(name: str) -> dict: matches = [ path for path in DBT.rglob(f"{name}.contract-fragment.yaml") if "target" not in path.parts and "dbt_packages" not in path.parts ] if not matches: return {} if len(matches) > 1: joined = ", ".join(str(path.relative_to(DBT)) for path in matches) raise RuntimeError(f"multiple contract fragments for {name}: {joined}") fragment = yaml.safe_load(matches[0].read_text()) or {} if not isinstance(fragment, dict): raise RuntimeError(f"contract fragment must be a mapping: {matches[0].relative_to(DBT)}") return fragment def _dsn() -> str: d = os.environ["DATABEND_DSN"] if d.startswith("lake://"): d = "databend://" + d[len("lake://"):] if "warehouse=" not in d: d += ("&" if "?" in d else "?") + f"warehouse={os.environ.get('DATABEND_WAREHOUSE', 'default')}" return d def columns(cur, db: str, rel: str) -> list[dict]: cur.execute(f"DESCRIBE {db}.{rel}") out = [] for r in cur.fetchall(): rr = tuple(r.values()) if hasattr(r, "values") else tuple(r) nullable = str(rr[2]).upper() in ("YES", "TRUE", "1", "NULLABLE") out.append({"name": rr[0], "type": str(rr[1]), "nullable": nullable}) return out def contract(name: str, spec: dict, cols: list[dict]) -> dict: upstreams = [{"kind": k, "ref": r} for k, r in spec["up"]] upstreams.extend((spec.get("declared_lineage") or {}).get("upstreams", [])) doc = { "schema_version": "0.1.0", "kind": "product", "id": spec["id"], "name": name, "version": spec.get("version", "0.1.0"), "layer": spec["layer"], "owner": {"type": "team", "name": "data-platform"}, "domain": {"name": spec["domain"]}, "purpose": spec.get("purpose", f"{spec['layer'].capitalize()}-layer product {name}, produced by the transform layer (dbt on Databend)."), "usage": spec.get("usage", { "intended": ["Downstream modeling", "Analytics and reporting", "Agent-mediated querying"], "prohibited": ["Use as a source-faithful raw record (see the raw-snapshot contract)."], }), "assets": [{ "id": name, "kind": spec["mat"] if spec["mat"] in ("table", "view") else "model", "ref": f"{spec['db']}.{name}", "materialization": spec["mat"], "produced_by": {"kind": "dbt", "ref": name}, }], "schema": {"kind": "relational", "columns": cols}, "freshness": spec.get("freshness", {"max_age": "P1D"}), "tests": spec.get("tests", [{ "id": f"{name}_smoke", "type": "sql", "severity": "informational", "description": f"Placeholder smoke check for {name}; replace with row-count / uniqueness / freshness assertions.", "sql": f"select count(*) from {spec['db']}.{name}", }]), "declared_lineage": {"upstreams": upstreams}, "governance": { "classifications": spec["cls"], "access": {"default": "restricted"}, "audit": {"required": True, "events": ["export", "agent_action"]}, }, "agent_permissions": {"default": "deny"}, "compatibility": {"change_policy": "semver"}, } # Layer-specific requirements (schema allOf): staging requires `source`; curated requires `semantics`. if spec["layer"] == "staging": s = spec["src"] ident = {k: v for k, v in s.items() if k not in ("kind", "system")} doc["source"] = {"kind": s["kind"], "system": s["system"], "identifiers": ident} if "semantics" in spec: doc["semantics"] = spec["semantics"] elif spec["layer"] == "curated": doc["semantics"] = spec.get("semantics", {}) return doc def main() -> None: import json schema = json.loads((REPO / "schemas" / "data-product-contract.authored.v0.schema.json").read_text()) validator = Draft202012Validator(schema) cur = BlockingDatabendClient(_dsn()).cursor() for name, base_spec in MODELS.items(): spec = _merge_dict(base_spec, contract_fragment(name)) cols = columns(cur, spec["db"], name) doc = contract(name, spec, cols) errors = sorted(validator.iter_errors(doc), key=lambda e: e.path) if errors: print(f" INVALID {name}: {errors[0].message}") raise SystemExit(1) out_dir = DBT / spec["dir"] out_dir.mkdir(parents=True, exist_ok=True) out = out_dir / f"{name}.{spec['layer']}.contract.yaml" out.write_text(yaml.safe_dump(doc, sort_keys=False, width=100)) print(f" OK {spec['layer']:8s} {len(cols):3d} cols {out.relative_to(REPO)}") print("all transform contracts generated and valid") if __name__ == "__main__": main()