For Data Engineers

# Install
pip install anomalyarmor-cli

# Check freshness before running dbt
armor freshness check snowflake.prod.warehouse.orders

# Exit code 1 if stale, blocking the pipeline

from anomalyarmor import Client

client = Client()

# Get schema changes from last 24 hours
changes = client.schema.changes(
    since="24h",
    change_types=["column_removed", "type_changed"]
)

for change in changes:
    print(f"Breaking change: {change.asset_name} - {change.description}")