armor-monitor

Set Up Monitoring

Configure freshness monitoring and schema drift detection for your data assets.

Prerequisites

• AnomalyArmor API key configured (~/.armor/config.yaml or ARMOR_API_KEY env var)
• Python SDK installed (pip install anomalyarmor)
• Data source already connected (use /armor:connect first)

When to Use

• "Set up freshness monitoring for the orders table"
• "Monitor my critical tables"
• "Enable schema drift detection"
• "Alert me when data is stale"
• "Track schema changes"

Steps

For Freshness Monitoring

1. Identify the asset and table to monitor
2. Determine check interval (how often to check)
3. Choose monitoring mode (auto_learn or explicit threshold)
4. Create schedule with client.freshness.create_schedule()

For Schema Monitoring

1. Identify the asset to monitor
2. Create baseline with client.schema.create_baseline()
3. Enable monitoring with client.schema.enable_monitoring()

Example Usage

Set Up Freshness Monitoring (Auto-Learn)

from anomalyarmor import Client

client = Client()

# List existing schedules for the asset
schedules = client.freshness.list_schedules(asset_id="asset-uuid")
print(f"Existing schedules: {len(schedules)}")

# Create freshness schedule with auto-learn
# System will learn normal update patterns and alert on deviations
schedule = client.freshness.create_schedule(
    asset_id="asset-uuid",
    table_path="public.orders",
    check_interval="1h",  # Check every hour
    monitoring_mode="auto_learn"
)

print(f"Created schedule: {schedule.id}")
print(f"Table: {schedule.table_path}")
print(f"Check interval: {schedule.check_interval}")

Set Up Freshness with Explicit Threshold

from anomalyarmor import Client

client = Client()

# Create schedule with explicit threshold
# Alert if table hasn't updated in 24 hours
schedule = client.freshness.create_schedule(
    asset_id="asset-uuid",
    table_path="public.daily_summary",
    check_interval="6h",
    monitoring_mode="explicit",
    expected_interval_hours=24,
    freshness_column="updated_at"  # Optional: specify column
)

print(f"Created schedule with {schedule.expected_interval_hours}h threshold")