Platform Integrations
=====================

metapod extracts metadata locally. You choose where it goes.

Dataflix (built-in)
-------------------

.. code-block:: bash

   metapod push --target https://dataflix.macalab.org \
                --project my-project \
                --api-key $API_KEY

Dataflix converts probe JSON to DDL, builds a lineage graph, and provides
a visual explorer with 5 Knowledge Graph views.

DataHub
-------

Convert metapod output to DataHub Metadata Change Events (MCEs):

.. code-block:: python

   """Push metapod probe results to DataHub."""
   import json
   from pathlib import Path
   from datahub.emitter.rest_emitter import DatahubRestEmitter
   from datahub.metadata.schema_classes import (
       DatasetSnapshotClass, DatasetPropertiesClass, SchemaMetadataClass,
       SchemaFieldClass, MySqlDDLClass,
   )

   emitter = DatahubRestEmitter("http://datahub-gms:8080")

   # Push tables
   tables = json.load(open("output/oracle/metadata/tables.json"))
   for t in tables["records"]:
       platform = tables["source"]["type"]
       urn = f"urn:li:dataset:(urn:li:dataPlatform:{platform},{t['table_name']},PROD)"
       snapshot = DatasetSnapshotClass(
           urn=urn,
           aspects=[
               DatasetPropertiesClass(
                   name=t["table_name"],
                   customProperties={"owner": t.get("owner", ""), "row_count": str(t.get("row_count", ""))},
               ),
           ],
       )
       emitter.emit_mce(snapshot)

   # Push columns
   columns = json.load(open("output/oracle/metadata/columns.json"))
   col_map = {}
   for c in columns["records"]:
       col_map.setdefault(c["table_name"], []).append(c)

   for table_name, cols in col_map.items():
       urn = f"urn:li:dataset:(urn:li:dataPlatform:{platform},{table_name},PROD)"
       fields = [
           SchemaFieldClass(
               fieldPath=c["column_name"],
               nativeDataType=c.get("data_type", ""),
               type={"type": {"com.linkedin.schema.StringType": {}}},
           )
           for c in sorted(cols, key=lambda x: x.get("ordinal_position", 0))
       ]
       emitter.emit_mce(DatasetSnapshotClass(
           urn=urn,
           aspects=[SchemaMetadataClass(
               schemaName=table_name, platform=f"urn:li:dataPlatform:{platform}",
               hash="", platformSchema=MySqlDDLClass(tableSchema=""),
               fields=fields,
           )],
       ))

OpenMetadata
------------

Use the OpenMetadata REST API to push probe results:

.. code-block:: python

   """Push metapod probe results to OpenMetadata."""
   import json
   import httpx

   OM_URL = "http://openmetadata:8585/api/v1"
   TOKEN = "your-jwt-token"
   headers = {"Authorization": f"Bearer {TOKEN}"}

   tables = json.load(open("output/postgres/metadata/tables.json"))
   for t in tables["records"]:
       httpx.put(
           f"{OM_URL}/tables",
           headers=headers,
           json={
               "name": t["table_name"],
               "databaseSchema": f"postgres.public",
               "columns": [],  # populate from columns.json
               "tableType": "Regular",
           },
       )

Apache Atlas / Purview
----------------------

metapod already has a Purview probe (``purview.metadata.*``). For bidirectional
sync, push results back:

.. code-block:: python

   """Push metapod results to Purview as custom entities."""
   import json
   import httpx

   PURVIEW_ACCOUNT = "my-purview"
   token = get_purview_token()  # OAuth2 client_credentials flow

   tables = json.load(open("output/oracle/metadata/tables.json"))
   for t in tables["records"]:
       httpx.post(
           f"https://{PURVIEW_ACCOUNT}.purview.azure.com/catalog/api/atlas/v2/entity",
           headers={"Authorization": f"Bearer {token}"},
           json={
               "entity": {
                   "typeName": "azure_sql_table",
                   "attributes": {
                       "qualifiedName": f"oracle://{t['table_name']}",
                       "name": t["table_name"],
                   },
               }
           },
       )

Custom REST API
---------------

Generic pattern for pushing to any REST endpoint:

.. code-block:: python

   """Push all probe results to a custom API."""
   import json
   import httpx
   from pathlib import Path

   API_URL = "https://my-catalog.example.com/api/ingest"
   API_KEY = "your-key"

   for jf in Path("./output").rglob("*.json"):
       data = json.load(jf.open())
       if not data.get("records"):
           continue
       httpx.post(
           API_URL,
           headers={"Authorization": f"Bearer {API_KEY}"},
           json=data,
           timeout=30,
       )

Webhook / Event Streaming
--------------------------

Pipe probe results to a message broker:

.. code-block:: python

   """Emit probe results as Kafka messages."""
   import json
   from pathlib import Path
   from kafka import KafkaProducer

   producer = KafkaProducer(
       bootstrap_servers="kafka:9092",
       value_serializer=lambda v: json.dumps(v).encode(),
   )

   for jf in Path("./output").rglob("*.json"):
       data = json.load(jf.open())
       producer.send("metadata-events", value=data)
   producer.flush()