> ## 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.

# Schema API

> Monitor schema changes and detect drift

<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 Schema API monitors schema changes and detects drift in your data assets. Use it to track column additions, removals, type changes, and trigger on-demand schema checks.

## Endpoints

| Method | Endpoint                          | Description                |
| ------ | --------------------------------- | -------------------------- |
| GET    | `/api/v1/sdk/schema/summary`      | Get schema drift summary   |
| GET    | `/api/v1/sdk/schema/changes`      | List recent schema changes |
| POST   | `/api/v1/sdk/schema/{id}/refresh` | Trigger schema check       |

## Get Schema Summary

```
GET /api/v1/sdk/schema/summary
```

Returns aggregate schema drift statistics.

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

  ```python Python SDK theme={null}
  from anomalyarmor import Client

  client = Client()
  summary = client.schema.summary()
  print(f"Assets with changes: {summary.assets_with_changes}")
  print(f"Changes last 24h: {summary.changes_last_24h}")
  ```

  ```bash CLI theme={null}
  armor schema summary
  ```
</CodeGroup>

### Response

```json theme={null}
{
  "data": {
    "total_assets": 245,
    "assets_with_changes": 12,
    "changes_last_24h": 5,
    "changes_last_7d": 23,
    "by_change_type": {
      "column_added": 15,
      "column_removed": 3,
      "column_type_changed": 8,
      "column_renamed": 2
    }
  }
}
```

## List Schema Changes

```
GET /api/v1/sdk/schema/changes
```

### Query Parameters

| Parameter     | Type     | Default | Description                    |
| ------------- | -------- | ------- | ------------------------------ |
| `asset_id`    | string   | -       | Filter by asset qualified name |
| `change_type` | string   | -       | Filter by change type          |
| `since`       | datetime | -       | Changes since timestamp        |
| `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/schema/changes?change_type=column_removed"
  ```

  ```python Python SDK theme={null}
  from anomalyarmor import Client

  client = Client()

  # List recent changes
  changes = client.schema.changes()
  for change in changes:
      print(f"{change.qualified_name}: {change.change_type} - {change.column_name}")

  # Filter by asset
  changes = client.schema.changes(asset_id="snowflake.prod.warehouse.orders")
  ```

  ```bash CLI theme={null}
  armor schema changes
  armor schema changes --asset snowflake.prod.warehouse.orders
  ```
</CodeGroup>

### Response

```json theme={null}
{
  "data": [
    {
      "id": "change_123",
      "asset_id": "550e8400-e29b-41d4-a716-446655440000",
      "qualified_name": "snowflake.prod.warehouse.orders",
      "change_type": "column_added",
      "column_name": "discount_code",
      "details": {
        "new_type": "VARCHAR(50)",
        "nullable": true
      },
      "detected_at": "2024-12-04T09:15:00Z",
      "previous_schema_hash": "abc123",
      "current_schema_hash": "def456"
    }
  ],
  "pagination": {
    "total": 23,
    "limit": 50,
    "offset": 0,
    "has_more": false
  }
}
```

## Trigger Schema Check

```
POST /api/v1/sdk/schema/{id}/refresh
```

<Note>
  Requires `read-write` or `admin` scope.
</Note>

### Query Parameters

| Parameter | Type    | Default | Description                |
| --------- | ------- | ------- | -------------------------- |
| `wait`    | boolean | false   | Wait for check to complete |

<CodeGroup>
  ```bash cURL theme={null}
  curl -X POST -H "Authorization: Bearer aa_live_xxx" \
    "https://api.anomalyarmor.ai/api/v1/sdk/schema/snowflake.prod.warehouse.orders/refresh?wait=true"
  ```

  ```python Python SDK theme={null}
  from anomalyarmor import Client

  client = Client()

  # Async
  result = client.schema.refresh("snowflake.prod.warehouse.orders")
  print(f"Job: {result.job_id}")

  # Wait for completion
  result = client.schema.refresh("snowflake.prod.warehouse.orders", wait=True)
  if result.changes_detected:
      for change in result.changes:
          print(f"Change: {change.change_type} - {change.column_name}")
  ```

  ```bash CLI theme={null}
  armor schema refresh snowflake.prod.warehouse.orders --wait
  ```
</CodeGroup>

### Response

```json theme={null}
{
  "data": {
    "job_id": "job_xyz789",
    "status": "completed",
    "asset_id": "snowflake.prod.warehouse.orders",
    "result": {
      "changes_detected": true,
      "changes": [
        {
          "change_type": "column_added",
          "column_name": "discount_code",
          "details": {"new_type": "VARCHAR(50)"}
        }
      ]
    }
  }
}
```

## Change Types

| Type                  | Description               |
| --------------------- | ------------------------- |
| `column_added`        | New column added to table |
| `column_removed`      | Column removed from table |
| `column_type_changed` | Column data type changed  |
| `column_renamed`      | Column name changed       |
| `table_created`       | New table created         |
| `table_dropped`       | Table dropped             |

## Use Case: Post-Deploy Schema Check

Trigger a schema check after deploying dbt models:

```python theme={null}
from anomalyarmor import Client

client = Client()

def verify_schema_after_deploy(models: list[str]):
    """Run schema check after dbt deployment."""
    for model in models:
        print(f"Checking schema: {model}")
        result = client.schema.refresh(model, wait=True)

        if result.changes_detected:
            print(f"  Schema changes detected:")
            for change in result.changes:
                print(f"    - {change.change_type}: {change.column_name}")
        else:
            print(f"  No schema changes")

# After dbt run
verify_schema_after_deploy([
    "snowflake.prod.mart.orders_mart",
    "snowflake.prod.mart.customers_mart",
])
```

## Common Questions

### How does AnomalyArmor detect schema changes?

A baseline schema snapshot is taken the first time an asset is profiled. On every scan (scheduled or triggered via `POST /schema/{id}/refresh`), the current column list, types, and nullability are diffed against the baseline and any delta is written to the changes feed. The `previous_schema_hash` and `current_schema_hash` fields let you correlate changes across runs.

### Can I trigger a schema check on demand after a dbt deploy?

Yes. Call `POST /api/v1/sdk/schema/{id}/refresh?wait=true` (or `client.schema.refresh(asset, wait=True)`) in your post-deploy step. The synchronous response includes `changes_detected` and the change list, so you can fail the deploy pipeline if a breaking change lands unexpectedly.

### Which change types are reported?

The feed reports `column_added`, `column_removed`, `column_type_changed`, `column_renamed`, `table_created`, and `table_dropped`. Renames are heuristic and are sometimes reported as paired add/remove when the column's data distribution has also shifted.