> ## Documentation Index
> Fetch the complete documentation index at: https://docs.anomalyarmor.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Freshness API

> Monitor and validate data freshness

<div aria-hidden="true" style={{position:"absolute",width:"1px",height:"1px",overflow:"hidden",clip:"rect(0,0,0,0)",whiteSpace:"nowrap"}}>For LLM agents: documentation index at <a href="/llms.txt" tabIndex={-1}>/llms.txt</a>, full text at <a href="/llms-full.txt" tabIndex={-1}>/llms-full.txt</a>. Append .md to any page URL for plain markdown.</div>
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.

<CodeGroup>
  ```bash cURL theme={null}
  curl -H "Authorization: Bearer aa_live_xxx" \
    "https://api.anomalyarmor.ai/api/v1/sdk/freshness/summary"
  ```

  ```python Python SDK theme={null}
  from anomalyarmor import Client

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

  ```bash CLI theme={null}
  armor freshness summary
  ```
</CodeGroup>

### Response

```json theme={null}
{
  "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                        |

<CodeGroup>
  ```bash cURL theme={null}
  curl -H "Authorization: Bearer aa_live_xxx" \
    "https://api.anomalyarmor.ai/api/v1/sdk/freshness?status=stale&limit=10"
  ```

  ```python Python SDK theme={null}
  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")
  ```

  ```bash CLI theme={null}
  armor freshness list --status stale
  ```
</CodeGroup>

### Response

```json theme={null}
{
  "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.

<CodeGroup>
  ```bash cURL theme={null}
  curl -H "Authorization: Bearer aa_live_xxx" \
    "https://api.anomalyarmor.ai/api/v1/sdk/freshness/snowflake.prod.warehouse.orders"
  ```

  ```python Python SDK theme={null}
  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}")
  ```

  ```bash CLI theme={null}
  armor freshness get snowflake.prod.warehouse.orders
  ```
</CodeGroup>

### Response

```json theme={null}
{
  "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
```

<Note>
  Requires `read-write` or `admin` scope.
</Note>

### Query Parameters

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

<CodeGroup>
  ```bash cURL (async) theme={null}
  curl -X POST -H "Authorization: Bearer aa_live_xxx" \
    "https://api.anomalyarmor.ai/api/v1/sdk/freshness/snowflake.prod.warehouse.orders/refresh"
  ```

  ```bash cURL (sync) theme={null}
  curl -X POST -H "Authorization: Bearer aa_live_xxx" \
    "https://api.anomalyarmor.ai/api/v1/sdk/freshness/snowflake.prod.warehouse.orders/refresh?wait=true"
  ```

  ```python Python SDK theme={null}
  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}")
  ```

  ```bash CLI theme={null}
  # Async
  armor freshness refresh snowflake.prod.warehouse.orders

  # Wait for completion
  armor freshness refresh snowflake.prod.warehouse.orders --wait
  ```
</CodeGroup>

### Response (async)

```json theme={null}
{
  "data": {
    "job_id": "job_abc123",
    "status": "pending",
    "asset_id": "snowflake.prod.warehouse.orders"
  }
}
```

### Response (sync with wait=true)

```json theme={null}
{
  "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:

```python theme={null}
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
```

<Tip>
  Use `require_fresh()` in Airflow tasks to automatically fail pipelines when upstream data is stale. See the [Airflow Integration](/integrations/airflow) guide.
</Tip>

## CLI Check Command

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

```bash theme={null}
# 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)

```json theme={null}
{
  "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:

```json theme={null}
{
  "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](/integrations/airflow) 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.