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.
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.
{
"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…