---

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.