Schema API

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.

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/schema/summary"

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}")

armor schema summary

Response

{
  "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

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/schema/changes?change_type=column_removed"

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")

armor schema changes
armor schema changes --asset snowflake.prod.warehouse.orders

Response

{
  "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

Requires read-write or admin scope.

Query Parameters

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

curl -X POST -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/schema/snowflake.prod.warehouse.orders/refresh?wait=true"

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}")

armor schema refresh snowflake.prod.warehouse.orders --wait

Response

{
  "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:

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.

API Overview

API Reference

API Guides

SDKs

CLI

Integrations

AI Agents

Endpoints

Get Schema Summary

Response

List Schema Changes

Query Parameters

Response

Trigger Schema Check

Query Parameters

Response

Change Types

Use Case: Post-Deploy Schema Check

Common Questions

How does AnomalyArmor detect schema changes?

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

Which change types are reported?

​Endpoints

​Get Schema Summary

​Response

​List Schema Changes

​Query Parameters

​Response

​Trigger Schema Check

​Query Parameters

​Response

​Change Types

​Use Case: Post-Deploy Schema Check

​Common Questions

​How does AnomalyArmor detect schema changes?

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

​Which change types are reported?

Endpoints

Get Schema Summary

Response

List Schema Changes

Query Parameters

Response

Trigger Schema Check

Query Parameters

Response

Change Types

Use Case: Post-Deploy Schema Check

Common Questions

How does AnomalyArmor detect schema changes?

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

Which change types are reported?