How to Monitor JSON API Freshness in dbt with Custom Assertions

Ever built an incremental JSON‑API pipeline only to discover weeks later that the upstream endpoint stopped emitting new rows, silently corrupting your dashboards? You can stop that guesswork by wiring a dbt test to a tiny Python freshness probe that flags any API lag as a CI failure, guaranteeing that stale data never slips into downstream models.

Key Takeaways

• dbt can run external Python scripts as part of a generic test, letting you enforce freshness SLAs for any HTTP‑based source.
• A lightweight probe that checks the HTTP Date header (or a payload timestamp) and validates JSON shape turns a silent outage into an immediate test failure.
• Integrating the probe’s CSV output with a dbt assertion test gives you version‑controlled, repeatable monitoring that works in CI/CD pipelines.

Data and Sources

The demo uses the public JSONPlaceholder posts endpoint, which reliably returns a list of 100 post objects. The endpoint is always reachable and its schema (userId, id, title, body) is well‑known, making it ideal for a reproducible example.

Data accessed on 2026-11-06.

Monitoring data freshness with dbt tests

dbt ships with source freshness checks for tables, but external APIs sit outside the warehouse. The pattern below bridges that gap:

1. Freshness probe script: A Python file (api_freshness.py) fetches the API, extracts a timestamp, validates the payload, and writes age_seconds to target/api_freshness.csv.
2. dbt generic test: A tiny .sql test reads the CSV via ref('api_freshness') (exposed as a seed) and asserts age_seconds <= {{ var('api_sla_seconds', 86400) }} (default 24 h).
3. CI integration: When dbt test runs in your CI pipeline, a stale API triggers a failing test, aborting the deployment and alerting the team.

Below is the dbt schema.yml snippet that registers the generic test:

version: 2

models:
  - name: api_freshness
    description: "One‑row table containing the most recent freshness probe results."
    columns:
      - name: run_timestamp
        description: "When the probe was executed (UTC)."
      - name: age_seconds
        description: "Seconds elapsed since the API reported its data."

tests:
  - dbt_utils.assert_equal:
      compare_model: ref('api_freshness')
      column_name: age_seconds
      value: "{{ var('api_sla_seconds', 86400) }}"

Run the probe before dbt test (e.g., in a pre‑test hook or as a separate CI step):

python api_freshness.py

Because the probe writes a CSV, dbt treats it as a seed; you only need to add the CSV to seeds/ and run dbt seed before testing.

Limitations and Tradeoffs

• Reliance on HTTP headers: Not all APIs expose a reliable Date header. If the header is missing or inaccurate, you must extract a timestamp from the payload (e.g., last_updated) or from a companion metadata endpoint.
• Network latency vs. true freshness: A high age_seconds can be caused by a slow network rather than stale data. Mitigate by measuring round‑trip time separately or by using a dedicated health‑check endpoint.
• CI runner constraints: Some CI environments block outbound traffic. Ensure the runner has internet access and keep the timeout short (5 seconds in the script) to avoid hanging builds.
• Static demo data: JSONPlaceholder never changes, so the probe will always report a small age (the server time). In production replace the URL with your own API and adjust the SLA accordingly.

Frequently Asked Questions

Can I use this pattern for APIs that return JSON without a Date header?

Yes. Modify the probe to read a last_updated field inside the payload or call a separate metadata endpoint that returns a timestamp.

What happens if the API is temporarily unavailable?

The script treats a timeout or non‑200 response as a hard failure, exiting with a non‑zero status. In CI this fails the test immediately, surfacing the outage.

Is it safe to run the script inside a restricted CI runner?

Make sure the runner permits outbound HTTP and that the script respects a short timeout (5 seconds). The shape validation step prevents malformed JSON from corrupting the downstream seed.

How do I surface the freshness result in my BI tool?

Persist the age_seconds column in a warehouse table (e.g., monitoring.api_freshness) and let dashboards query that table. The dbt test remains the guardrail; the table becomes the reporting source.

Can I combine this with dbt’s built‑in source freshness?

Absolutely. Use dbt’s native source freshness for warehouse tables and the custom generic test for external APIs, giving you a unified freshness strategy across all sources.

Complete Script

The script below is a self‑contained, production‑ready probe. It fetches the JSONPlaceholder posts endpoint, validates the payload, extracts the server’s Date header, computes the age in seconds, and writes a one‑row CSV that dbt can seed.

#!/usr/bin/env python3
"""
api_freshness.py

Fetches a JSON API, validates its schema, extracts the HTTP Date header,
computes how many seconds have elapsed since the server time, and writes
the result to target/api_freshness.csv for dbt consumption.

Failure modes:
  * Timeout or non‑200 response → exit(1)
  * Missing or unparsable Date header → exit(1)
  * Unexpected JSON shape → exit(1)
"""

import csv
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import requests
from email.utils import parsedate_to_datetime

# ----------------------------------------------------------------------
# Configuration
# ----------------------------------------------------------------------
API_URL = "https://jsonplaceholder.typicode.com/posts"
TIMEOUT_SECONDS = 5
OUTPUT_PATH = Path("target") / "api_freshness.csv"

# Expected minimal JSON schema for validation
REQUIRED_KEYS = {"userId", "id", "title", "body"}


def fetch_api(url: str) -> requests.Response:
    """GET the API with a short timeout, raise for HTTP errors."""
    try:
        response = requests.get(url, timeout=TIMEOUT_SECONDS)
        response.raise_for_status()
        return response
    except requests.exceptions.RequestException as exc:
        print(f"Error fetching API: {exc}", file=sys.stderr)
        sys.exit(1)


def validate_json(payload):
    """Ensure payload is a list of dicts containing the required keys."""
    if not isinstance(payload, list):
        print("Invalid payload: expected a list", file=sys.stderr)
        sys.exit(1)

    for i, item in enumerate(payload[:5]):  # check first few items only
        if not isinstance(item, dict):
            print(f"Invalid item at index {i}: not a dict", file=sys.stderr)
            sys.exit(1)
        missing = REQUIRED_KEYS - item.keys()
        if missing:
            print(f"Item at index {i} missing keys: {missing}", file=sys.stderr)
            sys.exit(1)


def extract_server_time(response: requests.Response) -> datetime:
    """Parse the HTTP Date header into a UTC datetime."""
    date_header = response.headers.get("Date")
    if not date_header:
        print("Missing Date header in response", file=sys.stderr)
        sys.exit(1)

    try:
        server_dt = parsedate_to_datetime(date_header)
        # Ensure timezone‑aware UTC
        if server_dt.tzinfo is None:
            server_dt = server_dt.replace(tzinfo=timezone.utc)
        else:
            server_dt = server_dt.astimezone(timezone.utc)
        return server_dt
    except (TypeError, ValueError) as exc:
        print(f"Failed to parse Date header: {exc}", file=sys.stderr)
        sys.exit(1)


def compute_age_seconds(server_time: datetime) -> int:
    """Return the age in seconds between now (UTC) and server_time."""
    now_utc = datetime.now(timezone.utc)
    age = (now_utc - server_time).total_seconds()
    # Guard against negative values caused by clock skew
    return max(int(age), 0)


def write_csv(run_ts: str, age_seconds: int, path: Path):
    """Write a single‑row CSV compatible with dbt seeds."""
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", newline="") as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=["run_timestamp", "age_seconds"])
        writer.writeheader()
        writer.writerow({"run_timestamp": run_ts, "age_seconds": age_seconds})


def main():
    response = fetch_api(API_URL)
    payload = response.json()
    validate_json(payload)

    server_time = extract_server_time(response)
    age_seconds = compute_age_seconds(server_time)

    run_timestamp = datetime.now(timezone.utc).isoformat()
    write_csv(run_timestamp, age_seconds, OUTPUT_PATH)

    print(f"Freshness probe succeeded: age_seconds={age_seconds}")
    # Exit with 0 for CI success; dbt test will decide pass/fail based on SLA


if __name__ == "__main__":
    main()

Recommendation and Next Step

Implement the probe as part of your data‑ingestion CI pipeline and pair it with a dbt generic test that enforces your SLA (e.g., 86 400 seconds for a daily refresh). This gives you immediate feedback when an upstream API stalls, protecting downstream analytics without manual monitoring.

Next step: add the script to your repository, schedule it to run before dbt test in your CI workflow, and adjust api_sla_seconds via a dbt variable to match your business requirements.