sicabe-core/deploy/scripts/airflow/run-om-native-dq.py

#!/usr/bin/env python
# =============================================================================
# eBdesk SiCABE — JALUR B: Data Quality NATIVE OpenMetadata (OM yang menjalankan)
# =============================================================================
# Bedanya dengan Jalur A (Great Expectations):
#   A: GE yang MENJALANKAN ekspektasi → hasilnya DIKIRIM ke OM
#      (test case ber-platform "GreatExpectations").
#   B: OPENMETADATA yang MENJALANKAN test (definisi bawaan OM:
#      tableRowCountToBeBetween, columnValuesToBeNotNull, columnValuesToBeBetween)
#      → hasilnya langsung tersimpan di OM (platform "OpenMetadata").
#
# Catatan implementasi (dua hal yang harus begini, sudah diverifikasi):
#   1. HARUS lewat `TestRunner.for_table`/`from_yaml` — `TestRunner(...)` mentah
#      tidak memuat entity tabel sehingga run() gagal "Table entity not provided".
#   2. Koneksi data HARUS dibawa eksplisit lewat workflow (serviceConnection +
#      verify=<sertifikat Trino>). Bila mengandalkan koneksi yang TERSIMPAN di OM,
#      eksekusi gagal 401 'Invalid credentials' pada OM 1.13.1. `metadata ingest`
#      CLI juga bukan jalur eksekusi TestSuite (registry source-nya terpisah).
#
# Jalankan DI DALAM container Airflow:
#   docker exec -e OPENMETADATA_ADMIN_PASSWORD=... \
#     sicabe-airflow-webserver python /opt/airflow/scripts/run-om-native-dq.py
#
# Exit code: 0 = semua test lulus; 1 = ada test gagal/aborted.
# =============================================================================
import base64
import json
import os
import sys
import urllib.request

from metadata.generated.schema.entity.services.connections.metadata.openMetadataConnection import (
    OpenMetadataConnection,
)
from metadata.generated.schema.security.client.openMetadataJWTClientConfig import (
    OpenMetadataJWTClientConfig,
)
from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.sdk.data_quality.runner import TestRunner
from metadata.sdk.data_quality.tests import (
    ColumnValuesToBeBetween,
    ColumnValuesToBeNotNull,
    TableRowCountToBeBetween,
)

OM_API_URL = os.getenv("OM_API_URL", "http://openmetadata-server:8585/api")
SERVICE = os.getenv("OM_DATABASE_SERVICE", "trino_sicabe")
CATALOG = os.getenv("OM_DATABASE_NAME", "iceberg")
TRINO_HOSTPORT = os.getenv("OM_TRINO_HOSTPORT", "trino-coordinator:8443")
TRINO_CERT = os.getenv("TRINO_CA_BUNDLE", "/etc/ssl/trino/trino.crt")

# Workflow TestSuite: membawa koneksi Trino eksplisit (username + password +
# verify=sertifikat) supaya eksekusi terautentikasi dan TLS tetap DIVERIFIKASI.
WORKFLOW_TEMPLATE = """
source:
  type: TestSuite
  serviceName: {service}
  sourceConfig:
    config:
      type: TestSuite
      entityFullyQualifiedName: {table}
      # WAJIB 'serviceConnections' (plural, list) — DQ source membaca field ini via
      # _load_yaml_service_connections(). Memakai 'serviceConnection' (tunggal)
      # diabaikan, sehingga eksekusi jatuh ke koneksi TERSIMPAN di OM yang
      # kehilangan secret → HTTP 401 'Invalid credentials' (lihat TODO upstream
      # open-metadata#21259). Dengan bentuk ini koneksi dari YAML yang dipakai.
      serviceConnections:
        - serviceName: {service}
          serviceConnection:
            config:
              type: Trino
              scheme: trino
              hostPort: {hostport}
              username: svc_om
              authType:
                password: "{password}"
              catalog: {catalog}
              verify: {cert}
sink:
  type: metadata-rest
  config: {{}}
workflowConfig:
  loggerLevel: INFO
  openMetadataServerConfig:
    hostPort: {om_api}
    authProvider: openmetadata
    securityConfig:
      jwtToken: "{token}"
"""

# Test NATIVE (dijalankan OM) — definisi bawaan OM, bukan ekspektasi GE.
TABLES = {
    f"{SERVICE}.{CATALOG}.bronze.orders": [
        TableRowCountToBeBetween(min_count=1, max_count=1000000),
        ColumnValuesToBeNotNull(column="id"),
        ColumnValuesToBeNotNull(column="komoditas"),
        ColumnValuesToBeBetween(column="harga_per_kg", min_value=0, max_value=1000000),
    ],
    f"{SERVICE}.{CATALOG}.silver.orders_silver": [
        TableRowCountToBeBetween(min_count=1, max_count=1000000),
        ColumnValuesToBeNotNull(column="id"),
        ColumnValuesToBeBetween(column="total_value", min_value=0, max_value=1000000000000),
    ],
    f"{SERVICE}.{CATALOG}.gold.orders_gold": [
        TableRowCountToBeBetween(min_count=1, max_count=1000000),
        ColumnValuesToBeNotNull(column="komoditas"),
        ColumnValuesToBeBetween(column="avg_harga_per_kg", min_value=0, max_value=1000000),
        ColumnValuesToBeBetween(column="total_nilai", min_value=0, max_value=1000000000000),
    ],
}


def _token():
    """Token OM: dari OM_JWT_TOKEN, atau login pakai password admin.

    Dua kandidat: OPENMETADATA_ADMIN_PASSWORD lalu 'admin' — default yang dipakai
    OM 2.0.x sampai bootstrap step 7 menyetel hash-nya dari .env. MAKSIMAL DUA
    percobaan dan tidak diulang dalam loop: akun admin OM bisa terkunci.

    Catatan: 401 pada kandidat pertama adalah hal NORMAL (mis. step 7 belum jalan),
    jadi urlopen-nya dibungkus try/except — tanpa itu HTTPError 401 menggagalkan
    seluruh skrip.
    """
    if os.getenv("OM_JWT_TOKEN"):
        return os.environ["OM_JWT_TOKEN"]
    email = os.getenv("OPENMETADATA_ADMIN_EMAIL", "admin@open-metadata.org")
    for password in (os.getenv("OPENMETADATA_ADMIN_PASSWORD"), "admin"):
        if not password:
            continue
        body = json.dumps({
            "email": email,
            "password": base64.b64encode(password.encode()).decode(),
        }).encode()
        req = urllib.request.Request(f"{OM_API_URL}/v1/users/login", data=body,
                                    headers={"Content-Type": "application/json"})
        try:
            tok = json.load(urllib.request.urlopen(req, timeout=20)).get("accessToken")
        except Exception:  # noqa: BLE001 — kandidat pertama gagal = hal biasa
            tok = None
        if tok:
            return tok
    return None


def main():
    token = _token()
    trino_password = os.getenv("TRINO_SVC_OM_PASSWORD")
    if not token:
        print("ERROR: token OpenMetadata tidak tersedia (OM_JWT_TOKEN / "
              "OPENMETADATA_ADMIN_PASSWORD)")
        return 1
    if not trino_password:
        print("ERROR: TRINO_SVC_OM_PASSWORD tidak diset — koneksi Trino tidak bisa dibuat")
        return 1

    client = OpenMetadata(
        OpenMetadataConnection(
            hostPort=OM_API_URL,
            authProvider="openmetadata",
            securityConfig=OpenMetadataJWTClientConfig(jwtToken=token),
        )
    )

    failed = 0
    total = 0
    for table_fqn, tests in TABLES.items():
        print(f"[DQ-native] tabel {table_fqn} — {len(tests)} test")
        workflow = WORKFLOW_TEMPLATE.format(
            service=SERVICE, hostport=TRINO_HOSTPORT, password=trino_password,
            catalog=CATALOG, cert=TRINO_CERT, table=table_fqn, om_api=OM_API_URL,
            token=token,
        )
        # use_connection_from_yaml=True → pakai koneksi dari workflow (bukan yang
        # tersimpan di OM), sehingga autentikasi Trino bekerja.
        runner = TestRunner.from_yaml(
            yaml_string=workflow, use_connection_from_yaml=True, client=client
        )
        for test in tests:
            runner.add_test(test)

        for result in runner.run():
            total += 1
            status = str(getattr(result.testCaseResult, "testCaseStatus", None))
            name = getattr(result, "name", None) or getattr(result, "fqn", "test")
            print(f"    - {name}: {status.split('.')[-1]}")
            if status.split(".")[-1] != "Success":
                failed += 1

    print(f"[DQ-native] ringkasan: {total - failed}/{total} test lulus")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())