Three Layers of Metadata Every Bank Must Manage — and How OpenMetadata Handles All of Them

docker compose --project-directory C:\reg_repo\platform-solution\corep-governance-pipeline ps | grep openmetadata
# Expected: openmetadata   running   0.0.0.0:8585->8585/tcp

# Default credentials (change immediately in production)
# URL: http://localhost:8585
# User: admin   Password: admin

# In your virtual environment
pip install "openmetadata-ingestion[postgres]==1.3.4"
pip install "openmetadata-ingestion[dbt]==1.3.4"
pip install "openmetadata-ingestion[airflow]==1.3.4"

# catalog/connectors/postgres_connector.yaml
source:
  type: postgres
  serviceName: corep-postgres
  serviceConnection:
    config:
      type: Postgres
      username: ${POSTGRES_USER}
      authType:
        password: ${POSTGRES_PASSWORD}
      hostPort: localhost:5432
      database: corep
  sourceConfig:
    config:
      type: DatabaseMetadata
      schemaFilterPattern:
        includes:
          - ^raw$
          - ^staging$
          - ^intermediate$
          - ^mart$
          - ^audit$
          - ^secure$
          - ^mapping$
        excludes:
          - ^pg_.*            # exclude internal Postgres schemas
          - ^information_schema$
      tableFilterPattern:
        includes:
          - .*                # discover all tables in included schemas
      includeViews: true
      includeTags: true
      markDeletedTables: true    # flag tables dropped from DB as Deleted in catalog
      includeDDL: false          # DDL may contain sensitive info — do not store

sink:
  type: metadata-rest
  config:
    api_endpoint: http://localhost:8585/api
    auth_provider: openmetadata
    security_config:
      jwt_token: ${OM_JWT_TOKEN}   # generated in OpenMetadata Settings → Bots

workflowConfig:
  openMetadataServerConfig:
    hostPort: http://localhost:8585/api
    authProvider: openmetadata
    securityConfig:
      jwtToken: ${OM_JWT_TOKEN}

# Run connector
metadata ingest -c catalog/connectors/postgres_connector.yaml

# Expected output
INFO  Ingesting metadata for service: corep-postgres
INFO  Processing schema: raw         — 6 tables found
INFO  Processing schema: staging     — 5 tables found
INFO  Processing schema: intermediate— 4 tables found
INFO  Processing schema: mart        — 4 tables found
INFO  Processing schema: audit       — 2 tables found
INFO  Writing 21 tables to OpenMetadata...
INFO  ✓ Ingestion completed successfully.

# Create the EBA glossary via OpenMetadata REST API
# catalog/scripts/create_eba_glossary.py

import requests, os

BASE = os.environ.get("OM_API", "http://localhost:8585/api")
HEADERS = {
    "Authorization": f"Bearer {os.environ['OM_JWT_TOKEN']}",
    "Content-Type": "application/json",
}

# ── 1. Create the top-level glossary ─────────────────────────────
glossary_payload = {
    "name": "eba-corep-glossary",
    "displayName": "EBA COREP Regulatory Glossary",
    "description": (
        "Canonical definitions for all data elements used in COREP regulatory "
        "reporting. Terms are aligned to EBA Data Point Model 4.0 concept IDs. "
        "Reference: https://www.eba.europa.eu/risk-analysis-and-data/reporting-frameworks"
    ),
    "owners": [],
    "reviewers": [],
    "tags": ["regulatory", "eba-dpm-4.0"],
    "mutuallyExclusive": False,
}
r = requests.post(f"{BASE}/v1/glossaries", json=glossary_payload, headers=HEADERS)
r.raise_for_status()
glossary_id = r.json()["id"]
print(f"Glossary created: {glossary_id}")

# ── 2. Create glossary terms ──────────────────────────────────────
EBA_TERMS = [
    {
        "name": "own-funds",
        "displayName": "Own Funds",
        "description": (
            "Total regulatory capital comprising CET1, AT1, and Tier 2 instruments. "
            "Defined in CRR Article 4(1)(118). EBA DPM concept: c0010. "
            "Reported in COREP template C 01.00 row 010."
        ),
        "synonyms": ["Total Capital", "Regulatory Capital", "Eligible Capital"],
        "relatedTerms": [],
        "references": [
            {"name": "CRR Article 4(1)(118)", "endpoint": "https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX%3A32013R0575"},
            {"name": "EBA DPM 4.0 — c0010", "endpoint": "https://www.eba.europa.eu/sites/default/documents/files/document_library/EBA_DPM_4.0.pdf"},
        ],
    },
    {
        "name": "cet1-capital",
        "displayName": "Common Equity Tier 1 Capital",
        "description": (
            "Highest quality regulatory capital: ordinary shares, retained earnings, "
            "other comprehensive income. CRR Article 26. Minimum ratio: 4.5% of RWA. "
            "EBA DPM concept: c0020. COREP C 01.00 row 020."
        ),
        "synonyms": ["CET1", "Core Tier 1"],
        "references": [
            {"name": "CRR Article 26", "endpoint": "https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX%3A32013R0575"},
        ],
    },
    {
        "name": "risk-weighted-assets",
        "displayName": "Risk-Weighted Assets",
        "description": (
            "Assets adjusted for credit, market, and operational risk. The denominator "
            "of all Basel III capital ratios. Calculated under CRR Part Three. "
            "EBA DPM concept: c0060. COREP C 02.00."
        ),
        "synonyms": ["RWA", "Risk Weighted Exposures"],
        "references": [
            {"name": "CRR Part Three Title II", "endpoint": "https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX%3A32013R0575"},
        ],
    },
    {
        "name": "hqla",
        "displayName": "High Quality Liquid Assets",
        "description": (
            "Assets that can be converted to cash with minimal loss in stressed market "
            "conditions. Classified as Level 1, 2A, or 2B under Delegated Regulation 2015/61. "
            "Numerator of the Liquidity Coverage Ratio. EBA DPM: c0010 in C 47.00."
        ),
        "synonyms": ["High-Quality Liquid Assets", "LCR Buffer", "Liquidity Buffer"],
        "references": [
            {"name": "Del. Reg. 2015/61 Article 7", "endpoint": "https://eur-lex.europa.eu/legal-content/EN/TXT/?uri=CELEX%3A32015R0061"},
        ],
    },
    {
        "name": "lcr",
        "displayName": "Liquidity Coverage Ratio",
        "description": (
            "Ratio of HQLA stock to net cash outflows over a 30-day stress period. "
            "Minimum requirement: 100%. Del. Reg. 2015/61 Article 4. "
            "EBA DPM concept: c0090 in COREP C 47.00."
        ),
        "synonyms": ["Liquidity Ratio", "30-Day Liquidity"],
    },
    {
        "name": "exposure-at-default",
        "displayName": "Exposure at Default",
        "description": (
            "The expected outstanding loan balance at the time of borrower default. "
            "Input to RWA calculation. CRR Article 166. EBA DPM: ead column in C 02.00."
        ),
        "synonyms": ["EAD", "Credit Exposure"],
    },
]

for term in EBA_TERMS:
    term["glossary"] = {"id": glossary_id, "type": "glossary"}
    r = requests.post(f"{BASE}/v1/glossaryTerms", json=term, headers=HEADERS)
    r.raise_for_status()
    print(f"  Term created: {term['displayName']}")

print("EBA glossary complete.")

# catalog/scripts/tag_columns.py — attach glossary terms to columns

import requests, os

BASE    = os.environ.get("OM_API", "http://localhost:8585/api")
HEADERS = {
    "Authorization": f"Bearer {os.environ['OM_JWT_TOKEN']}",
    "Content-Type": "application/json",
}

# Maps (schema.table, column) → glossary term name
COLUMN_TERM_MAP = [
    ("mart.corep_c0100", "own_funds",         "own-funds"),
    ("mart.corep_c0100", "cet1_capital",       "cet1-capital"),
    ("mart.corep_c0100", "total_rwa",           "risk-weighted-assets"),
    ("mart.corep_c0200", "ead",                 "exposure-at-default"),
    ("mart.corep_c0200", "rwa",                 "risk-weighted-assets"),
    ("mart.corep_c4700", "hqla_buffer",         "hqla"),
    ("mart.corep_c4700", "lcr_ratio",           "lcr"),
    ("raw.capital_instruments", "amount",       "own-funds"),
    ("raw.rwa_exposures", "ead",               "exposure-at-default"),
    ("raw.liquidity_assets", "market_value",   "hqla"),
]

def get_table_fqn(schema_table: str) -> str:
    """Build fully qualified table name for OpenMetadata API."""
    schema, table = schema_table.split(".")
    return f"corep-postgres.corep.{schema}.{table}"

def get_table_id(fqn: str) -> str:
    r = requests.get(f"{BASE}/v1/tables/name/{fqn}", headers=HEADERS)
    r.raise_for_status()
    return r.json()["id"]

def get_term_id(term_name: str) -> str:
    r = requests.get(
        f"{BASE}/v1/glossaryTerms/name/eba-corep-glossary.{term_name}",
        headers=HEADERS,
    )
    r.raise_for_status()
    return r.json()["id"]

for schema_table, column, term_name in COLUMN_TERM_MAP:
    fqn      = get_table_fqn(schema_table)
    table_id = get_table_id(fqn)
    term_id  = get_term_id(term_name)

    # PATCH the column with the glossary term tag
    patch = [{
        "op": "add",
        "path": f"/columns/{column}/tags/-",
        "value": {
            "tagFQN": f"eba-corep-glossary.{term_name}",
            "source": "Glossary",
            "labelType": "Automated",
            "state": "Confirmed",
        }
    }]
    r = requests.patch(
        f"{BASE}/v1/tables/{table_id}",
        json=patch,
        headers={**HEADERS, "Content-Type": "application/json-patch+json"},
    )
    r.raise_for_status()
    print(f"  Tagged: {schema_table}.{column} → {term_name}")

# catalog/scripts/apply_pii_tags.py — create classification + apply to raw PII columns

# Step 1: Create PII classification tag (one-time setup)
classification_payload = {
    "name": "GDPR",
    "displayName": "GDPR Data Classification",
    "description": "Tags for GDPR Article 4 personal data classification.",
    "mutuallyExclusive": True,
}
r = requests.post(f"{BASE}/v1/classifications", json=classification_payload, headers=HEADERS)
r.raise_for_status()

# Step 2: Create tags within the classification
PII_TAGS = [
    {"name": "PII",          "description": "Directly identifies a natural or legal person. Access restricted."},
    {"name": "Pseudonymised", "description": "Personal data that has been pseudonymised per GDPR Article 4(5)."},
    {"name": "Non-PII",       "description": "Confirmed non-personal data — safe to aggregate and share."},
]
for tag in PII_TAGS:
    tag["classification"] = "GDPR"
    r = requests.post(f"{BASE}/v1/tags", json=tag, headers=HEADERS)
    r.raise_for_status()
    print(f"  Tag created: GDPR.{tag['name']}")

# Step 3: Apply PII tag to raw columns that contain personal data
PII_COLUMNS = [
    ("raw.counterparties", "name"),   # legal entity name — PII
    ("raw.counterparties", "lei"),    # LEI — PII under GDPR
    ("raw.loans",          "borrower_name"),
    ("raw.loans",          "borrower_id"),
]

NON_PII_COLUMNS = [
    ("mart.corep_c0100", "own_funds"),
    ("mart.corep_c0100", "cet1_capital"),
    ("mart.corep_c0300", "cet1_ratio"),
    ("mart.corep_c4700", "lcr_ratio"),
]

def apply_tag(schema_table, column, tag_fqn):
    fqn      = get_table_fqn(schema_table)
    table_id = get_table_id(fqn)
    patch = [{
        "op": "add",
        "path": f"/columns/{column}/tags/-",
        "value": {"tagFQN": tag_fqn, "source": "Classification", "labelType": "Manual", "state": "Confirmed"},
    }]
    r = requests.patch(
        f"{BASE}/v1/tables/{table_id}", json=patch,
        headers={**HEADERS, "Content-Type": "application/json-patch+json"},
    )
    r.raise_for_status()
    print(f"  Applied {tag_fqn} → {schema_table}.{column}")

for schema_table, column in PII_COLUMNS:
    apply_tag(schema_table, column, "GDPR.PII")

for schema_table, column in NON_PII_COLUMNS:
    apply_tag(schema_table, column, "GDPR.Non-PII")

# catalog/connectors/dbt_connector.yaml
source:
  type: dbt
  serviceName: corep-postgres      # attach dbt metadata to the existing Postgres service
  sourceConfig:
    config:
      type: DBT
      dbtConfigSource:
        dbtCatalogFilePath:  dbt/target/catalog.json
        dbtManifestFilePath: dbt/target/manifest.json
        dbtRunResultsFilePath: dbt/target/run_results.json   # test pass/fail status
      dbtUpdateDescriptions: true   # push dbt model descriptions into OM table descriptions
      includeDbtTags: true          # push dbt tags (e.g. tag: corep, tag: pii_free)

sink:
  type: metadata-rest
  config:
    api_endpoint: http://localhost:8585/api
    auth_provider: openmetadata
    security_config:
      jwt_token: ${OM_JWT_TOKEN}

workflowConfig:
  openMetadataServerConfig:
    hostPort: http://localhost:8585/api
    authProvider: openmetadata
    securityConfig:
      jwtToken: ${OM_JWT_TOKEN}

# Generate dbt artefacts first, then run the connector
cd corep-governance-pipeline
dbt docs generate --project-dir dbt --profiles-dir dbt

# Push dbt metadata to OpenMetadata
metadata ingest -c catalog/connectors/dbt_connector.yaml

# catalog/connectors/openlineage_connector.yaml
source:
  type: OpenLineage
  serviceName: corep-pipeline-lineage
  serviceConnection:
    config:
      type: OpenLineage
      openLineageApiEndpoint: http://localhost:5000/api/v1   # Marquez API
      lineageNamespace: corep-governance-pipeline
  sourceConfig:
    config:
      type: PipelineMetadata
      includeLineage: true

sink:
  type: metadata-rest
  config:
    api_endpoint: http://localhost:8585/api
    auth_provider: openmetadata
    security_config:
      jwt_token: ${OM_JWT_TOKEN}

workflowConfig:
  openMetadataServerConfig:
    hostPort: http://localhost:8585/api
    authProvider: openmetadata
    securityConfig:
      jwtToken: ${OM_JWT_TOKEN}

# Add table profile + freshness config to postgres_connector.yaml
source:
  type: postgres
  serviceName: corep-postgres
  serviceConnection:
    config:
      type: Postgres
      # ... (same as before)
  sourceConfig:
    config:
      type: DatabaseMetadata
      # ... (same as before)

# Add a SEPARATE profiler workflow for freshness
---
source:
  type: postgres
  serviceName: corep-postgres
  sourceConfig:
    config:
      type: Profiler
      generateSampleData: true
      computeMetrics: true
      profileSample: 30.0                   # profile 30% of rows (fast)
      timeoutSeconds: 300
      tableConfig:
        - fullyQualifiedName: corep-postgres.corep.mart.corep_c0100
          profileSample: 100.0              # always full-scan mart tables (small)
        - fullyQualifiedName: corep-postgres.corep.mart.corep_c0300
          profileSample: 100.0

"""
modules/catalog.py — Push pipeline run metadata to OpenMetadata after each run.

Responsibilities:
  - Trigger PostgreSQL connector ingestion (refresh table stats)
  - Trigger dbt connector ingestion (push manifest + lineage)
  - Write pipeline run completion time to audit.pipeline_run_log
    so OpenMetadata freshness checks have a timestamp to read
"""

import logging, os, subprocess
from pathlib import Path

from modules.base import BaseModule

log = logging.getLogger(__name__)

CATALOG_DIR = Path(os.environ.get("CATALOG_DIR", "catalog"))


class CatalogModule(BaseModule):
    MODULE_NAME = "catalog"

    _CONNECTORS = [
        "connectors/postgres_connector.yaml",
        "connectors/dbt_connector.yaml",
        "connectors/openlineage_connector.yaml",
    ]

    def input_check(self) -> None:
        """Verify connector YAML files and OM_JWT_TOKEN are present."""
        if not os.environ.get("OM_JWT_TOKEN"):
            raise RuntimeError(
                "[catalog] OM_JWT_TOKEN not set. "
                "Generate a bot token in OpenMetadata Settings → Bots."
            )
        missing = [
            str(CATALOG_DIR / c)
            for c in self._CONNECTORS
            if not (CATALOG_DIR / c).exists()
        ]
        if missing:
            raise RuntimeError("[catalog] Missing connector files: " + ", ".join(missing))

    def _execute(self) -> None:
        for connector_path in self._CONNECTORS:
            full_path = CATALOG_DIR / connector_path
            log.info("[catalog] Running ingestion: %s", connector_path)
            result = subprocess.run(
                ["metadata", "ingest", "-c", str(full_path)],
                capture_output=True, text=True, timeout=300,
            )
            if result.returncode != 0:
                log.warning(
                    "[catalog] Ingestion failed for %s (non-fatal):\n%s",
                    connector_path, result.stderr[-2000:]
                )
            else:
                log.info("[catalog] Ingestion complete: %s", connector_path)

    def emit_lineage(self) -> None:
        log.info("[catalog] No lineage event — catalog is metadata, not data.")

    def output_check(self) -> None:
        """Verify OM API responds with tables present."""
        import requests
        api = os.environ.get("OM_API", "http://localhost:8585/api")
        headers = {"Authorization": f"Bearer {os.environ['OM_JWT_TOKEN']}"}
        r = requests.get(f"{api}/v1/tables?service=corep-postgres&limit=1", headers=headers, timeout=10)
        r.raise_for_status()
        count = r.json().get("paging", {}).get("total", 0)
        if count == 0:
            raise RuntimeError("[catalog] OpenMetadata reports 0 tables for corep-postgres service.")
        log.info("[catalog] OpenMetadata has %d tables indexed for corep-postgres.", count)