Freshness API

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

from anomalyarmor import Client

client = Client()
summary = client.freshness.summary()
print(f"Fresh: {summary.fresh}/{summary.total_assets} ({summary.fresh_percentage}%)")

armor 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 -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/freshness?status=stale&limit=10"

from anomalyarmor import Client

client = Client()

# List all
all_status = client.freshness.list()

# Filter stale only
stale = client.freshness.list(status="stale")
for item in stale:
    print(f"{item.qualified_name}: {item.hours_since_update}h stale")

armor freshness list --status stale

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 -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/freshness/snowflake.prod.warehouse.orders"

from anomalyarmor import Client

client = Client()
status = client.freshness.get("snowflake.prod.warehouse.orders")

print(f"Fresh: {status.is_fresh}")
print(f"Last updated: {status.last_updated}")
print(f"Hours since update: {status.hours_since_update}")

armor freshness get 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 -X POST -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/freshness/snowflake.prod.warehouse.orders/refresh"

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

from anomalyarmor import Client

client = Client()

# Async (returns immediately)
result = client.freshness.refresh("snowflake.prod.warehouse.orders")
print(f"Job ID: {result.job_id}")

# Sync (waits for completion)
result = client.freshness.refresh(
    "snowflake.prod.warehouse.orders",
    wait=True
)
print(f"Fresh: {result.is_fresh}")

# Async
armor freshness refresh snowflake.prod.warehouse.orders

# Wait for completion
armor freshness refresh snowflake.prod.warehouse.orders --wait

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"}
  }
}

Common Questions

What’s the difference between the async and sync refresh calls?

Without wait=true, POST /freshness/{id}/refresh returns immediately with a job_id and status=pending. With wait=true, the request blocks until the check completes and returns the actual is_fresh result. Use async for fire-and-forget cron triggers and sync when a pipeline needs the answer before proceeding.

How do I gate an Airflow or dbt pipeline on freshness?

In Python, call client.freshness.require_fresh("...") which raises StalenessError when stale, which fails the task cleanly. In shell, run armor freshness check <asset> and rely on exit code 1 for stale. The Airflow integration guide shows the full DAG pattern.

Where does the freshness threshold come from?

Each asset has a threshold_hours configured from the dashboard (Freshness tab on the asset page) or inferred from historical update cadence. The API response shows the active threshold_hours next to hours_since_update so you can display both in your own UI.

Why does freshness return status=unknown?

unknown means AnomalyArmor hasn’t yet observed enough update history to decide, typically for newly connected assets or tables that haven’t been written to since the source was connected. Trigger a refresh to get a current read, and give the asset a few update cycles for the baseline to fill in.

API Overview

API Reference

API Guides

SDKs

CLI

Integrations

AI Agents

Endpoints

Get Freshness Summary

Response

List Freshness Status

Query Parameters

Response

Get Asset Freshness

Response

Trigger Freshness Check

Query Parameters

Response (async)

Response (sync with wait=true)

Gate Pattern: require_fresh()

CLI Check Command

Error Responses

Asset Not Found (404)

Forbidden (403)

Common Questions

What’s the difference between the async and sync refresh calls?

How do I gate an Airflow or dbt pipeline on freshness?

Where does the freshness threshold come from?

Why does freshness return status=unknown?

​Endpoints

​Get Freshness Summary

​Response

​List Freshness Status

​Query Parameters

​Response

​Get Asset Freshness

​Response

​Trigger Freshness Check

​Query Parameters

​Response (async)

​Response (sync with wait=true)

​Gate Pattern: require_fresh()

​CLI Check Command

​Error Responses

​Asset Not Found (404)

​Forbidden (403)

​Common Questions

​What’s the difference between the async and sync refresh calls?

​How do I gate an Airflow or dbt pipeline on freshness?

​Where does the freshness threshold come from?

​Why does freshness return status=unknown?

Endpoints

Get Freshness Summary

Response

List Freshness Status

Query Parameters

Response

Get Asset Freshness

Response

Trigger Freshness Check

Query Parameters

Response (async)

Response (sync with wait=true)

Gate Pattern: require_fresh()

CLI Check Command

Error Responses

Asset Not Found (404)

Forbidden (403)

Common Questions

What’s the difference between the async and sync refresh calls?

How do I gate an Airflow or dbt pipeline on freshness?

Where does the freshness threshold come from?

Why does freshness return status=unknown?