Lineage API

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

Method	Endpoint	Description
GET	`/api/v1/sdk/lineage`	List 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

Parameter	Type	Default	Description
`source`	string	-	Filter by data source
`has_upstream`	boolean	-	Only assets with upstream dependencies
`has_downstream`	boolean	-	Only assets with downstream dependencies
`limit`	integer	50	Max results
`offset`	integer	0	Results to skip

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/lineage?has_downstream=true"

from anomalyarmor import Client

client = Client()
assets = client.lineage.list()
for asset in assets:
    print(f"{asset.qualified_name}: {asset.upstream_count} up, {asset.downstream_count} down")

armor lineage list

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

Parameter	Type	Default	Description
`direction`	string	`both`	`upstream`, `downstream`, or `both`
`depth`	integer	1	Levels 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"

from anomalyarmor import Client

client = Client()

# Get full lineage
lineage = client.lineage.get("snowflake.prod.warehouse.orders")

# Upstream only
lineage = client.lineage.get(
    "snowflake.prod.warehouse.orders",
    direction="upstream"
)

# Multi-level
lineage = client.lineage.get(
    "snowflake.prod.warehouse.orders",
    depth=3
)

armor lineage get snowflake.prod.warehouse.orders
armor lineage get snowflake.prod.warehouse.orders --direction upstream
armor lineage get snowflake.prod.warehouse.orders --depth 3

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

Type	Description
`derives_from`	Target is derived from source (transformation)
`joins_with`	Assets are joined together
`feeds_into`	Source feeds into target
`copies_to`	Direct copy relationship
`references`	Lookup/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.

API Overview

API Reference

API Guides

SDKs

CLI

Integrations

AI Agents

Endpoints

List Assets with Lineage

Query Parameters

Response

Get Asset Lineage

Query Parameters

Response

Relationship Types

Use Case: Check Upstream Before Pipeline

Use Case: Impact Analysis

Common Questions

How deep can I traverse lineage in one request?

Where does lineage come from?

What do the relationship types mean?

​Endpoints

​List Assets with Lineage

​Query Parameters

​Response

​Get Asset Lineage

​Query Parameters

​Response

​Relationship Types

​Use Case: Check Upstream Before Pipeline

​Use Case: Impact Analysis

​Common Questions

​How deep can I traverse lineage in one request?

​Where does lineage come from?

​What do the relationship types mean?

Endpoints

List Assets with Lineage

Query Parameters

Response

Get Asset Lineage

Query Parameters

Response

Relationship Types

Use Case: Check Upstream Before Pipeline

Use Case: Impact Analysis

Common Questions

How deep can I traverse lineage in one request?

Where does lineage come from?

What do the relationship types mean?