FastAPI endpoints cached with cashews in-memory backend: working versions (tested 2026-10-15)

fastapi==0.143.2
cashews==7.6.0
httpx==0.28.1
pytest==9.1.1
pytest-asyncio==1.4.0

from fastapi import FastAPI, HTTPException
from cashews import cache
from cashews.contrib.fastapi import CacheRequestControlMiddleware
from pydantic import BaseModel

cache.setup("mem://")


class Item(BaseModel):
    id: int
    name: str
    stock: int


class ItemNotFound(Exception):
    pass


class _Store:
    def __init__(self) -> None:
        self.items: dict[int, Item] = {}
        self.loads = 0

    def reset(self) -> None:
        self.items = {
            1: Item(id=1, name="widget", stock=3),
            2: Item(id=2, name="gadget", stock=9),
        }
        self.loads = 0


store = _Store()
store.reset()

app = FastAPI()
app.add_middleware(CacheRequestControlMiddleware)


@cache(ttl="5m", key="item:{item_id}")
async def load_item(item_id: int) -> Item:
    store.loads += 1
    try:
        return store.items[item_id].model_copy(deep=True)
    except KeyError:
        raise ItemNotFound(item_id) from None


@app.get("/items/{item_id}", response_model=Item)
async def get_item(item_id: int) -> Item:
    try:
        return await load_item(item_id)
    except ItemNotFound:
        raise HTTPException(status_code=404, detail="item not found")


async def set_stock(item_id: int, stock: int) -> Item:
    item = store.items.get(item_id)
    if item is None:
        raise ItemNotFound(item_id)
    item.stock = stock
    await cache.delete(f"item:{item_id}")
    return item.model_copy(deep=True)


@app.put("/items/{item_id}", response_model=Item)
async def put_item(item_id: int, stock: int) -> Item:
    try:
        return await set_stock(item_id, stock)
    except ItemNotFound:
        raise HTTPException(status_code=404, detail="item not found")

import httpx
import pytest
from httpx import ASGITransport

import app as app_module
from cashews import cache


@pytest.fixture(autouse=True)
async def fresh_state():
    cache.setup("mem://")
    app_module.store.reset()
    yield
    await cache.clear()


@pytest.fixture
async def client():
    transport = ASGITransport(app=app_module.app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


async def test_get_is_cached(client):
    first = await client.get("/items/1")
    second = await client.get("/items/1")
    assert first.status_code == second.status_code == 200
    assert first.json() == {"id": 1, "name": "widget", "stock": 3}
    assert app_module.store.loads == 1


async def test_key_tracks_path_param(client):
    await client.get("/items/1")
    await client.get("/items/2")
    assert app_module.store.loads == 2


async def test_update_invalidates_entry(client):
    assert (await client.get("/items/1")).json()["stock"] == 3
    updated = await client.put("/items/1", params={"stock": 10})
    assert updated.json()["stock"] == 10
    assert (await client.get("/items/1")).json()["stock"] == 10
    assert app_module.store.loads == 2


async def test_missing_item_is_not_cached(client):
    assert (await client.get("/items/99")).status_code == 404
    assert (await client.get("/items/99")).status_code == 404
    assert app_module.store.loads == 2


async def test_ttl_is_exposed(client):
    await client.get("/items/1")
    ttl = await cache.get_expire("item:1")
    assert 0 < ttl <= 300


async def test_cache_control_header(client):
    await client.get("/items/1")
    hit = await client.get("/items/1")
    assert hit.headers["cache-control"].startswith("private, max-age=")
    assert int(hit.headers["age"]) > 0

[pytest]
pythonpath = .
asyncio_mode = auto