armor-lineage

Data Lineage

Explore upstream dependencies and downstream consumers of your data assets.

Prerequisites

• AnomalyArmor API key configured (~/.armor/config.yaml or ARMOR_API_KEY env var)
• Python SDK installed (pip install anomalyarmor)

When to Use

• "What depends on this table?"
• "Where does this data come from?"
• "Impact analysis for changes"
• "Show upstream dependencies"
• "List downstream consumers"
• "Trace data flow"

Concepts

Directions

• upstream: Tables that this table depends on (data sources)
• downstream: Tables that depend on this table (data consumers)
• both: Full lineage graph in both directions

Depth

• depth=1: Direct dependencies only
• depth=2: Two levels of dependencies
• depth=3+: Extended dependency chain (max 5)

Steps

1. Identify the asset to analyze
2. Determine direction (upstream, downstream, or both)
3. Choose appropriate depth (start with 1-2)
4. Call client.lineage.get() to fetch lineage graph
5. Analyze the graph for impact or root cause

Example Usage

Find Upstream Dependencies

from anomalyarmor import Client

client = Client()

# Get upstream lineage (where data comes from)
lineage = client.lineage.get(
    asset_id="asset-uuid",
    direction="upstream",
    depth=2
)

print(f"Table: {lineage.root.qualified_name}")
print(f"\nUpstream Dependencies ({len(lineage.upstream)} tables):")
for node in lineage.upstream:
    print(f"  {node.qualified_name}")
    if node.asset_type:
        print(f"    Type: {node.asset_type}")

Find Downstream Consumers

# Get downstream lineage (what depends on this)
lineage = client.lineage.get(
    asset_id="asset-uuid",
    direction="downstream",
    depth=2
)

print(f"Table: {lineage.root.qualified_name}")
print(f"\nDownstream Consumers ({len(lineage.downstream)} tables):")
for node in lineage.downstream:
    print(f"  {node.qualified_name}")

Impact Analysis

# Full impact analysis before making changes
lineage = client.lineage.get(
    asset_id="asset-uuid",
    direction="both",
    depth=3
)

print("=== IMPACT ANALYSIS ===")
print(f"\nTable: {lineage.root.qualified_name}")

# Upstream (what feeds this table)
print(f"\nData Sources ({len(lineage.upstream)} tables):")
for node in lineage.upstream:
    print(f"  <- {node.qualified_name}")

# Downstream (what will be affected by changes)
print(f"\nWill Impact ({len(lineage.downstream)} tables):")
for node in lineage.downstream:
    print(f"  -> {node.qualified_name}")

# Edges show the relationships
print(f"\nRelationships ({len(lineage.edges)} edges):")
for edge in lineage.edges:
    print(f"  {edge.source} -> {edge.target}")