FastAPI with Redis caching: TTLs, invalidation, and failure handling

A FastAPI Redis cache can avoid repeating an expensive read, but it also creates a second copy of data that can expire, disappear, or become stale. Start with one read path whose freshness requirements you can explain. Keep the source of truth authoritative and make a cache miss a normal application path.

This guide builds a small cache-aside example with an asynchronous Redis client, tenant-specific keys, a TTL, an invalidation helper, and bounded cache calls. Use the FastAPI deployment guide for the HTTP service and the private Redis Compose guide for the separate Redis service. The example below does not combine those deployments or establish cloud performance.

Validation, November 10, 2026: 21 local Python tests passed using FastAPI TestClient and a deterministic cache test double. They cover hit/miss behavior, simulated expiry, invalidation, tenant separation, rejected credentials, malformed cached data, connection/timeout fallback, and client configuration without connecting. A separate credential-helper check passed. No real Redis server, Redis TTL timing, authentication handshake, Linux/container networking, multi-process race, load test, or Hivenet deployment was run. Docker’s Linux engine was unavailable. The in-memory origin is a fixture, not a database; this article makes no speedup claim.

Choose a read path before choosing a caching library

Cache-aside means the application checks the cache, reads the origin on a miss, then stores a copy for later requests. It can help when reads repeat and the origin work is expensive enough to justify serialization and an extra network hop. It can add latency and maintenance work when the source is already cheap or requests rarely repeat.

Write down what may be stale and for how long. A product description and an account balance need different rules. Do not cache authorization decisions or personalized output under a shared product key. The sample deliberately returns a tiny tenant-specific catalog fixture, with no write endpoint and no production identity system.

Caching also differs from queuing work. A missed cache entry should be recoverable from the origin; it is not a pending job. For deferred work, use the separate FastAPI background-tasks guide. For service ownership and operating cost, start with Redis hosting.

Keep identity, keys, and response caching separate

The sample maps two randomly generated demo API keys to fixed tenant identifiers. Authorization runs before every cache lookup, including hits. FastAPI’s APIKeyHeader helper extracts the header; our dependency performs the credential check. A real application must derive tenant and resource access from its established identity and authorization system on every request.

Keys include a namespace, payload version, tenant, and product ID:

lab:fastapi:v1:tenant:north:product:1

The tenant is server-derived, not accepted from a query parameter. Include every dimension that changes a response, such as locale or permission-dependent fields, or choose not to cache that response. Never use an API token as the cache key. Tenant names in the key prevent this example’s collisions; they do not create Redis ACL isolation between tenants sharing the same service credential.

Successful HTTP responses use Cache-Control: no-store, while the application may still store data in Redis. Those are separate layers. The diagnostic X-Cache header reports HIT, MISS, or BYPASS in this lab. It is not a measured latency or capacity result.

Create and close one Redis client per application process

Use FastAPI’s lifespan context to create the client once and close it at shutdown. Creating a client for every request wastes connections. Each Uvicorn worker has its own pool, so a per-process maximum of ten connections is not a ten-connection cap for an entire multi-worker deployment.

The sample pins redis-py 6.4.0, which was available in the local test environment. It uses the versioned asyncio API, including aclose(). Current documentation describes newer client defaults; these pins record what was checked, not a recommendation to use the newest or an older release in every service. Upgrade separately and rerun both local and real-service checks.

Save the following dependencies as requirements.txt in a new project. The observed environment also used Python 3.12.14 and Starlette 1.10.0. These direct pins are not a full dependency lock:

fastapi==0.162.0
pydantic==2.14.1
redis==6.4.0
uvicorn==0.54.0
httpx==0.28.1
pytest==9.2.0

Save the application as app.py. DemoOrigin is intentionally an in-memory fixture. Replace it with a properly bounded asynchronous source adapter before using the pattern for real data:

"""Disposable cache-aside example. The origin is an in-memory fixture."""
import asyncio
import json
import logging
import os
import secrets
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Path as ApiPath, Request, Response
from fastapi.security import APIKeyHeader
from pydantic import BaseModel, ConfigDict, Field, ValidationError
from redis.asyncio import Redis
from redis.asyncio.retry import Retry
from redis.backoff import NoBackoff
from redis.exceptions import RedisError

TTL_SECONDS = 30
CACHE_TIMEOUT = 0.15
log = logging.getLogger("cache_demo")
header = APIKeyHeader(name="X-API-Key", auto_error=False)


class Product(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    tenant: str
    id: int
    name: str = Field(max_length=120)


class DemoOrigin:
    def __init__(self):
        self.rows = {("north", 1): "North notebook", ("south", 1): "South notebook"}
        self.reads = 0

    async def load(self, tenant, product_id):
        self.reads += 1
        name = self.rows.get((tenant, product_id))
        return None if name is None else Product(tenant=tenant, id=product_id, name=name)


def cache_key(tenant, product_id):
    return f"lab:fastapi:v1:tenant:{tenant}:product:{product_id}"


def make_redis():
    password = Path(os.environ["REDIS_PASSWORD_FILE"]).read_text().strip()
    if not password:
        raise ValueError("Redis password file must not be empty")
    return Redis(
        host=os.environ["REDIS_HOST"], port=6379, username="app",
        password=password, db=0, protocol=2, decode_responses=True,
        socket_connect_timeout=0.1, socket_timeout=0.1, max_connections=10,
        retry=Retry(NoBackoff(), 0), health_check_interval=0,
        lib_name=None, lib_version=None,
    )


async def invalidate(cache, tenant, product_id):
    """Call after a successful source commit; failure needs reconciliation."""
    try:
        await asyncio.wait_for(cache.delete(cache_key(tenant, product_id)), CACHE_TIMEOUT)
        return True
    except (RedisError, TimeoutError):
        log.warning("cache_delete_failed")
        return False


def create_app(cache_factory=make_redis, origin=None, tokens=None):
    @asynccontextmanager
    async def lifespan(app):
        credentials = tokens
        if credentials is None:
            credentials = json.loads(Path(os.environ["API_KEYS_FILE"]).read_text())
        if (not isinstance(credentials, dict) or set(credentials) != {"north", "south"}
                or any(not isinstance(v, str) or len(v) != 64
                       or any(c not in "0123456789abcdef" for c in v)
                       for v in credentials.values())
                or len(set(credentials.values())) != 2):
            raise ValueError("Supply two distinct 64-character lowercase hex demo keys")
        app.state.tokens = credentials
        app.state.origin = origin if origin is not None else DemoOrigin()
        app.state.cache = cache_factory()
        try:
            yield
        finally:
            await app.state.cache.aclose()

    app = FastAPI(lifespan=lifespan)

    async def authorized_tenant(request: Request, supplied: str | None = Depends(header)):
        if supplied is not None and supplied.isascii():
            for tenant, expected in request.app.state.tokens.items():
                if secrets.compare_digest(supplied, expected):
                    return tenant
        raise HTTPException(status_code=401, detail="Invalid API key")

    @app.get("/products/{product_id}", response_model=Product)
    async def product(
        product_id: Annotated[int, ApiPath(ge=1, le=100000)],
        request: Request, response: Response,
        tenant: str = Depends(authorized_tenant),
    ):
        response.headers["Cache-Control"] = "no-store"
        cache = request.app.state.cache
        key = cache_key(tenant, product_id)
        cache_available = True
        try:
            raw = await asyncio.wait_for(cache.get(key), CACHE_TIMEOUT)
        except (RedisError, TimeoutError, UnicodeError):
            raw, cache_available = None, False
            log.warning("cache_read_failed")
        if raw is not None and len(raw) <= 8192:
            try:
                cached = Product.model_validate_json(raw)
                if cached.tenant == tenant and cached.id == product_id:
                    response.headers["X-Cache"] = "HIT"
                    return cached
            except ValidationError:
                pass
            log.warning("cache_value_rejected")
        value = await request.app.state.origin.load(tenant, product_id)
        if value is None:
            raise HTTPException(status_code=404, detail="Product not found")
        response.headers["X-Cache"] = "MISS" if cache_available else "BYPASS"
        if cache_available:
            try:
                await asyncio.wait_for(cache.set(key, value.model_dump_json(), ex=TTL_SECONDS), CACHE_TIMEOUT)
            except (RedisError, TimeoutError):
                log.warning("cache_write_failed")
                response.headers["X-Cache"] = "BYPASS"
        return value

    return app

Understand the hit, miss, and failure paths

A hit returns only a schema-checked value whose tenant and product ID match the request. Missing, malformed, oversized, wrongly scoped, or undecodable cached data leads to an origin read. Only the fixture’s valid product responses are cached; missing products return 404 without creating a negative-cache entry.

On a successful miss, SET writes the value and its 30-second expiry together through ex=TTL_SECONDS. This avoids a separate set-then-expire sequence. The SET reference defines the expiration option. Thirty seconds is an illustrative value, not a freshness requirement established for your application.

The client uses 100 ms connect/socket timeouts, zero automatic retries, and a 150 ms application deadline per cache call. These are lab settings, not a request-wide latency promise. A successful miss can attempt both GET and SET, and the origin read has its own cost. Versioned connection options document the constructor settings; Redis client production guidance covers timeout and retry tradeoffs.

If a cache read fails, the app skips the cache write for that request and reads the origin. If the write fails, it still returns the origin response. Redis errors are logged as short event names without credentials or cached values. Cancellation and unrelated programming errors are not swallowed. A cache outage therefore increases origin traffic: this example has no circuit breaker, admission control, origin timeout, or request coalescing. Those gaps matter before production use.

Invalidate after the source commit, and account for races

When the authoritative value changes, commit that change first, then call invalidate(cache, tenant, product_id). The helper removes one exact key with DEL and reports whether the request succeeded. It does not expose a public deletion endpoint or scan the keyspace. A successful delete request can be harmless when the key is already absent.

A false result needs an explicit response: retry through a durable reconciliation path, bypass caching for the affected data, or accept a documented staleness window. The sample only logs and returns failure. It has no durable invalidation queue. Do not tell a caller that a database commit failed solely because cache deletion failed; the database outcome and cache outcome are separate.

Delete-after-commit also has a race: a reader can fetch the old source value before the write, then populate the cache after deletion. TTL limits the life of that cached write, not the time since the source changed. Versioned source values, revision-aware keys, or coordinated reads/writes may be needed for stronger consistency. The sequential local test checks helper behavior; it does not solve or validate that race.

Simultaneous misses can also duplicate source work. Adding a lock requires its own ownership, expiry, timeout, and failure design. This guide does not include a distributed lock or promise protection from a cache stampede. Measure the cold-cache and unavailable-cache paths before relying on the warm path.

Keep the real connection private and scoped

The example Redis client is configured for a trusted private connection to the earlier Compose lab. It uses user app, database 0, RESP2, and keys under lab:*. It disables optional client-library metadata announcements and issues GET, SET, and DEL, which the lab ACL grants. Constructor checks do not prove that a real authentication handshake succeeds. Do not widen the ACL just to suppress an unexplained error.

Supply REDIS_HOST, REDIS_PASSWORD_FILE, and API_KEYS_FILE to the application process. The hostname redis resolves only for a service attached to the appropriate Compose network; it is not automatically reachable from your laptop or an unrelated container. Keep port 6379 unpublished. Password authentication alone does not encrypt traffic, as the Redis security guidance explains. This sample does not configure TLS, an external managed endpoint, or a cross-host route.

For a disposable evaluation, create new API credentials with this prepare_api_keys.py helper. It refuses to overwrite an existing file. Keep that file private, exclude it from version control, and use different credentials from any real service:

"""Generate two disposable API credentials; never print or overwrite them."""
import json
import secrets
from pathlib import Path

path = Path(__file__).resolve().parent / "api_keys.json"
with path.open("x", encoding="ascii") as handle:
    json.dump({tenant: secrets.token_hex(32) for tenant in ("north", "south")}, handle)
path.chmod(0o600)
print("Created api_keys.json. Keep it private and out of version control.")

With the environment and private route configured, the proposed local-only HTTP launch is:

python -m uvicorn app:create_app --factory --host 127.0.0.1 --port 8000 --workers 1

That launch and the real Redis connection were not executed here. Do not substitute a public bind address without the authentication and HTTPS work in the deployment guide. A file permission declaration on Windows is not evidence of Linux permission enforcement. Missing configuration is a startup error; Redis unavailability during a request follows the fallback path.

Read the local evidence before running an integration test

The downloadable sample includes test_app.py. From a Python environment containing the pinned dependencies, run:

python -m pytest test_app.py -q

Observed locally: 21 passed, 1 warning. The warning was Starlette’s deprecation notice for the installed HTTPX-based TestClient path. It did not fail the suite; migrating that test dependency was not part of this sample. TestClient ran the lifespan and request handlers in process. The FastAPI test-lifespan guidance explains why the context manager matters.

• First miss, subsequent hit, unchanged response data, and a 30-second TTL argument.
• Expiry by advancing the test double’s clock; sequential invalidation and visible invalidation failure.
• Separate tenant keys; unauthorized requests never reaching the cache or origin.
• Connection errors, a simulated client timeout, and a real application timeout around a sleeping test double.
• Write failure, invalid JSON, wrong tenant or product, oversized data, decoding failure, 404 behavior, and invalid IDs.
• Client configuration inspected without a connection, client closure, and rejection of duplicate demo API keys.

These checks validate application branches, not Redis protocol behavior. Before deployment, test a disposable real Redis instance for credentials, ACL acceptance, actual TTL expiry, eviction, failed writes, connection loss, and recovery. Verify each tenant’s response through the HTTP route, including warm-cache authorization. Then test multiple processes, cold starts, peak concurrency, origin overload, and invalidation races. Record which checks ran rather than presenting the list as results.

Use caching results to inform CPU and memory choices

Compare the same representative workload with caching disabled, cold, warm, and unavailable. Record response correctness, origin reads, achieved request rate, latency, errors, CPU, and memory. The fixture’s avoided dictionary lookup is not evidence that your database-backed API will be faster. Extra application workers also multiply connection pools and memory use.

Choose a cache eviction policy only for data the application can rebuild. A TTL does not prevent earlier eviction. The Redis memory and eviction guide separates maxmemory from process and host limits. Keep a cache’s policy separate from a queue’s delivery requirements.

Use the CPU VM buying guide with measured resource needs, then confirm the actual access path and capacity in Hivenet’s CPU documentation. This walkthrough describes a customer-operated application pattern; it establishes no Hivenet managed Redis service, availability, benchmark, or production deployment result.