Lineage API

Explore data asset relationships and dependencies

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

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

Parameter	Type	Default	Description
direction	string	both	upstream, downstream, or both
depth	integer	1	Levels to traverse (1-5)

cURL

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

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