---
title: "Stop Playing Data Detective: Automate Pipeline Dependency Tracing with DataLineage"
published: false
tags: [dataengineering, python, tutorial, dataquality]
---

# Stop Playing Data Detective: Automate Pipeline Dependency Tracing with DataLineage

You know the feeling. It's 2pm on a Tuesday. Someone changed a column name in a source table. By 4pm, three dashboards are broken, a Spark job is throwing cryptic NullPointerExceptions, and your Slack DMs look like a crime scene.

The culprit isn't the schema change — it's the fact that nobody *knew* what depended on that column.

This tutorial walks you through wiring DataLineage into your stack so that the next time a schema shifts, you're the person who already has the answer before anyone asks the question.

---

## What We're Building

By the end of this post, you'll have a working Python script that:

1. Traces dependencies across a mixed pipeline (dbt + Airflow + custom ETL)
2. Stores a lineage graph you can query later
3. Runs an impact analysis *before* a schema change ships

We'll use three endpoints throughout:
- `POST /lineage/trace` — discover and register dependencies
- `GET /lineage/{id}` — retrieve a stored lineage graph
- `POST /lineage/impact` — simulate the blast radius of a change

---

## Prerequisites

Set up your environment:

---

## Step 1: Build Your API Client

Before touching any pipeline logic, let's build a thin client wrapper. This keeps auth and error handling in one place — a pattern you'll thank yourself for at 2am.

def _handle_response(self, response: requests.Response) -> dict:
    """Centralized response handling with meaningful error messages."""
    try:
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        error_body = response.json() if response.content else {}
        raise RuntimeError(
            f"API error {response.status_code}: "
            f"{error_body.get('message', str(e))}"
        ) from e
    except requests.exceptions.ConnectionError:
        raise RuntimeError(
            "Could not reach DataLineage API. Check your DATALINEAGE_BASE_URL."
        )

def trace(self, payload: dict) -> dict:
    response = requests.post(
        f"{self.base_url}/lineage/trace",
        json=payload,
        headers=self.headers,
        timeout=30
    )
    return self._handle_response(response)

def get_lineage(self, lineage_id: str) -> dict:
    response = requests.get(
        f"{self.base_url}/lineage/{lineage_id}",
        headers=self.headers,
        timeout=15
    )
    return self._handle_response(response)

def impact(self, payload: dict) -> dict:
    response = requests.post(
        f"{self.base_url}/lineage/impact",
        json=payload,
        headers=self.headers,
        timeout=30
    )
    return self._handle_response(response)

---

## Step 2: Trace Your First Pipeline

Now let's register a real dependency graph. The payload describes your pipeline topology — what tools are involved and which assets connect them.

**Expected output:**

Save that `lineage_id`. It's your handle to everything downstream.

---

## Step 3: Inspect the Dependency Graph

Got your ID? Let's pull the full graph and make it human-readable.

# Find root nodes (no incoming edges)
targets = {e["to"] for e in edges}
roots = [n for n in nodes if n not in targets]

def render(node_id, depth=0):
    node = nodes[node_id]
    prefix = "  " * depth + ("└─ " if depth > 0 else "")
    print(f"{prefix}[{node['tool']}] {node['asset_name']}")
    children = [e["to"] for e in edges if e["from"] == node_id]
    for child in children:
        render(child, depth + 1)

print("\n📊 Dependency Tree:")
print("=" * 40)
for root in roots:
    render(root)

**Expected output:**

This is the map that saves you on Tuesday afternoons.

---

## Step 4: Run Impact Analysis Before Shipping a Change

Here's where DataLineage earns its keep. Before you rename `amount_cents` to `amount_usd` in your source schema, ask the API what breaks.

print("\n⚡ Impact Analysis Report")
print("=" * 40)
print(f"Proposed change: {impact['change_summary']}")
print(f"Affected assets: {impact['total_affected']}\n")

for affected in impact["affected_assets"]:
    icon = severity_icons.get(affected["severity"], "⚪")
    print(f"{icon} [{affected['tool']}] {affected['asset_name']}")
    print(f"     Reason: {affected['reason']}")
    print(f"     Columns at risk: {', '.join(affected['columns_at_risk'])}")
    print()

if impact["total_affected"] == 0:
    print("✅ No downstream consumers affected. Safe to ship.")

**Expected output:**

Two affected assets, surfaced in seconds. Before you type a single migration script.

---

## Putting It All Together: A Pre-Deployment Hook

Combine everything into a script you can run in CI before any schema migration merges:

print(f"🔍 Running impact analysis for lineage: {lineage_id}")

try:
    impact = client.impact({"lineage_id": lineage_id, **change_payload})
    high_severity = [
        a for a in impact["affected_assets"] 
        if a["severity"] == "high"
    ]

    if high_severity:
        print(f"🚫 Deployment blocked: {len(high_severity)} high-severity impact(s) detected.")
        for asset in high_severity:
            print(f"   - [{asset['tool']}] {asset['asset_name']}: {asset['reason']}")
        return False

    print(f"✅ No high-severity impacts. Deployment cleared.")
    return True

except RuntimeError as e:
    print(f"⚠️  Could not complete impact analysis: {e}")
    print("   Blocking deployment as a precaution.")
    return False

Drop this in your CI pipeline. Schema changes that break downstream consumers never reach production again.

---

## What's Next

You've got the foundation. From here, consider:

- **Scheduling `trace` calls** in Airflow after each pipeline run to keep lineage fresh
- **Alerting on `health_status` changes** from `GET /lineage/{id}` — catch drift before users do
- **Extending the impact payload** with `change_type: "column_drop"` for deletion risk analysis

The goal isn't just knowing what broke — it's building a system where you know *before* it breaks. That's the difference between being reactive and being the engineer everyone trusts to ship safely.

Now go enjoy your Tuesday afternoons.