Atlas lineage for compliance

Atlas is the data governance tool that records what produced what. When a dashboard shows a number, Atlas lets you trace backward through every table and transform that fed it. That is non-optional once an auditor is in the room, and it is a weekly-saver even without them.

services/atlas_service.py
python
import httpx

class AtlasService:
    def __init__(self, opts):
        self.opts = opts

    async def register_lineage(self, payload: str) -> dict:
        if not self.opts.base_url:
            return {'registered': False, 'reason': 'atlas disabled'}
        async with httpx.AsyncClient(timeout=self.opts.request_timeout_seconds) as client:
            response = await client.post(
                f'{self.opts.base_url}/api/atlas/v2/entity/bulk',
                content=payload,
                headers={'Content-Type': 'application/json'},
                auth=(self.opts.username, self.opts.password),
            )
            response.raise_for_status()
            return response.json()

The register_lineage service makes an HTTP POST to Atlas with the lineage payload. The payload links inputs to outputs and includes the DAG run id so every transformation is traceable to the run that produced it.

lineage_payload.json
json
{
  "entities": [
    {
      "typeName": "DataFlow",
      "attributes": {
        "name": "batch_transform_2026_04_23",
        "qualifiedName": "pipeline.batch_transform@PROD",
        "inputs": ["mysql://orders", "minio://raw-data/orders/dt=2026-04-23"],
        "outputs": ["postgres://fact_orders", "postgres://dim_customers"],
        "producedBy": "airflow.dag.batch_ingestion_dag.run.2026-04-23T00:00:00"
      }
    }
  ]
}

A minimal lineage entity. Inputs reference the source tables. Outputs reference the produced table. Atlas links them in the UI so a compliance user can click from the dashboard down to the MySQL source that fed it.

Register per transformation, as the transformation succeeds. Batch registration at end of DAG loses the per-table trace and makes partial-failure analysis painful. A small per-task call keeps Atlas current and the lineage graph useful during an active incident.

Validation checklist: Verify lineage registration

Loading practice…

Quiz: Quiz

Loading practice…