Freshness API

Monitor and validate data freshness

The Freshness API enables monitoring and validation of data freshness. Use it to check if your data is up-to-date and trigger on-demand freshness checks.

Endpoints

Method	Endpoint	Description
GET	/api/v1/sdk/freshness	List freshness status for all assets
GET	/api/v1/sdk/freshness/summary	Get freshness summary statistics
GET	/api/v1/sdk/freshness/{id}	Get freshness status for an asset
POST	/api/v1/sdk/freshness/{id}/refresh	Trigger freshness check

Get Freshness Summary

GET /api/v1/sdk/freshness/summary

Returns aggregate freshness statistics across all monitored assets.

cURL

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

Response

{
  "data": {
    "total_assets": 245,
    "fresh": 230,
    "stale": 10,
    "unknown": 5,
    "fresh_percentage": 93.9,
    "by_source": {
      "snowflake": {"total": 150, "fresh": 145, "stale": 5},
      "bigquery": {"total": 95, "fresh": 85, "stale": 5}
    }
  }
}

List Freshness Status

GET /api/v1/sdk/freshness

Query Parameters

Parameter	Type	Default	Description
status	string	-	Filter: fresh, stale, or unknown
source	string	-	Filter by data source
limit	integer	50	Max results (max: 100)
offset	integer	0	Results to skip

cURL

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/freshness?status=stale&limit=10"

Response

{
  "data": [
    {
      "asset_id": "550e8400-e29b-41d4-a716-446655440000",
      "qualified_name": "snowflake.prod.warehouse.orders",
      "status": "fresh",
      "is_fresh": true,
      "last_updated": "2024-12-04T08:15:00Z",
      "threshold_hours": 24,
      "hours_since_update": 2.5,
      "checked_at": "2024-12-04T10:30:00Z"
    }
  ],
  "pagination": {
    "total": 100,
    "limit": 50,
    "offset": 0,
    "has_more": true
  }
}

Get Asset Freshness

GET /api/v1/sdk/freshness/{id}

Get detailed freshness status for a specific asset.

cURL

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/freshness/snowflake.prod.warehouse.orders"

Response

{
  "data": {
    "asset_id": "550e8400-e29b-41d4-a716-446655440000",
    "qualified_name": "snowflake.prod.warehouse.orders",
    "status": "fresh",
    "is_fresh": true,
    "last_updated": "2024-12-04T08:15:00Z",
    "threshold_hours": 24,
    "hours_since_update": 2.5,
    "checked_at": "2024-12-04T10:30:00Z",
    "history": [
      {"checked_at": "2024-12-04T10:30:00Z", "status": "fresh"},
      {"checked_at": "2024-12-04T06:30:00Z", "status": "fresh"},
      {"checked_at": "2024-12-03T22:30:00Z", "status": "stale"}
    ]
  }
}

Trigger Freshness Check

POST /api/v1/sdk/freshness/{id}/refresh

Requires read-write or admin scope.

Query Parameters

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

cURL (async)

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

Response (async)

{
  "data": {
    "job_id": "job_abc123",
    "status": "pending",
    "asset_id": "snowflake.prod.warehouse.orders"
  }
}

Response (sync with wait=true)

{
  "data": {
    "job_id": "job_abc123",
    "status": "completed",
    "asset_id": "snowflake.prod.warehouse.orders",
    "result": {
      "is_fresh": true,
      "last_updated": "2024-12-04T10:35:00Z"
    }
  }
}

Gate Pattern: require_fresh()

The SDK provides a convenient gate pattern for pipelines:

from anomalyarmor import Client
from anomalyarmor.exceptions import StalenessError

client = Client()

try:
    # Raises StalenessError if data is stale
    client.freshness.require_fresh("snowflake.prod.warehouse.orders")
    print("Data is fresh, proceeding...")
except StalenessError as e:
    print(f"Data is stale: last updated {e.last_updated}")
    raise  # Fail the pipeline

Use require_fresh() in Airflow tasks to automatically fail pipelines when upstream data is stale. See the Airflow Integration guide.

CLI Check Command

The CLI provides a check command that exits with code 1 if data is stale:

# Exit 0 if fresh, exit 1 if stale
armor freshness check snowflake.prod.warehouse.orders
echo $?  # 0 = fresh, 1 = stale

# Use in shell scripts
if armor freshness check snowflake.prod.warehouse.orders; then
    echo "Data is fresh"
    dbt run
else
    echo "Data is stale, aborting"
    exit 1
fi

Error Responses

Asset Not Found (404)

{
  "error": {
    "code": "ASSET_NOT_FOUND",
    "message": "Asset not found",
    "details": {"asset_id": "invalid.qualified.name"}
  }
}

Forbidden (403)

When attempting to trigger refresh without proper scope:

{
  "error": {
    "code": "FORBIDDEN",
    "message": "Insufficient permissions. Required scope: read-write",
    "details": {"current_scope": "read-only", "required_scope": "read-write"}
  }
}