DEV Community: Xitong Shan

A Redis Lua Rate Limiter for FastAPI That Returns Useful 429 Headers

Xitong Shan — Wed, 26 Aug 2026 09:25:39 +0000

Rate limiting is easy to describe and surprisingly easy to implement incorrectly. A request counter must change atomically, the counter needs a reliable expiry, different endpoints need different identities, and blocked clients need enough information to back off.

This article explains the small Redis-backed limiter in my public FastAPI project, mini-agent. It protects an authenticated AI-agent chat endpoint and a login endpoint. The design uses one Lua script for the counter and expiry, hashes identities before placing them in Redis keys, and returns 429 Too Many Requests with Retry-After and rate-limit headers.

The examples are pinned to this public commit, not an unspecified future version of the repository.

Define the result before the storage details

The rest of the application should not need to understand Redis return values. The limiter translates them into a small data object:

@dataclass
class RateLimitResult:
    allowed: bool
    scope: str
    limit: int
    count: int
    remaining: int
    retry_after_seconds: int

scope distinguishes policies such as chat_per_minute, chat_per_day, login_ip_per_minute, and login_user_ip_per_minute. Returning the count and remaining allowance makes the decision observable without requiring callers to read the Redis key.

The full implementation is in rate_limit.py.

Keep increment, first expiry, and TTL in one Lua script

A naive implementation issues separate INCR and EXPIRE commands. If the process fails between them, the counter may be left without an expiry. Concurrent requests can also observe an incomplete update.

Redis executes a Lua script atomically, so this implementation performs the related operations together:

local current = redis.call("INCR", KEYS[1])

if current == 1 then
    redis.call("EXPIRE", KEYS[1], tonumber(ARGV[1]))
end

local ttl = redis.call("TTL", KEYS[1])

return {current, ttl}

The first hit creates the counter and starts its expiry. Later hits increment the same key without extending the window. The Python layer calculates the decision:

count_raw, ttl_raw = await self.client.eval(
    _FIXED_WINDOW_SCRIPT, 1, key, window_seconds
)

count = int(count_raw)
ttl = int(ttl_raw)
if ttl < 0:
    ttl = window_seconds

allowed = count <= limit
remaining = max(limit - count, 0)

This is an expiry-backed fixed window anchored by the first hit. It is not a sliding-window algorithm and it is not aligned to wall-clock minute boundaries. That simplicity is acceptable for many APIs, but it permits bursts around the moment one key expires and another starts.

There is also a recovery detail worth noticing. If TTL unexpectedly returns a negative value, the code reports the configured window as the retry delay, but it does not repair the missing expiry. A hardened version should either restore the expiry inside the Lua script or fail closed and emit an operational alert.

Build scoped keys without storing raw identities

The Redis key combines the policy scope, window size, and a SHA-256 digest:

def hash_identity(identity: str) -> str:
    return hashlib.sha256(identity.encode("utf-8")).hexdigest()

def build_key(self, *, scope, identity, window_seconds):
    identity_hash = hash_identity(identity)
    return f"rate:{scope}:{window_seconds}:{identity_hash}"

This prevents raw user IDs, usernames, and IP addresses from appearing directly in normal key listings. It does not make low-entropy identities secret: a deterministic, unsalted hash of a known IP range can be guessed offline. If the threat model requires stronger pseudonymization, use an HMAC with a rotated server-side key and document how rotations affect active windows.

Including the scope prevents a login attempt from consuming the chat allowance. Including the window length avoids a collision if the same scope name is mistakenly reused with different durations, although explicit unique scope names remain clearer.

Apply two policies to authenticated chat

The chat endpoint checks both a per-minute and per-day limit against the authenticated user_id:

per_minute = await check_limit_or_raise(
    limiter=limiter,
    scope="chat_per_minute",
    identity=current_user.user_id,
    limit=CHAT_RATE_LIMIT_PER_MINUTE,
    window_seconds=60,
)

await check_limit_or_raise(
    limiter=limiter,
    scope="chat_per_day",
    identity=current_user.user_id,
    limit=CHAT_RATE_LIMIT_PER_DAY,
    window_seconds=60 * 60 * 24,
)

FastAPI dependency injection makes this run after authentication but before the route body. get_rate_limited_current_user receives the authenticated user, checks the limits, and saves the successful minute result on request.state.

The chat route then exposes the successful minute allowance in both response headers and its structured response. The daily result is enforced but is not included in a successful response. If clients need both budgets, return or attach both results rather than implying one set of headers describes every active policy.

Rate-limit login without helping username enumeration

The login route has a different identity problem because there is no authenticated user yet. The implementation checks:

attempts from one client IP; and
attempts for one username + IP pair.

Both checks happen regardless of whether the username exists. That avoids creating a simple side channel in which nonexistent usernames skip one rate-limit path.

The code includes an important proxy warning. It currently takes the first address from X-Forwarded-For when present, otherwise it uses request.client.host. A public client can forge X-Forwarded-For unless a trusted reverse proxy strips and rewrites it. Production code must trust forwarding headers only when the direct peer is an approved proxy or when the framework's proxy configuration has already validated the chain.

This is not a minor deployment detail. If arbitrary clients can choose the identity used by the IP limiter, they can evade one of the login protections.

Make a blocked response actionable

When a limit is exceeded, the helper raises FastAPI's HTTPException with status 429:

def rate_limit_headers(result):
    return {
        "X-RateLimit-Scope": result.scope,
        "X-RateLimit-Limit": str(result.limit),
        "X-RateLimit-Remaining": str(result.remaining),
        "X-RateLimit-Reset": str(result.retry_after_seconds),
    }

def raise_rate_limited(result):
    headers = rate_limit_headers(result)
    headers["Retry-After"] = str(result.retry_after_seconds)
    raise HTTPException(
        status_code=429,
        detail=f"请求过于频繁，请 {result.retry_after_seconds} 秒后再试。",
        headers=headers,
    )

Retry-After gives an automated client a concrete delay. The additional headers help a UI explain which policy was reached. Here, X-RateLimit-Reset contains seconds remaining, not an absolute timestamp. Because header conventions differ across APIs, document that meaning or adopt the current standardized RateLimit fields consistently.

Clients should still add jitter and avoid firing every queued request at the exact expiry boundary.

Test the behavior through the API boundary

The repository has two concise integration tests in test_rate_limit.py:

with the login username-and-IP limit set to two, two bad passwords return 401 and the third attempt returns 429;
with the per-minute chat limit set to two, two authenticated chat calls return 200 and the third returns 429.

Both blocked responses must include Retry-After. Those checks validate the observable contract rather than only the internal counter.

Additional tests I would add before deploying this pattern include:

simultaneous hits at the limit boundary;
key expiry followed by a fresh allowed request;
separate counters for two users and two scopes;
a simulated key with no TTL and the intended repair behavior;
a daily-limit rejection after the minute limit succeeds;
trusted-proxy and forged-forwarding-header cases;
Redis timeout behavior, with an explicit fail-open or fail-closed policy.

The public CI run for the pinned commit completed successfully. That is useful evidence for this exact revision, but a passing test suite does not resolve the algorithm's documented limitations.

Know when to move beyond this design

An expiry-backed counter is compact and fast. It works well when approximate fixed-window fairness is acceptable and the main goals are abuse resistance and predictable resource use.

It is a weaker fit when customers require smooth quotas, globally coordinated multi-region enforcement, or billing-grade accuracy. Those cases may need a sliding-window log or counter, token bucket, gateway-level limiter, or a dedicated distributed rate-limit service. Whatever algorithm you choose, keep the same application-level disciplines: scoped identities, atomic state changes, useful 429 responses, proxy-aware client identity, and tests at the HTTP boundary.

Disclosure: This article describes a self-built public demonstration project, not a client production incident. It was prepared with AI-assisted editing. The author is responsible for reviewing the code, technical claims, links, and final text before publication.``

Tracing a FastAPI AI Agent Without Logging the Whole Prompt

Xitong Shan — Wed, 26 Aug 2026 09:24:45 +0000

An AI-agent endpoint is harder to debug than a conventional CRUD route. A single request may load a session, call a model, run one or more tools, pause for approval, and save new state. When the final response is slow or wrong, an ordinary access log often tells you only that POST /chat returned a status code.

At the same time, logging the complete prompt, tool arguments, or retrieved documents is a poor default. Those values can contain customer text, credentials, or other sensitive data.

This article walks through a deliberately small tracing layer from my public FastAPI project, mini-agent. The implementation writes one JSON object per line, keeps related operations under a shared trace ID, supports nested spans, and scrubs selected sensitive fields. It is not a replacement for OpenTelemetry, but it makes the data model and privacy decisions visible before adding a larger observability stack.

The code referenced here is pinned to one public commit, so the examples remain reproducible even if the main branch changes.

Start with correlation, not message bodies

The tracer creates a few identifiers with different jobs:

trace_id groups all recorded work for one agent request.
request_id connects the trace to the API request.
session_id connects multiple requests in the same conversation.
span_id identifies one timed operation.
parent_span_id represents nesting.

That structure is more useful than a large unstructured message. It lets an operator search for one failed request, reconstruct its hierarchy, and compare durations without requiring the full user prompt.

The constructor accepts IDs supplied by the application and generates the missing ones:

class AgentTracer:
    def __init__(self, *, trace_id=None, request_id=None,
                 session_id=None, user_id=None):
        self.trace_id = trace_id or gen_id("trace")
        self.request_id = request_id or gen_id("req")
        self.session_id = session_id
        self.user_id = user_id
        self._span_stack: list[str] = []

See the complete implementation in observability.py.

Model a span as a context manager

A context manager keeps the timing and error path in one place. The caller opens a span around an operation, and the tracer emits the result in finally, whether the operation succeeds or raises.

@contextmanager
def span(self, name, *, kind="internal", attributes=None):
    span_id = gen_id("span")
    parent_span_id = self._span_stack[-1] if self._span_stack else None
    start_perf = time.perf_counter()
    start_time_ms = now_ms()
    self._span_stack.append(span_id)

    status = "ok"
    error = None
    try:
        yield {
            "trace_id": self.trace_id,
            "span_id": span_id,
            "parent_span_id": parent_span_id,
        }
    except Exception as exc:
        status = "error"
        error = {
            "type": type(exc).__name__,
            "message": str(exc),
            "stack": traceback.format_exc(limit=5),
        }
        raise
    finally:
        duration_ms = round((time.perf_counter() - start_perf) * 1000, 2)
        popped = self._span_stack.pop()
        assert popped == span_id
        self._emit({
            "type": "span",
            "trace_id": self.trace_id,
            "span_id": span_id,
            "parent_span_id": parent_span_id,
            "name": name,
            "kind": kind,
            "status": status,
            "start_time_ms": start_time_ms,
            "duration_ms": duration_ms,
            "attributes": scrub(attributes or {}),
            "error": scrub(error),
        })

There are two clocks on purpose. Wall-clock milliseconds make records searchable by time, while time.perf_counter() measures elapsed time without depending on wall-clock adjustments.

The stack gives a new span the current span as its parent. A model span can contain a tool span, for example. Because records are written when each context exits, a child span will normally appear in the JSONL file before its parent. Consumers should reconstruct the tree from IDs rather than assume log-line order is start order.

The assertion after pop() is a useful development guard, but this simple list assumes one sequential execution context. If multiple asynchronous tasks share one tracer and overlap their spans, a plain shared stack can produce the wrong parent or fail the assertion. A production version should use context-local state, such as contextvars, or create separate span objects for concurrent branches.

Emit events for points that do not need timing

Not every useful observation is an interval. The same tracer also writes instantaneous events. An event uses the active span as its parent when one exists:

def event(self, name, *, attributes=None):
    parent_span_id = self._span_stack[-1] if self._span_stack else None
    self._emit({
        "type": "event",
        "trace_id": self.trace_id,
        "request_id": self.request_id,
        "parent_span_id": parent_span_id,
        "name": name,
        "time_ms": now_ms(),
        "attributes": scrub(attributes or {}),
    })

The chat route uses an event for an unhandled endpoint error. A tool runner could use events for decisions such as approval_requested or fallback_selected, provided their attributes remain safe.

Put a narrow span around the agent execution

In the /chat route, the application creates a tracer after authentication and session-ID selection. It then wraps the synchronous session.send call, which is moved to Starlette's thread pool:

tracer = AgentTracer(
    request_id=request_id,
    session_id=session_id,
    user_id=current_user.user_id,
)

with tracer.span(
    "http.post./chat",
    kind="server",
    attributes={
        "session_id": session_id,
        "user_id": current_user.user_id,
        "message_preview": safe_preview(req.message),
        "has_existing_session": req.session_id is not None,
    },
):
    answer = await run_in_threadpool(session.send, req.message, tracer)

The response returns trace_id and request_id. That is operationally important: a user or support engineer can report a trace ID instead of copying private conversation text into a ticket.

The span name says http.post./chat, but its current boundary covers the agent execution, not session loading and saving. Naming it agent.session.send would be more precise, or the route could add a true outer server span and keep this as a child span. Trace names should describe the measured boundary, not merely the surrounding function.

Redact by default, then document the gaps

The helper safe_preview() replaces newlines and keeps at most 120 characters by default. The recursive scrub() function:

replaces values whose exact, case-insensitive key is in a sensitive-key set;
truncates strings longer than 500 characters;
limits logged list items to the first 20;
applies the same rules inside nested dictionaries and lists.

This is safer than serializing arbitrary inputs unchanged, but it is not a complete data-loss-prevention system. Exact-key matching will redact token but not necessarily auth_token_value. A short preview may still contain a name, address, or account number. Exception messages and stack traces can also expose values.

For production use, I would tighten the policy in four ways:

Disable prompt previews by default and enable them only in an explicitly safe environment.
Use allow-listed attributes for each event or span instead of accepting arbitrary dictionaries.
Add pattern-based secret detection and domain-specific personal-data filtering.
Define retention, access control, and deletion rules for trace storage.

JSONL on local disk is convenient for development, but it also needs rotation and multi-process handling. The implementation uses one Python logger with a file and console handler; it does not provide distributed export, sampling, backpressure, or cross-service propagation.

Verify behavior at the record boundary

Useful checks do not need a live model. A focused test can replace the logger handler, open nested spans, and parse each emitted JSON line. It should verify:

child records contain the expected parent_span_id;
successful and failing spans have the correct status;
exceptions are re-raised after being recorded;
sensitive exact keys become [REDACTED];
long strings and lists are bounded;
a multiline preview contains the escaped \\n representation;
response payloads include non-empty trace and request IDs.

The repository's public CI run for the pinned commit completed successfully. The API test also checks that a successful chat response includes a trace ID; see test_chat_api.py.

A small tracer is a design exercise, not the destination

Building this layer clarified the contract I would carry into OpenTelemetry: stable correlation IDs, explicit parent-child relationships, duration, status, bounded attributes, and a way to connect a client-visible error to an internal trace.

The next step is not to keep expanding a custom tracing system indefinitely. It is to map the same boundaries to standard spans, propagate context across services and worker tasks, export through OTLP, and view latency and error distributions in an observability backend. Starting with a small implementation makes that migration easier because the privacy boundary and the meaning of each span are already explicit.

Fixing a pgvector CI mismatch in a FastAPI RAG backend

Xitong Shan — Sat, 22 Aug 2026 15:33:05 +0000

This is a submission for DEV's Summer Bug Smash: Clear the Lineup, powered by Sentry.

Project Overview

mini-agent is a public FastAPI backend for an AI support-agent demo. Its test suite covers API behavior, authentication, rate limiting, approval flows, and PostgreSQL/pgvector-backed retrieval.

The GitHub Actions workflow starts PostgreSQL and Redis service containers before running the Python test suite. The application database initialization also executes:

CREATE EXTENSION IF NOT EXISTS vector

The dependency is also visible in the DocumentChunk.embedding column, which uses pgvector's Vector type. That made the database image part of the test contract, not just incidental infrastructure.

Bug Fix or Performance Improvement

On August 12, 2026, the CI run for the preceding commit reached the test step and failed:

The workflow was using the general-purpose postgres:17-alpine service image, while the application required the pgvector extension during database initialization. The test environment therefore did not match the database capability required by the code.

The failure was specific enough to avoid a broad rewrite: the container initialized successfully, dependency installation passed, and the workflow stopped only at Run tests. That pointed to the application/database boundary rather than the GitHub Actions runner or Python installation.

The fix changed one line:

 services:
   postgres:
-    image: postgres:17-alpine
+    image: pgvector/pgvector:0.8.6-pg17

Full change: Use pgvector image in CI

The PostgreSQL major version, credentials, port mapping, health check, application environment, dependency installation, and test command all remained unchanged. This kept the patch narrow and made the CI database expose the same required extension as the application.

Code

The evidence is a direct before-and-after pair:

The preceding workflow failed at Run tests.
The one-line database-image commit triggered a new workflow.
The new run completed Run tests successfully and the overall workflow passed.

Check	Before	After
PostgreSQL service image	`postgres:17-alpine`	`pgvector/pgvector:0.8.6-pg17`
Container initialization	Passed	Passed
Dependency installation	Passed	Passed
`Run tests`	Failed	Passed
Overall workflow	Failed	Passed

No tests were disabled, no failure was ignored, and no application feature was added to make the build green.

My Improvements

The visible patch is small, but it fixes an important reliability boundary: integration tests are only meaningful when their service dependencies provide the capabilities the application actually uses.

The change:

restores the full CI test run;
makes the PostgreSQL service compatible with the repository's pgvector-backed model;
preserves PostgreSQL 17 rather than changing database versions as a side effect;
avoids installing database extensions ad hoc during every CI run;
keeps the workflow readable and reproducible with a pinned pgvector image tag.

The general lesson is to treat database extensions as explicit runtime dependencies. If application startup creates or queries extension-backed types, CI must supply that extension too. A healthy generic PostgreSQL container is not enough when the application's schema contract includes extension-defined types.

Disclosure

The code change, commit timestamps, and workflow outcomes are public and independently inspectable through the links above. This write-up was prepared with AI-assisted editing and was personally reviewed by the entrant before publication. No production data, customer credentials, or private incident details are included.