Alerts API

Query alert history and status

The Alerts API provides access to data quality alerts triggered by AnomalyArmor monitoring. Use it to query alert history, check active incidents, and integrate with your incident response workflows.

Endpoints

Method	Endpoint	Description
GET	/api/v1/sdk/alerts/summary	Get alerts summary
GET	/api/v1/sdk/alerts	List alerts
GET	/api/v1/sdk/alerts/rules	List alert rules

Get Alerts Summary

GET /api/v1/sdk/alerts/summary

Returns aggregate alert statistics.

cURL

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

Response

{
  "data": {
    "total": 45,
    "by_status": {
      "triggered": 5,
      "acknowledged": 3,
      "resolved": 37
    },
    "by_type": {
      "freshness": 20,
      "schema_drift": 15,
      "row_count": 10
    },
    "triggered_last_24h": 8,
    "triggered_last_7d": 25
  }
}

List Alerts

GET /api/v1/sdk/alerts

Query Parameters

Parameter	Type	Default	Description
status	string	-	Filter: triggered, acknowledged, resolved
alert_type	string	-	Filter: freshness, schema_drift, row_count
asset_id	string	-	Filter by asset qualified name
since	datetime	-	Alerts 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/alerts?status=triggered"

Response

{
  "data": [
    {
      "id": "alert_123",
      "alert_type": "freshness",
      "status": "triggered",
      "asset_id": "550e8400-e29b-41d4-a716-446655440000",
      "qualified_name": "snowflake.prod.warehouse.orders",
      "message": "Data is 26 hours stale (threshold: 24 hours)",
      "triggered_at": "2024-12-04T08:00:00Z",
      "acknowledged_at": null,
      "resolved_at": null,
      "details": {
        "last_updated": "2024-12-03T06:00:00Z",
        "threshold_hours": 24,
        "hours_stale": 26
      },
      "rule_id": "rule_freshness_orders"
    }
  ],
  "pagination": {
    "total": 5,
    "limit": 50,
    "offset": 0,
    "has_more": false
  }
}

List Alert Rules

GET /api/v1/sdk/alerts/rules

Returns configured alert rules for your organization.

cURL

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

Response

{
  "data": [
    {
      "id": "rule_freshness_orders",
      "name": "Orders Freshness",
      "alert_type": "freshness",
      "asset_id": "snowflake.prod.warehouse.orders",
      "enabled": true,
      "config": {
        "threshold_hours": 24
      },
      "destinations": ["slack", "email"],
      "created_at": "2024-01-15T10:00:00Z"
    }
  ],
  "pagination": {
    "total": 15,
    "limit": 50,
    "offset": 0,
    "has_more": false
  }
}

Alert Types

Type	Description
freshness	Data is stale beyond threshold
schema_drift	Schema change detected
row_count	Row count anomaly detected
null_percentage	Null percentage exceeded threshold
duplicate_keys	Duplicate primary keys detected

Alert Statuses

Status	Description
triggered	Alert is active and needs attention
acknowledged	Alert has been seen but not resolved
resolved	Alert has been resolved

Use Case: Incident Response

Build an incident response script:

from anomalyarmor import Client

client = Client()

def get_active_incidents():
    """Get all active alerts for incident response."""
    alerts = client.alerts.list(status="triggered")

    by_type = {"schema": [], "freshness": [], "discovery": []}
    for alert in alerts:
        by_type[alert.alert_type].append(alert)

    print("=== Active Incidents ===")
    for alert_type in ["schema", "freshness", "discovery"]:
        if by_type[alert_type]:
            print(f"\n{alert_type.upper()} ({len(by_type[alert_type])}):")
            for alert in by_type[alert_type]:
                print(f"  - {alert.qualified_name}: {alert.message}")

    return alerts

incidents = get_active_incidents()

Output:

=== Active Incidents ===

SCHEMA (1):
  - snowflake.prod.warehouse.payments: Column removed: deprecated_field

FRESHNESS (2):
  - snowflake.prod.warehouse.orders: Data is 26 hours stale
  - bigquery.analytics.events: Data is 4 hours stale

Use Case: Slack Bot Integration

Post alerts to Slack:

from anomalyarmor import Client
import requests

client = Client()
SLACK_WEBHOOK = "https://hooks.slack.com/services/..."

def post_alerts_to_slack():
    """Post triggered alerts to Slack."""
    alerts = client.alerts.list(status="triggered")

    if not alerts:
        return

    blocks = [{"type": "header", "text": {"type": "plain_text", "text": f"{len(alerts)} Active Alerts"}}]

    for alert in alerts[:5]:  # Limit to 5
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*{alert.qualified_name}*\n{alert.message}"
            }
        })

    requests.post(SLACK_WEBHOOK, json={"blocks": blocks})

# Run periodically
post_alerts_to_slack()