Tracking API Usage and Analytics for Commercial Python APIs

If you sell API access and cannot answer "what did customer 4471 cost us last month?" in under ten seconds, you are not running a business, you are running a hobby with a Stripe account attached. This guide builds the measurement layer that turns raw traffic into per-key usage events, live quota counters, aggregation windows, and an invoice line item. Part of the Building & Monetizing API-Driven Micro-SaaS guide.

The pattern here is deliberately boring: capture one immutable event per request, buffer it in memory, flush it in batches to Postgres, keep a separate fast counter in Redis for quota decisions, and roll the raw events into progressively coarser windows on a schedule. Every piece is replaceable. What is not negotiable is that billing reads from a durable event log rather than from a counter you can lose, because a counter that vanishes when a pod restarts will eventually undercharge your largest customer and you will not notice for a quarter.

Prerequisites

Assume Python 3.11 or newer, an async FastAPI application already serving authenticated traffic, and a working API key authentication layer that resolves a request to a customer identity before the route handler runs. Usage tracking without stable key identity is just a web server log.

Bash
python -m pip install \
  "fastapi>=0.115,<1.0" \
  "uvicorn[standard]>=0.32" \
  "sqlalchemy[asyncio]>=2.0.36" \
  "asyncpg>=0.30" \
  "redis>=5.2" \
  "stripe>=11.1" \
  "pydantic>=2.9" \
  "httpx>=0.28"

Set these before you start. Every snippet below reads configuration through os.getenv so the same image runs in staging and production without edits.

Bash
export USAGE_DATABASE_URL="postgresql+asyncpg://api:secret@db.internal:5432/apiprod"
export USAGE_REDIS_URL="redis://cache.internal:6379/2"
export USAGE_FLUSH_INTERVAL_SECONDS="2.0"
export USAGE_FLUSH_MAX_BATCH="500"
export USAGE_QUEUE_MAXSIZE="20000"
export USAGE_QUOTA_ENFORCED="true"
export STRIPE_API_KEY="sk_live_..."
export STRIPE_METER_EVENT_NAME="api_requests"

You also need a Postgres 14+ instance you control. Managed Postgres on Render, Fly, or Neon is fine; the rollup queries below use plain SQL with no extensions, so TimescaleDB is optional rather than assumed.

Step 1: Model the usage event as an immutable fact

Design the event row first, because everything downstream inherits its mistakes. A usage event records what happened, not what you decided to charge. Keep pricing out of it. If you bake amount_cents into the event and later change your rate card, you can no longer recompute history, and recomputing history is exactly what you will need the first time a customer disputes an invoice.

Store six things per request: the API key hash, the customer id, the route template, the HTTP status, the duration, and the billable unit count. The route template matters more than the raw path — /v1/documents/8fa1/pages is useless for aggregation, while /v1/documents/{doc_id}/pages groups cleanly. FastAPI exposes the template on request.scope["route"].path, so grab it after the handler resolves.

Python
# usage/models.py
import os
from datetime import datetime

from sqlalchemy import BigInteger, Index, Integer, SmallInteger, String, text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class UsageEvent(Base):
    __tablename__ = "usage_event"
    __table_args__ = (
        Index("ix_usage_customer_time", "customer_id", "occurred_at"),
        Index("ix_usage_route_time", "route", "occurred_at"),
        {"postgresql_partition_by": "RANGE (occurred_at)"},
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    occurred_at: Mapped[datetime] = mapped_column(
        server_default=text("now()"), primary_key=True, index=True
    )
    customer_id: Mapped[str] = mapped_column(String(64))
    api_key_id: Mapped[str] = mapped_column(String(64))
    route: Mapped[str] = mapped_column(String(160))
    method: Mapped[str] = mapped_column(String(8))
    status_code: Mapped[int] = mapped_column(SmallInteger)
    duration_ms: Mapped[int] = mapped_column(Integer)
    billable_units: Mapped[int] = mapped_column(Integer, server_default=text("1"))


DATABASE_URL = os.getenv("USAGE_DATABASE_URL", "postgresql+asyncpg://localhost/api")

Note billable_units. Not every request is worth the same. A document extraction that processes 40 pages should write billable_units=40 while a health check writes zero. Decide this inside the handler and stash it on request.state so the middleware can read it on the way out. That one integer is the difference between charging per call and charging per unit of value delivered, and it is the single highest-leverage field in the whole schema.

Partition the table by month from day one. RANGE (occurred_at) with monthly child tables makes dropping expired raw data a metadata operation instead of a multi-hour DELETE that bloats your write-ahead log. The detailed DDL, partition automation, and index tuning live in logging API usage events to Postgres.

Usage capture pipeline from request to Postgres A request passes through usage middleware, which increments a Redis quota counter on the hot path and pushes an event onto an in-process queue that a batch flusher drains into a partitioned Postgres table. Client + API key Bearer token Usage middleware adds ~0.3 ms Route handler sets units Response Redis quota counter blocking, decides 429 In-process queue put_nowait, never awaits Batch flusher task 2 s or 500 rows usage_event partitioned monthly source of truth: enforcement source of truth: billing

Step 2: Capture events in middleware without slowing the hot path

The cardinal rule of usage instrumentation: never write to your primary database inside the request cycle. A single INSERT per request costs a round trip, a connection from the pool, and a WAL flush. At 300 requests per second that is 300 extra transactions per second on the same Postgres instance serving your product, and it will exhaust your pool long before it exhausts your CPU.

Instead, push a tuple onto an asyncio.Queue with put_nowait and return immediately. put_nowait raises QueueFull rather than blocking, which is precisely the behaviour you want: under catastrophic backpressure you drop analytics rather than drop revenue-generating requests. Count the drops in a metric so you know when the buffer is undersized.

Python
# usage/middleware.py
import asyncio
import os
import time
from datetime import datetime, timezone

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

QUEUE_MAXSIZE = int(os.getenv("USAGE_QUEUE_MAXSIZE", "20000"))
usage_queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAXSIZE)
dropped_events = 0


class UsageMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        started = time.perf_counter()
        response = await call_next(request)
        elapsed_ms = int((time.perf_counter() - started) * 1000)

        principal = getattr(request.state, "principal", None)
        if principal is None:
            return response  # unauthenticated traffic is not billable

        route = request.scope.get("route")
        event = {
            "occurred_at": datetime.now(timezone.utc),
            "customer_id": principal.customer_id,
            "api_key_id": principal.api_key_id,
            "route": getattr(route, "path", request.url.path),
            "method": request.method,
            "status_code": response.status_code,
            "duration_ms": elapsed_ms,
            "billable_units": getattr(request.state, "billable_units", 1),
        }

        global dropped_events
        try:
            usage_queue.put_nowait(event)
        except asyncio.QueueFull:
            dropped_events += 1
        return response

Two details earn their keep. First, the middleware records billable_units from request.state, so a handler that processed 40 pages simply sets request.state.billable_units = 40 and the accounting follows. Second, unauthenticated requests exit early — you do not want to bill anyone for the 401s generated by a misconfigured client, and you certainly do not want an unauthenticated flood filling the buffer. Pair this with structured logging so that the same request id appears in both your logs and your usage rows; correlating them during a billing dispute takes minutes instead of days.

Step 3: Flush batches with a background drain task

The flusher is a single long-lived task started in the lifespan handler. It drains up to USAGE_FLUSH_MAX_BATCH events or waits USAGE_FLUSH_INTERVAL_SECONDS, whichever comes first, then writes them in one multi-row insert. Batching 500 rows into a single statement turns 500 transactions into one, which on a small managed Postgres is the difference between 8% and 0.4% of your write capacity spent on telemetry.

Python
# usage/flusher.py
import asyncio
import contextlib
import os

from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from usage.models import DATABASE_URL, UsageEvent
from usage.middleware import usage_queue

FLUSH_INTERVAL = float(os.getenv("USAGE_FLUSH_INTERVAL_SECONDS", "2.0"))
MAX_BATCH = int(os.getenv("USAGE_FLUSH_MAX_BATCH", "500"))

engine = create_async_engine(DATABASE_URL, pool_size=5, max_overflow=2)
Session = async_sessionmaker(engine, expire_on_commit=False)


async def _collect_batch() -> list[dict]:
    first = await usage_queue.get()
    batch = [first]
    deadline = asyncio.get_running_loop().time() + FLUSH_INTERVAL
    while len(batch) < MAX_BATCH:
        remaining = deadline - asyncio.get_running_loop().time()
        if remaining <= 0:
            break
        try:
            batch.append(await asyncio.wait_for(usage_queue.get(), remaining))
        except TimeoutError:
            break
    return batch


async def flush_loop() -> None:
    while True:
        batch = await _collect_batch()
        try:
            async with Session() as session:
                await session.execute(insert(UsageEvent), batch)
                await session.commit()
        except Exception:
            # Re-queue what fits; a persistent DB outage must not kill the task.
            for event in batch:
                with contextlib.suppress(asyncio.QueueFull):
                    usage_queue.put_nowait(event)
            await asyncio.sleep(FLUSH_INTERVAL)

Wire it into the application lifespan and, critically, drain the queue on shutdown. A rolling deploy that kills workers mid-buffer silently discards up to two seconds of billable events per worker. Multiply that by ten deploys a day and thirty workers and you are throwing away real money.

Python
# app.py
import asyncio
from contextlib import asynccontextmanager

from fastapi import FastAPI

from usage.flusher import flush_loop, usage_queue
from usage.middleware import UsageMiddleware


@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(flush_loop())
    yield
    await usage_queue.join() if usage_queue.qsize() else None
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass


app = FastAPI(lifespan=lifespan)
app.add_middleware(UsageMiddleware)

The engine here uses its own small pool rather than sharing the request pool. If the flusher and your handlers compete for the same twenty connections, a slow batch insert starves user traffic. The reasoning behind separate engines and pool sizing is covered in async database access with SQLAlchemy.

Step 4: Enforce quotas with atomic Redis counters

Quota enforcement cannot wait for the flusher. A customer on a 50,000-call plan who buys 60,000 calls in a burst must be stopped at call 50,001, not two seconds later. Keep a separate counter in Redis, incremented on the hot path, and treat it as a cache of the durable event log rather than as the log itself.

Do the increment and the expiry in one Lua script. The naive INCR followed by EXPIRE has a real race: if the process dies between the two commands, the key never expires and the customer is permanently locked out of a billing period that ended weeks ago. One script, one round trip, no window.

Python
# usage/quota.py
import os
from datetime import datetime, timezone

import redis.asyncio as aioredis

REDIS_URL = os.getenv("USAGE_REDIS_URL", "redis://localhost:6379/0")
QUOTA_ENFORCED = os.getenv("USAGE_QUOTA_ENFORCED", "true").lower() == "true"

_INCR_WITH_TTL = """
local current = redis.call('INCRBY', KEYS[1], tonumber(ARGV[1]))
if current == tonumber(ARGV[1]) then
  redis.call('EXPIRE', KEYS[1], tonumber(ARGV[2]))
end
return current
"""

redis_client = aioredis.from_url(REDIS_URL, decode_responses=True)
_script = redis_client.register_script(_INCR_WITH_TTL)

# 40 days covers a full month plus late reconciliation.
KEY_TTL_SECONDS = int(os.getenv("USAGE_QUOTA_TTL_SECONDS", str(40 * 86400)))


def period_key(customer_id: str, now: datetime | None = None) -> str:
    now = now or datetime.now(timezone.utc)
    return f"usage:{customer_id}:{now:%Y-%m}"


async def consume(customer_id: str, units: int, limit: int) -> tuple[bool, int]:
    """Return (allowed, new_total). Fails open if Redis is unreachable."""
    key = period_key(customer_id)
    try:
        total = int(await _script(keys=[key], args=[units, KEY_TTL_SECONDS]))
    except aioredis.RedisError:
        return True, -1
    if not QUOTA_ENFORCED:
        return True, total
    return total <= limit, total

Fail open, not closed. If Redis blinks, serving a few thousand over-quota calls costs you a rounding error; returning 429 to every paying customer costs you the account. The durable event log still records everything, so you can bill for the overage later even though you did not block it in the moment. That asymmetry is the whole argument for keeping enforcement and billing on separate substrates. For the harder abuse cases — burner emails, key sharing, distributed free-tier farming — see preventing free-tier abuse, and for the response-shaping side of limits, rate limiting best practices.

Quota check sequence for an over-limit request Sequence showing the client calling the middleware, the middleware running an atomic Redis increment, and branching to either a 429 response or the route handler depending on the returned counter. Client Middleware Redis Handler POST /v1/extract EVALSHA incr+expire total = 50001 branch on total vs plan limit 429 + Retry-After under limit: await call_next 200 + billable_units

Step 5: Roll raw events into aggregation windows

Raw events answer every question but answer them slowly. Once you hold 200 million rows, a dashboard query scanning a customer's month takes seconds and your dashboard becomes the heaviest workload on the database. Fix that with a two-level rollup: an hourly table written by a scheduled job, and a daily table derived from the hourly one.

Hourly granularity is the right first cut. It is coarse enough to shrink 3.6 million rows into a few thousand, and fine enough to show a customer the traffic spike that blew their quota at 14:00. Anything finer and you have not saved much; anything coarser and support tickets become guesswork.

Sql
-- migrations/usage_rollup.sql
CREATE TABLE IF NOT EXISTS usage_hourly (
  bucket        timestamptz  NOT NULL,
  customer_id   text         NOT NULL,
  route         text         NOT NULL,
  calls         bigint       NOT NULL,
  billable      bigint       NOT NULL,
  errors        bigint       NOT NULL,
  duration_sum  bigint       NOT NULL,
  PRIMARY KEY (bucket, customer_id, route)
);

INSERT INTO usage_hourly
SELECT date_trunc('hour', occurred_at) AS bucket,
       customer_id,
       route,
       count(*)                                        AS calls,
       sum(billable_units)                             AS billable,
       count(*) FILTER (WHERE status_code >= 500)      AS errors,
       sum(duration_ms)                                AS duration_sum
FROM usage_event
WHERE occurred_at >= $1 AND occurred_at < $2
GROUP BY 1, 2, 3
ON CONFLICT (bucket, customer_id, route) DO UPDATE
SET calls = EXCLUDED.calls,
    billable = EXCLUDED.billable,
    errors = EXCLUDED.errors,
    duration_sum = EXCLUDED.duration_sum;

The ON CONFLICT ... DO UPDATE makes the job idempotent, which matters because you will re-run it. Late events arrive: a flusher that retried through a database outage writes rows with an occurred_at from twenty minutes ago. Always re-aggregate a trailing window rather than only the last completed hour.

Python
# usage/rollup.py
import asyncio
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path

from sqlalchemy import text

from usage.flusher import Session

LOOKBACK_HOURS = int(os.getenv("USAGE_ROLLUP_LOOKBACK_HOURS", "3"))
SQL_PATH = Path(os.getenv("USAGE_ROLLUP_SQL", "migrations/usage_rollup.sql"))


async def rebuild_recent_hours() -> int:
    now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
    start = now - timedelta(hours=LOOKBACK_HOURS)
    statement = SQL_PATH.read_text().split(";", 1)[1]
    async with Session() as session:
        result = await session.execute(
            text(statement.replace("$1", ":start").replace("$2", ":end")),
            {"start": start, "end": now + timedelta(hours=1)},
        )
        await session.commit()
    return result.rowcount


if __name__ == "__main__":
    print(asyncio.run(rebuild_recent_hours()), "buckets rebuilt")

Run it every ten minutes from cron, or as a periodic task if you already operate Celery for background jobs. Then set retention: drop raw partitions after 14 days, keep hourly for 13 months, keep daily forever. Thirteen months lets a customer compare this July to last July, which is the only comparison enterprise buyers ever ask for.

Aggregation windows and retention horizons Three retention bars showing raw events kept 14 days, hourly rollups kept 13 months, and daily rollups kept indefinitely, with row counts shrinking at each level. Retention horizon for one customer at 3.6M calls / month Raw events 14 days 1.7M rows, dropped by partition Hourly rollup 13 months ~57k rows Daily rollup kept indefinitely, ~2.4k rows now 6 months ago 3 years ago Rebuild a 3-hour trailing window every run so late-flushed events land correctly.

Step 6: Compute revenue and margin per endpoint

This is the step most builders skip, and it is the one that changes decisions. Join your rollups against a cost model and a rate card, and you get gross margin per route. Almost every API business discovers at least one endpoint that loses money on every call — usually something that fans out to a paid third-party service or burns CPU on rendering.

Python
# usage/margin.py
import asyncio
import os
import tomllib
from dataclasses import dataclass
from pathlib import Path

from sqlalchemy import text

from usage.flusher import Session

RATE_CARD = Path(os.getenv("USAGE_RATE_CARD", "config/rates.toml"))

MARGIN_SQL = """
SELECT route,
       sum(calls)    AS calls,
       sum(billable) AS billable,
       sum(duration_sum)::float / 1000 AS compute_seconds
FROM usage_hourly
WHERE bucket >= date_trunc('month', now())
GROUP BY route
ORDER BY calls DESC
"""


@dataclass(frozen=True)
class RouteMargin:
    route: str
    revenue: float
    cost: float

    @property
    def margin_pct(self) -> float:
        return 0.0 if not self.revenue else (self.revenue - self.cost) / self.revenue * 100


async def margin_by_route() -> list[RouteMargin]:
    rates = tomllib.loads(RATE_CARD.read_text())
    vcpu_second = float(rates["infra"]["usd_per_vcpu_second"])
    async with Session() as session:
        rows = (await session.execute(text(MARGIN_SQL))).all()

    out: list[RouteMargin] = []
    for route, calls, billable, compute_seconds in rows:
        card = rates["routes"].get(route, {})
        revenue = billable * float(card.get("usd_per_unit", 0.0))
        cost = compute_seconds * vcpu_second + calls * float(card.get("usd_vendor_per_call", 0.0))
        out.append(RouteMargin(route=route, revenue=revenue, cost=cost))
    return sorted(out, key=lambda m: m.margin_pct)


if __name__ == "__main__":
    for m in asyncio.run(margin_by_route()):
        print(f"{m.route:<34} rev ${m.revenue:>9,.2f}  cost ${m.cost:>9,.2f}  {m.margin_pct:>6.1f}%")

Sorting ascending by margin puts your worst endpoint first, which is exactly what you want to see every Monday morning. On one month of real numbers from a document API, the output looked like this: POST /v1/extract served 2.10M billable units for $2,940 in revenue against $882 of cost — a healthy 70% margin. POST /v1/render served 180,000 calls for $252 of revenue against $558 of cost, because each render fanned out to a paid headless browser service. That endpoint was losing 121% of its revenue on every call and had been shipped as a "small convenience feature". The methodology for building the cost side of that equation properly is in calculating cost per API request.

Monthly revenue against cost for three endpoints Paired bars comparing revenue and cost per endpoint: extract earns 2940 dollars against 882 of cost, status earns nothing against 64 of cost, and render earns 252 against 558 of cost. Revenue vs cost by endpoint, one month (USD) revenue cost /v1/extract 2940 882 /v1/status 0 (unbilled) 64 /v1/render 252 558 /v1/render burns 2.2x its revenue: reprice it, cache it, or retire it.

Three options exist for a loss-making route, and only three: raise its price, cut its cost, or remove it. Delaying the decision is the same as choosing to subsidise it. Once you have this table, the arguments in designing API pricing tiers stop being theoretical, because you know exactly which unit you should be charging for.

Step 7: Report usage into Stripe billing

The final hop pushes aggregated usage into your billing provider. Report once per hour from the durable rollup, never from the request path, and carry a deterministic idempotency key so a retried job cannot double-bill. Stripe's meter events accept an identifier; build it from customer, meter, and bucket so the same hour can be submitted a hundred times and land once.

Python
# usage/billing_sync.py
import asyncio
import os
from datetime import datetime, timezone

import stripe
from sqlalchemy import text

from usage.flusher import Session

stripe.api_key = os.getenv("STRIPE_API_KEY", "")
METER_EVENT_NAME = os.getenv("STRIPE_METER_EVENT_NAME", "api_requests")

PENDING_SQL = """
SELECT h.bucket, h.customer_id, c.stripe_customer_id, sum(h.billable) AS units
FROM usage_hourly h
JOIN customer c ON c.id = h.customer_id
WHERE h.bucket >= :start AND h.bucket < :end AND c.stripe_customer_id IS NOT NULL
GROUP BY h.bucket, h.customer_id, c.stripe_customer_id
HAVING sum(h.billable) > 0
"""


async def report_hour(start: datetime, end: datetime) -> int:
    async with Session() as session:
        rows = (await session.execute(text(PENDING_SQL), {"start": start, "end": end})).all()

    sent = 0
    for bucket, customer_id, stripe_customer_id, units in rows:
        identifier = f"{METER_EVENT_NAME}:{customer_id}:{bucket:%Y%m%dT%H}"
        await asyncio.to_thread(
            stripe.billing.MeterEvent.create,
            event_name=METER_EVENT_NAME,
            identifier=identifier,
            timestamp=int(bucket.replace(tzinfo=timezone.utc).timestamp()),
            payload={"stripe_customer_id": stripe_customer_id, "value": str(int(units))},
        )
        sent += 1
    return sent

asyncio.to_thread keeps the synchronous Stripe SDK from blocking the event loop. It is not elegant, but it is correct and it ships today; the alternative is hand-rolling the REST calls with httpx and re-implementing Stripe's retry semantics for no commercial gain. Meter configuration, aggregation formulas, and the subscription wiring are covered in Stripe metered billing configuration, and the broader payment integration in integrating Stripe with Python APIs.

Reconcile monthly. Before the invoice finalises, compare your usage_hourly sum against what Stripe recorded for the period. A drift of more than 0.1% means events were lost or double-sent, and you want to find that on the 28th, not in a support thread on the 3rd.

Lifecycle of a usage event from buffer to invoice State machine showing an event moving from buffered to persisted to aggregated to reported to invoiced, with a retry loop on flush failure and a backfill edge from reconciliation. buffered persisted aggregated reported invoiced reconcile drift backfill may be dropped idempotent rebuild keyed identifier Only the persisted state is authoritative for money.

Configuration reference

Env varDefaultProduction recommendation
USAGE_DATABASE_URLnoneDedicated asyncpg URL, separate pool
USAGE_REDIS_URLredis://localhost:6379/0Own logical db, no eviction policy
USAGE_FLUSH_INTERVAL_SECONDS2.01.0–2.0; lower loses less on crash
USAGE_FLUSH_MAX_BATCH500500–1000 rows per insert
USAGE_QUEUE_MAXSIZE2000030x peak RPS per worker
USAGE_QUOTA_ENFORCEDtruetrue in prod, false in staging
USAGE_QUOTA_TTL_SECONDS345600040 days, covers late reconciliation
USAGE_ROLLUP_LOOKBACK_HOURS33; raise if flush retries run long
STRIPE_METER_EVENT_NAMEapi_requestsMatch the meter you created in Stripe

Set USAGE_QUEUE_MAXSIZE from measured traffic, not intuition. At 200 requests per second per worker with a 2-second flush interval, steady state holds 400 events; a 20,000 slot buffer absorbs a 100-second database outage before it starts dropping. Redis must never run with an LRU eviction policy on the quota database, because evicting a counter silently resets a customer's consumption to zero mid-month.

Gotchas and failure modes

Counting requests when you should count units. Symptom: revenue stays flat while infrastructure cost climbs, because heavy customers batch 200 items into one call. Fix: set request.state.billable_units in the handler based on actual work done, and price the unit rather than the call. Ship this before you have customers; changing the billable unit after launch requires grandfathering every existing plan.

Dropping the buffer on deploy. Symptom: usage totals dip predictably at deploy time, always slightly under. Fix: drain the queue in the lifespan shutdown path and give your platform a terminationGracePeriod longer than the flush interval. On Render or Fly this is a single setting, and skipping it costs a few tenths of a percent of revenue per deploy — small until you deploy forty times a month.

Trusting Redis for billing. Symptom: a Redis failover, a FLUSHDB in the wrong shell, or a maxmemory eviction, and a customer's month reads zero. Fix: Redis decides enforcement in real time, Postgres decides money. Rehydrate the counter from usage_hourly on a cache miss instead of assuming zero, otherwise a restart hands every customer a fresh quota.

Unbounded route cardinality. Symptom: usage_hourly grows as fast as usage_event and the rollup stops helping. Fix: always aggregate on the route template, and if the template is missing — a 404, a proxy passthrough — write the literal string unmatched. One bad path parameter leaking into the route column can add millions of distinct values in a week.

Timezone drift at the period boundary. Symptom: a customer's usage looks 4% high or low in the first days of a month, and disputes pile up on the 1st. Fix: store and bucket everything in UTC, then convert only at the presentation layer. If your Stripe subscription anchors to a non-UTC billing cycle, align the rollup query to the subscription's current_period_start rather than to date_trunc('month').

Verification

Prove the pipeline end to end before you point billing at it. Make an authenticated call, then confirm the event landed and the counter moved.

Bash
curl -s -o /dev/null -w '%{http_code}\n' \
  -H "Authorization: Bearer $TEST_API_KEY" \
  "$API_BASE_URL/v1/extract" -d '{"pages": 3}' -H 'content-type: application/json'

sleep 3
psql "$PGURL" -c \
  "SELECT route, billable_units, status_code FROM usage_event ORDER BY occurred_at DESC LIMIT 1;"

redis-cli -u "$USAGE_REDIS_URL" GET "usage:cus_test123:$(date -u +%Y-%m)"

You should see 200, a single row with route = /v1/extract and billable_units = 3, and a Redis value that increased by exactly 3. If the row is absent after three seconds, the flusher task is not running — check that lifespan is attached to the app object you actually serve.

Python
# tests/test_usage_pipeline.py
import asyncio

import pytest
from httpx import ASGITransport, AsyncClient

from app import app
from usage.middleware import usage_queue


@pytest.mark.anyio
async def test_request_enqueues_billable_units():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.post(
            "/v1/extract", json={"pages": 3}, headers={"Authorization": "Bearer test"}
        )
    assert response.status_code == 200
    event = await asyncio.wait_for(usage_queue.get(), timeout=1.0)
    assert event["route"] == "/v1/extract"
    assert event["billable_units"] == 3

Add one more test that asserts the quota counter returns False on the call that crosses the limit. Enforcement bugs are silent in staging and expensive in production, so pin the boundary with an explicit assertion.

Cost and performance at scale

The measurable overhead of this design is small and predictable. The middleware adds roughly 0.3 ms per request: one Redis round trip on a local network plus a dictionary construction and a non-blocking queue put. At 10 million requests a month that is about 50 minutes of aggregate CPU, which on a $25 Render instance is well under a dollar. Storage dominates instead. A usage_event row costs roughly 90 bytes with indexes, so 10 million requests produce about 900 MB per month of raw data — real money on managed Postgres, which is exactly why the 14-day raw retention and the hourly rollup exist. After rollup, the same 10 million requests compress to a few hundred thousand hourly rows, under 40 MB, and the daily table is noise. Redis holds one small integer key per customer per month, so even 5,000 customers fit in a few megabytes on the cheapest instance available. The failure mode that actually costs money is not compute, it is a rollup job left un-indexed: without ix_usage_customer_time, the hourly aggregation degrades to a sequential scan over the current partition and starts competing with production traffic for I/O. Add the index, keep the trailing window to three hours, and the job finishes in under two seconds on a hundred million rows. Once the numbers are landing reliably, expose them to customers — a self-serve view of consumption cuts billing support tickets sharply, and building a customer usage dashboard walks through the queries and caching that make it fast.

FAQ

How much does usage tracking cost to run at 10 million requests a month? Under five dollars if you follow the retention plan above. Compute overhead is roughly 50 minutes of aggregate CPU across the month, storage peaks at about 900 MB of raw events before the 14-day partition drop, and Redis needs only a few megabytes. Skip the rollups and keep raw events forever and that same volume grows past 10 GB a year, at which point managed Postgres storage becomes the largest line on the bill.

Should I bill from Redis counters or from the Postgres event log? Always from Postgres. Redis is a real-time enforcement cache that can be evicted, failed over, or flushed by accident, and none of those events should change what a customer owes. The event log is append-only and replayable, so if you change your rate card in October you can recompute September for a refund. Use Redis to answer "block this request now" and Postgres to answer "what is the invoice".

Does per-request tracking hurt my p99 latency enough to lose customers? No, if you keep the database off the request path. The measured cost is one Redis round trip plus a queue put — around 0.3 ms, which disappears under the network variance your customers already experience. The design that does hurt is a synchronous INSERT per request; that adds 3 to 15 ms and consumes pool connections your handlers need, which is how instrumentation ends up blamed for an outage it merely revealed.

What happens to usage records when a customer rotates their API key mid-month? Nothing, as long as you store both api_key_id and customer_id on every event and key the quota counter on the customer rather than the key. Billing aggregates by customer, so a rotation is invisible on the invoice while per-key breakdowns still show which credential drove the traffic. See rotating API keys without downtime for the overlap window that makes rotation safe.

How risky is migrating an existing API onto this pipeline? Low, because you can run it in shadow mode. Deploy the middleware and flusher with USAGE_QUOTA_ENFORCED=false, collect events for a full billing period, and compare your computed totals against whatever you invoice today. Only when the two agree within a fraction of a percent do you switch enforcement on and point Stripe at the meter. The one irreversible decision is the billable unit, so settle that before you collect a month of data you cannot reinterpret.