Skip to main content
The Lineage API provides access to data asset relationships, showing upstream sources and downstream dependencies. Use it for impact analysis and understanding data flow.

​
Endpoints

MethodEndpointDescription
GET/api/v1/sdk/lineageList assets with lineage info
GET/api/v1/sdk/lineage/{id}Get lineage for specific asset

​
List Assets with Lineage

GET /api/v1/sdk/lineage

​
Query Parameters

ParameterTypeDefaultDescription
sourcestring-Filter by data source
has_upstreamboolean-Only assets with upstream dependencies
has_downstreamboolean-Only assets with downstream dependencies
limitinteger50Max results
offsetinteger0Results to skip
curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/lineage?has_downstream=true"

​
Response

{
  "data": [
    {
      "asset_id": "550e8400-e29b-41d4-a716-446655440000",
      "qualified_name": "snowflake.prod.warehouse.orders",
      "upstream_count": 3,
      "downstream_count": 5
    }
  ],
  "pagination": {
    "total": 100,
    "limit": 50,
    "offset": 0,
    "has_more": true
  }
}

​
Get Asset Lineage

GET /api/v1/sdk/lineage/{id}

​
Query Parameters

ParameterTypeDefaultDescription
directionstringbothupstream, downstream, or both
depthinteger1Levels to traverse (1-5)
curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/lineage/snowflake.prod.warehouse.orders?direction=both&depth=2"

​
Response

{
  "data": {
    "asset_id": "550e8400-e29b-41d4-a716-446655440000",
    "qualified_name": "snowflake.prod.warehouse.orders",
    "upstream": [
      {
        "asset_id": "raw_orders_id",
        "qualified_name": "snowflake.raw.stripe.orders",
        "asset_type": "table",
        "relationship": "derives_from",
        "depth": 1,
        "upstream": [
          {
            "qualified_name": "stripe.api.orders",
            "asset_type": "api",
            "depth": 2
          }
        ]
      },
      {
        "asset_id": "raw_customers_id",
        "qualified_name": "snowflake.raw.crm.customers",
        "asset_type": "table",
        "relationship": "joins_with",
        "depth": 1
      }
    ],
    "downstream": [
      {
        "asset_id": "orders_mart_id",
        "qualified_name": "snowflake.prod.mart.orders_mart",
        "asset_type": "table",
        "relationship": "feeds_into",
        "depth": 1
      },
      {
        "asset_id": "dashboard_id",
        "qualified_name": "looker.sales_dashboard",
        "asset_type": "dashboard",
        "relationship": "feeds_into",
        "depth": 1
      }
    ]
  }
}

​
Relationship Types

TypeDescription
derives_fromTarget is derived from source (transformation)
joins_withAssets are joined together
feeds_intoSource feeds into target
copies_toDirect copy relationship
referencesLookup/reference relationship

​
Use Case: Check Upstream Before Pipeline

Verify all upstream sources are fresh before running: 
from anomalyarmor import Client

client = Client()

def check_upstream_freshness(asset: str):
    """Check all upstream sources are fresh."""
    lineage = client.lineage.get(asset, direction="upstream")

    print(f"Checking {len(lineage.upstream)} upstream sources...")

    for upstream in lineage.upstream:
        client.freshness.require_fresh(upstream.qualified_name)
        print(f"  {upstream.qualified_name}")

    print("All upstream sources are fresh!")

check_upstream_freshness("snowflake.prod.mart.orders_summary")

​
Use Case: Impact Analysis

Before making schema changes, check downstream impact: 
from anomalyarmor import Client

client = Client()

def check_impact(asset: str):
    """Check downstream impact before schema change."""
    lineage = client.lineage.get(asset, direction="downstream", depth=3)

    print(f"Downstream impact for {asset}:")
    print(f"Total downstream assets: {len(lineage.downstream)}")

    # Group by type
    by_type = {}
    for dep in lineage.downstream:
        by_type.setdefault(dep.asset_type, []).append(dep)

    for asset_type, assets in by_type.items():
        print(f"  {asset_type}: {len(assets)}")
        for asset in assets[:3]:
            print(f"    - {asset.qualified_name}")

check_impact("snowflake.prod.warehouse.orders")
Output: 
Downstream impact for snowflake.prod.warehouse.orders:
Total downstream assets: 8
  table: 5
    - snowflake.prod.mart.orders_mart
    - snowflake.prod.mart.revenue_daily
    - snowflake.prod.mart.customer_orders
  dashboard: 3
    - looker.sales_dashboard
    - looker.executive_summary
    - tableau.orders_analysis

​
Common Questions

​
How deep can I traverse lineage in one request?

Pass depth from 1 to 5 on GET /lineage/{id}. Depth 1 returns direct neighbors, which is cheap and enough for most impact-analysis UIs. Higher depths fan out quickly, so avoid depth 5 in tight loops and paginate the results you consume.

​
Where does lineage come from?

Lineage is derived from warehouse query history, dbt manifests, and view definitions for supported sources, then enriched with dashboard consumers (Looker, Tableau, Mode). You won’t see relationships for sources that don’t expose query history or for ad-hoc scripts run outside the warehouse.

​
What do the relationship types mean?

derives_from marks transformation-based derivation (CREATE TABLE AS, dbt models). joins_with captures co-used inputs. feeds_into describes a non-derivation downstream consumer (e.g., a dashboard). copies_to is a direct copy, and references is a lookup/FK-style join.