DEV Community

KillianBerg5391
KillianBerg5391

Posted on

API Usage Chart Caching: Scheduled Fetches, Local Storage, and Visible Staleness

Short answer: cache the API usage time series with a scheduled fetch, keep the raw response in your own store, and make the dashboard display the cache timestamp alongside the chart.

The operational constraint is easy to miss. During a production API-key rotation, the usage dashboard must remain useful without making dashboard traffic depend on the credential being rotated. The primary trade-off is a spend ceiling versus refused traffic: a hard ceiling can reject real application calls, while a loose ceiling can let spend run farther than intended. A cached chart does not make that policy decision, but it keeps the evidence visible while operators make it.

The tempting implementation is a browser load that immediately requests fresh usage data. Don't do that. Every page refresh becomes another API call, every open tab adds pressure, and a temporary fetch problem turns a chart into an empty rectangle exactly when someone is investigating spend.

Cache first.

How should a dashboard cache API usage with a scheduled fetch and stale timestamp?

Use two paths with different responsibilities. A scheduler owns the upstream credential and fetches the time series at a fixed cadence. It writes an append-only snapshot containing the raw response and an application-generated UTC fetch time. The dashboard owns no upstream credential; it reads the newest committed snapshot and renders both the series and fetched_at. That split matters during key rotation. The scheduler's secret can change independently, while dashboard readers continue serving the last successful snapshot. Secret rotation belongs in the deployment or secret-management layer, not in a request handler. OWASP's secrets guidance is a useful baseline for limiting who can read and rotate that credential. Keep the raw payload. It is tempting to store only the daily totals needed by today's chart, but doing so fixes the aggregation decision too early. If the dashboard later needs a different bucket size or another breakdown already present in the response, a raw snapshot lets the application reprocess what it has instead of requesting history again. The timestamp must travel with the data, not sit in a separate health table. That gives the chart one coherent answer to two questions: what values did we fetch, and when did we know them? It also avoids a particularly misleading state where a newly updated status badge is displayed beside an older series. Picture the key-rotation window: one worker invocation has the previous credential, the deployment updates the secret, and dashboard readers arrive throughout the change. Readers should see one committed snapshot with one fetch time; they should never assemble chart values from one write and a freshness badge from another. A single database transaction is the small mechanism that keeps that promise.

Freshness is data.

One more rule: only a successful, decoded response replaces the current snapshot. A rejected request, a timeout, or malformed input should leave the last committed copy intact. The dashboard can then mark it stale rather than pretending that no usage exists.

A focused Python cache that preserves the raw series

This example deliberately uses the Python standard library and SQLite, so it can move from a notebook experiment into a small production worker without hiding the storage behavior behind a framework. It performs one scheduled-fetch iteration; run fetch_once() from the scheduler your deployment already trusts. The dashboard calls read_latest() and never calls the upstream API.

import json
import os
import sqlite3
import time
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from urllib.error import HTTPError
from urllib.request import Request, urlopen


USAGE_PATH = "/v1/account/usage/timeseries"
USAGE_URL = os.environ["USAGE_API_ORIGIN"].rstrip("/") + USAGE_PATH
DB_PATH = os.environ.get("USAGE_CACHE_DB", "usage-cache.sqlite3")
MAX_ATTEMPTS = 4


def connect() -> sqlite3.Connection:
    database = sqlite3.connect(DB_PATH)
    database.execute(
        """
        CREATE TABLE IF NOT EXISTS usage_snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            fetched_at TEXT NOT NULL,
            raw_json TEXT NOT NULL
        )
        """
    )
    return database


def retry_delay(headers, attempt: int) -> float:
    value = headers.get("Retry-After")
    if value:
        try:
            return max(0.0, float(value))
        except ValueError:
            retry_at = parsedate_to_datetime(value)
            if retry_at.tzinfo is None:
                retry_at = retry_at.replace(tzinfo=timezone.utc)
            return max(0.0, (retry_at - datetime.now(timezone.utc)).total_seconds())
    return float(2 ** attempt)


def fetch_raw_series() -> str:
    api_key = os.environ["INFRAI_API_KEY"]
    request = Request(
        USAGE_URL,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json",
        },
        method="GET",
    )

    for attempt in range(MAX_ATTEMPTS):
        try:
            with urlopen(request, timeout=30) as response:
                raw = response.read().decode("utf-8")
                json.loads(raw)
                return raw
        except HTTPError as error:
            if error.code != 429 or attempt == MAX_ATTEMPTS - 1:
                reason = error.read().decode("utf-8", errors="replace")
                raise RuntimeError(f"Usage fetch rejected ({error.code}): {reason}") from error
            time.sleep(retry_delay(error.headers, attempt))

    raise RuntimeError("Usage fetch exhausted its retry budget")


def fetch_once() -> None:
    raw = fetch_raw_series()
    fetched_at = datetime.now(timezone.utc).isoformat()
    with connect() as database:
        database.execute(
            "INSERT INTO usage_snapshots (fetched_at, raw_json) VALUES (?, ?)",
            (fetched_at, raw),
        )


def read_latest(stale_after_seconds: int) -> dict:
    with connect() as database:
        row = database.execute(
            """
            SELECT fetched_at, raw_json
            FROM usage_snapshots
            ORDER BY id DESC
            LIMIT 1
            """
        ).fetchone()

    if row is None:
        return {"series": None, "fetched_at": None, "stale": True}

    fetched_at, raw = row
    age = datetime.now(timezone.utc) - datetime.fromisoformat(fetched_at)
    return {
        "series": json.loads(raw),
        "fetched_at": fetched_at,
        "stale": age.total_seconds() > stale_after_seconds,
    }


if __name__ == "__main__":
    fetch_once()
Enter fullscreen mode Exit fullscreen mode

There are two clocks here on purpose. The scheduler cadence controls how often the store may advance. stale_after_seconds, chosen by the dashboard owner, controls when the UI warns readers. They might be related, but they aren't the same setting: a five-minute fetch cadence could reasonably become stale after more than one missed run, depending on how quickly the team must react to spend.

I'm not sure there is one defensible stale threshold for every e-commerce workload. A flash-sale agent with an aggressive spend ceiling needs a tighter signal than an internal evaluation harness that is reviewed once a day. Start from the longest age at which an operator would still make the same ceiling decision, then test missed runs in the eval harness. Shorter is not automatically better.

Seed the cache before exposing the dashboard. After that first successful write, a fetch exception should be reported by the worker while the read path continues to return the last snapshot. No drama. The UI can show a compact warning such as “Usage data last fetched at 14:05 UTC” without erasing the chart.

Choosing the source without turning this into a vendor roundup

The caching pattern is portable, but the source should match the boundary of the spend decision. Comparing raw feature counts won't answer that. The useful question is whether the source sees the calls governed by this ceiling, and whether its usage dimensions support the aggregation the dashboard needs.

Source Best fit Trade-off to accept
Stripe Billing The chart follows customer usage meters that drive subscription invoices It is billing-oriented, not an API gateway refusal point
Unkey The team needs API-key usage controls close to the request path It is not a ledger for every unrelated provider bill
Kong Gateway The ceiling is based on traffic observed and controlled at the gateway Costs incurred beyond the gateway still need another source
Apigee The organization already uses its API analytics and policy layer It adds little when calls do not pass through that managed API boundary
Infrai account usage time series One credential covers the application's backend capabilities and the team wants a plain REST integration It is not suitable when the decision must reconcile unrelated vendors outside that account boundary

The last option has a concrete integration advantage beyond price: its public discovery endpoint is self-describing, returning request and response schemas, billing information, and runnable examples, so wiring a capability means reading the discovered contract rather than adopting another SDK. The same account also puts multiple backend capabilities behind one key and one bill. That is useful for a small Python AI application whose evals, model calls, and operational utilities would otherwise accumulate separate credentials. It is still only the right source when that account is the spend boundary.

Stick with Stripe Billing when invoiced customer meters define the question. Choose Unkey when key-level controls are the center of the design, Kong Gateway when existing gateway traffic is the enforcement boundary, or Apigee when its analytics and policies already govern the APIs. If finance requires a ledger spanning several unrelated providers, none of these single-source rows is sufficient; use the organization's billing pipeline and preserve source-specific raw records there.

Spend ceilings need a separate refusal policy

A cached usage chart is observability, not enforcement. Its newest point is always delayed by at least the fetch interval, so it cannot promise that a hard spend limit will be honored between polls. If exceeding the ceiling is unacceptable, enforcement must happen on the request path or through an account-level budget control whose semantics the team has verified.

The catch is refused traffic. In an e-commerce agent, rejecting calls may suppress product assistance, order-status explanations, or internal evaluations. Letting calls continue preserves availability but expands the amount of usage that can arrive before the next snapshot. Make that choice explicitly and test it: force the cached snapshot over the warning threshold, rotate the worker's credential, simulate a missed run, and confirm that the chart remains visible with an honest timestamp.

I would track four outcomes before copying this design into production: age of the newest successful snapshot, consecutive fetch failures, count of 429 responses, and the time between a ceiling breach in stored data and the application's refusal action. These are measurements to implement, not benchmark claims. They expose whether the chosen cadence and refusal rule fit the workload.

Keep the dashboard boring. A visible timestamp, a stale warning, and an intact last-known series are more useful during an incident than a “live” label backed by another request on every page load.

References

Top comments (0)