Schema API

Monitor schema changes and detect drift

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

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/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

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

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

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

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