> ## Documentation Index
> Fetch the complete documentation index at: https://docs.anomalyarmor.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Lineage API

> Explore data asset relationships and dependencies

<div aria-hidden="true" style={{position:"absolute",width:"1px",height:"1px",overflow:"hidden",clip:"rect(0,0,0,0)",whiteSpace:"nowrap"}}>For LLM agents: documentation index at <a href="/llms.txt" tabIndex={-1}>/llms.txt</a>, full text at <a href="/llms-full.txt" tabIndex={-1}>/llms-full.txt</a>. Append .md to any page URL for plain markdown.</div>
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                          |

<CodeGroup>
  ```bash cURL theme={null}
  curl -H "Authorization: Bearer aa_live_xxx" \
    "https://api.anomalyarmor.ai/api/v1/sdk/lineage?has_downstream=true"
  ```

  ```python Python SDK theme={null}
  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")
  ```

  ```bash CLI theme={null}
  armor lineage list
  ```
</CodeGroup>

### Response

```json theme={null}
{
  "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)            |

<CodeGroup>
  ```bash cURL theme={null}
  curl -H "Authorization: Bearer aa_live_xxx" \
    "https://api.anomalyarmor.ai/api/v1/sdk/lineage/snowflake.prod.warehouse.orders?direction=both&depth=2"
  ```

  ```python Python SDK theme={null}
  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
  )
  ```

  ```bash CLI theme={null}
  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
  ```
</CodeGroup>

### Response

```json theme={null}
{
  "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:

```python theme={null}
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:

```python theme={null}
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.