Skip to content

Caching & Conditional Requests

Server-side response caching and HTTP conditional-request handling. Import from fastapi_views.cache, which exports Cache, cache, CacheControl, CacheHeaders, CacheMiddleware, CachedAPIView, ConditionalCachedAPIView and use_cache.

The Redis backend requires the cache extra: pip install "fastapi-views[cache]".

For a walkthrough see Caching & Conditional Requests.

Views and decorator

CacheControl dataclass

Builder for a Cache-Control response header value.

Pass to :func:use_cache instead of a raw string to compose directives safely. private keeps a per-user/per-tenant response out of shared caches; s_maxage sets a separate freshness lifetime for them.

Boolean fields render as bare directives (no-store); int fields render as name=value (max-age=30); None / False are omitted. Field names map to directives by replacing _ with -.

Source code in fastapi_views/cache/view.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@dataclass(frozen=True)
class CacheControl:
    """Builder for a ``Cache-Control`` response header value.

    Pass to :func:`use_cache` instead of a raw string to compose directives
    safely. ``private`` keeps a per-user/per-tenant response out of shared
    caches; ``s_maxage`` sets a separate freshness lifetime for them.

    Boolean fields render as bare directives (``no-store``); int fields render
    as ``name=value`` (``max-age=30``); ``None`` / ``False`` are omitted. Field
    names map to directives by replacing ``_`` with ``-``.
    """

    no_store: bool = False
    no_cache: bool = False
    public: bool = False
    private: bool = False
    max_age: int | None = None
    s_maxage: int | None = None
    must_revalidate: bool = False
    immutable: bool = False
    stale_while_revalidate: int | None = None
    stale_if_error: int | None = None

    def render(self) -> str:
        """Render the directives into a header value, in declaration order."""
        directives: list[str] = []
        for field in fields(self):
            value = getattr(self, field.name)
            if value is None or value is False:
                continue
            name = field.name.replace("_", "-")
            directives.append(name if value is True else f"{name}={value}")
        return ", ".join(directives)

render() -> str

Render the directives into a header value, in declaration order.

Source code in fastapi_views/cache/view.py 
58
59
60
61
62
63
64
65
66
67
def render(self) -> str:
    """Render the directives into a header value, in declaration order."""
    directives: list[str] = []
    for field in fields(self):
        value = getattr(self, field.name)
        if value is None or value is False:
            continue
        name = field.name.replace("_", "-")
        directives.append(name if value is True else f"{name}={value}")
    return ", ".join(directives)

CacheHeaders

Bases: ResponseHeaders

Cache-specific response headers documented for cached endpoints.

The ETag / Last-Modified validators are not declared here; they are contributed dynamically by :class:~fastapi_views.views.mixins.ConditionalMixin only when the view actually emits them.

Source code in fastapi_views/cache/view.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
class CacheHeaders(ResponseHeaders):
    """Cache-specific response headers documented for cached endpoints.

    The ``ETag`` / ``Last-Modified`` validators are not declared here; they are
    contributed dynamically by :class:`~fastapi_views.views.mixins.ConditionalMixin`
    only when the view actually emits them.
    """

    x_cache: Literal["HIT", "MISS"] = Field(
        alias="X-Cache", description="Whether the response was served from cache"
    )
    cache_control: str | None = Field(
        default=None, alias="Cache-Control", description="Cache-control directive"
    )
    vary: str | None = Field(
        default=None,
        alias="Vary",
        description="Request headers the cached response varies on",
    )

CachedAPIView

Bases: APIView

APIView whose endpoints can cache their serialised responses.

Override :meth:build_key to control the cache key, then decorate the relevant endpoint with :func:use_cache. The cache backend is supplied at the application level via :class:~fastapi_views.cache.middleware.CacheMiddleware::

app.add_middleware(CacheMiddleware, backend=RedisCache(...))

class ItemView(CachedAPIView, AsyncRetrieveAPIView):
    cache_key_headers = ["X-Tenant-Id"]

    @use_cache(ttl=60)
    async def retrieve(self, id: UUID) -> Item: ...

This view caches only; it does not handle conditional requests. Use :class:ConditionalCachedAPIView to also revalidate with ETag / Last-Modified and answer 304 Not Modified.

Source code in fastapi_views/cache/view.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
class CachedAPIView(APIView):
    """``APIView`` whose endpoints can cache their serialised responses.

    Override :meth:`build_key` to control the cache key, then decorate the
    relevant endpoint with :func:`use_cache`. The cache backend is supplied at
    the application level via :class:`~fastapi_views.cache.middleware.CacheMiddleware`::

        app.add_middleware(CacheMiddleware, backend=RedisCache(...))

        class ItemView(CachedAPIView, AsyncRetrieveAPIView):
            cache_key_headers = ["X-Tenant-Id"]

            @use_cache(ttl=60)
            async def retrieve(self, id: UUID) -> Item: ...

    This view caches only; it does not handle conditional requests. Use
    :class:`ConditionalCachedAPIView` to also revalidate with ``ETag`` /
    ``Last-Modified`` and answer ``304 Not Modified``.
    """

    cache_key_headers: ClassVar[Sequence[str]] = ()
    vary: ClassVar[Sequence[str]] = ()

    @classmethod
    def get_response_headers(
        cls, action: Action | None = None
    ) -> type[ResponseHeaders] | None:
        if action in ("retrieve", "list"):
            return CacheHeaders
        return None

    @property
    def cache(self) -> Cache:
        return cache

    def build_key(self) -> str:
        """Cache key for the current request.

        Query parameters are sorted for a stable key regardless of ordering.
        Headers listed in :attr:`cache_key_headers` are appended to the key.
        Override for custom schemes.
        """
        request = self.request
        path = request.url.path

        query = urlencode(sorted(parse_qsl(request.url.query)))
        parts = [f"{path}?{query}" if query else path]

        parts.extend(
            f"{name}={value}"
            for name in self.cache_key_headers
            if (value := request.headers.get(name.lower()))
        )

        return hashlib.md5("|".join(parts).encode(), usedforsecurity=False).hexdigest()

    def get_vary_headers(self) -> list[str]:
        """Request header names the cached response varies on.

        Combines :attr:`cache_key_headers` (which key the server-side cache) with
        :attr:`vary`, so downstream caches key on at least the same headers and
        cannot serve one client's response to another. Names are de-duplicated
        case-insensitively, preserving declaration order.
        """
        seen: set[str] = set()
        names: list[str] = []
        for name in (*self.cache_key_headers, *self.vary):
            lowered = name.lower()
            if lowered not in seen:
                seen.add(lowered)
                names.append(name)
        return names

    def get_cache_headers(
        self,
        *,
        hit: bool,
        ttl: int | None,
        cache_control: str | CacheControl | None,
    ) -> dict[str, str]:
        """Build the cache-related response headers (``X-Cache`` / ``Cache-Control`` / ``Vary``)."""
        cache_headers: dict[str, str] = {"X-Cache": "HIT" if hit else "MISS"}

        if isinstance(cache_control, CacheControl):
            # ``ttl`` provides the default freshness when not set explicitly.
            if cache_control.max_age is None and ttl is not None:
                cache_control = replace(cache_control, max_age=ttl)
            directive = cache_control.render()
        elif cache_control is not None:
            directive = cache_control
        elif ttl is not None:
            directive = f"max-age={ttl}"
        else:
            directive = None
        if directive:
            cache_headers["cache-control"] = directive

        vary = self.get_vary_headers()
        if vary:
            cache_headers["Vary"] = ", ".join(vary)
        return cache_headers

build_key() -> str

Cache key for the current request.

Query parameters are sorted for a stable key regardless of ordering. Headers listed in :attr:cache_key_headers are appended to the key. Override for custom schemes.

Source code in fastapi_views/cache/view.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def build_key(self) -> str:
    """Cache key for the current request.

    Query parameters are sorted for a stable key regardless of ordering.
    Headers listed in :attr:`cache_key_headers` are appended to the key.
    Override for custom schemes.
    """
    request = self.request
    path = request.url.path

    query = urlencode(sorted(parse_qsl(request.url.query)))
    parts = [f"{path}?{query}" if query else path]

    parts.extend(
        f"{name}={value}"
        for name in self.cache_key_headers
        if (value := request.headers.get(name.lower()))
    )

    return hashlib.md5("|".join(parts).encode(), usedforsecurity=False).hexdigest()

get_vary_headers() -> list[str]

Request header names the cached response varies on.

Combines :attr:cache_key_headers (which key the server-side cache) with :attr:vary, so downstream caches key on at least the same headers and cannot serve one client's response to another. Names are de-duplicated case-insensitively, preserving declaration order.

Source code in fastapi_views/cache/view.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def get_vary_headers(self) -> list[str]:
    """Request header names the cached response varies on.

    Combines :attr:`cache_key_headers` (which key the server-side cache) with
    :attr:`vary`, so downstream caches key on at least the same headers and
    cannot serve one client's response to another. Names are de-duplicated
    case-insensitively, preserving declaration order.
    """
    seen: set[str] = set()
    names: list[str] = []
    for name in (*self.cache_key_headers, *self.vary):
        lowered = name.lower()
        if lowered not in seen:
            seen.add(lowered)
            names.append(name)
    return names

get_cache_headers(*, hit: bool, ttl: int | None, cache_control: str | CacheControl | None) -> dict[str, str]

Build the cache-related response headers (X-Cache / Cache-Control / Vary).

Source code in fastapi_views/cache/view.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def get_cache_headers(
    self,
    *,
    hit: bool,
    ttl: int | None,
    cache_control: str | CacheControl | None,
) -> dict[str, str]:
    """Build the cache-related response headers (``X-Cache`` / ``Cache-Control`` / ``Vary``)."""
    cache_headers: dict[str, str] = {"X-Cache": "HIT" if hit else "MISS"}

    if isinstance(cache_control, CacheControl):
        # ``ttl`` provides the default freshness when not set explicitly.
        if cache_control.max_age is None and ttl is not None:
            cache_control = replace(cache_control, max_age=ttl)
        directive = cache_control.render()
    elif cache_control is not None:
        directive = cache_control
    elif ttl is not None:
        directive = f"max-age={ttl}"
    else:
        directive = None
    if directive:
        cache_headers["cache-control"] = directive

    vary = self.get_vary_headers()
    if vary:
        cache_headers["Vary"] = ", ".join(vary)
    return cache_headers

ConditionalCachedAPIView

Bases: ConditionalMixin, CachedAPIView

:class:CachedAPIView that also handles conditional requests.

Combines server-side caching with ETag / Last-Modified revalidation: a cache hit can be downgraded to 304 Not Modified when the client is current. Opt into validators exactly as on :class:~fastapi_views.views.mixins.ConditionalMixin (etag = True, last_modified = True, or the manual check_* / not_modified helpers)::

class ItemView(ConditionalCachedAPIView, AsyncRetrieveAPIView):
    etag = True

    @use_cache(ttl=60)
    async def retrieve(self, id: UUID) -> Item: ...

A miss whose response is downgraded to 304 still populates the cache with the full body, so revalidating clients warm the cache for everyone instead of re-running the view on every request.

Source code in fastapi_views/cache/view.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
class ConditionalCachedAPIView(ConditionalMixin, CachedAPIView):
    """:class:`CachedAPIView` that also handles conditional requests.

    Combines server-side caching with ``ETag`` / ``Last-Modified`` revalidation:
    a cache hit can be downgraded to ``304 Not Modified`` when the client is
    current. Opt into validators exactly as on
    :class:`~fastapi_views.views.mixins.ConditionalMixin` (``etag = True``,
    ``last_modified = True``, or the manual ``check_*`` / ``not_modified``
    helpers)::

        class ItemView(ConditionalCachedAPIView, AsyncRetrieveAPIView):
            etag = True

            @use_cache(ttl=60)
            async def retrieve(self, id: UUID) -> Item: ...

    A miss whose response is downgraded to ``304`` still populates the cache
    with the full body, so revalidating clients warm the cache for everyone
    instead of re-running the view on every request.
    """

    def finalize_response(self, response: Response) -> Response:
        """Hand the full body to the cache before any ``304`` downgrade.

        :meth:`~fastapi_views.views.mixins.ConditionalMixin.make_conditional`
        replaces a successful response with an empty ``304`` when the client is
        current, leaving the middleware nothing to persist. Recording the
        pre-downgrade body on the request's cache context keeps the write in the
        middleware (a single write per miss) while caching the representation
        the next non-conditional request should be served.
        """
        ctx: _CacheContext | None = self.request.scope.get(_CACHE_SCOPE_KEY)
        if (
            ctx is not None
            and isinstance(response.body, bytes)
            and response.status_code < _STATUS_CACHEABLE_MAX
        ):
            ctx.body = response.body
        return super().finalize_response(response)

finalize_response(response: Response) -> Response

Hand the full body to the cache before any 304 downgrade.

:meth:~fastapi_views.views.mixins.ConditionalMixin.make_conditional replaces a successful response with an empty 304 when the client is current, leaving the middleware nothing to persist. Recording the pre-downgrade body on the request's cache context keeps the write in the middleware (a single write per miss) while caching the representation the next non-conditional request should be served.

Source code in fastapi_views/cache/view.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def finalize_response(self, response: Response) -> Response:
    """Hand the full body to the cache before any ``304`` downgrade.

    :meth:`~fastapi_views.views.mixins.ConditionalMixin.make_conditional`
    replaces a successful response with an empty ``304`` when the client is
    current, leaving the middleware nothing to persist. Recording the
    pre-downgrade body on the request's cache context keeps the write in the
    middleware (a single write per miss) while caching the representation
    the next non-conditional request should be served.
    """
    ctx: _CacheContext | None = self.request.scope.get(_CACHE_SCOPE_KEY)
    if (
        ctx is not None
        and isinstance(response.body, bytes)
        and response.status_code < _STATUS_CACHEABLE_MAX
    ):
        ctx.body = response.body
    return super().finalize_response(response)

use_cache(ttl: int | None = None, *, cache_control: str | CacheControl | None = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Cache a :class:CachedAPIView endpoint's serialised response.

On a hit the cached body is returned immediately. On a miss the response is produced normally and a cache context is stored on the ASGI scope so :class:~fastapi_views.cache.middleware.CacheMiddleware can inject the cache headers and persist the body.

Source code in fastapi_views/cache/view.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def use_cache(
    ttl: int | None = None,
    *,
    cache_control: str | CacheControl | None = None,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Cache a :class:`CachedAPIView` endpoint's serialised response.

    On a hit the cached body is returned immediately. On a miss the response is
    produced normally and a cache context is stored on the ASGI ``scope`` so
    :class:`~fastapi_views.cache.middleware.CacheMiddleware` can inject the cache
    headers and persist the body.
    """

    def decorator(
        func: Callable[..., Awaitable[Any]],
    ) -> Callable[..., Any]:

        @functools.wraps(func)
        async def wrapper(self: CachedAPIView, *args: Any, **kwargs: Any) -> Any:
            key = self.build_key()
            cached_body = await self.cache.get(key)
            if cached_body is not None:
                return Response(
                    content=cached_body,
                    status_code=self.get_status_code(func.__name__),
                    media_type="application/json",
                    headers=self.get_cache_headers(
                        hit=True, ttl=ttl, cache_control=cache_control
                    ),
                )

            result = await func(self, *args, **kwargs)
            if result is None:
                return None

            self.request.scope[_CACHE_SCOPE_KEY] = _CacheContext(
                key=key,
                ttl=ttl,
                headers=self.get_cache_headers(
                    hit=False, ttl=ttl, cache_control=cache_control
                ),
            )
            return result

        return wrapper

    return decorator

Conditional requests

ConditionalMixin provides the ETag / Last-Modified validators and 304 handling reused by ConditionalCachedAPIView. It can be combined with any view independently of caching and needs no middleware or backend — it works purely from request and response headers.

When you do want caching as well, subclass ConditionalCachedAPIView rather than mixing ConditionalMixin into CachedAPIView by hand: finalize_response below does not call super(), so a hand-rolled combination loses the cache write on a 304-downgraded miss.

fastapi_views.views.mixins.ConditionalMixin

ETag / Last-Modified validators and 304 Not Modified handling.

Two ways to opt in:

  • Automatic — set etag = True for a strong ETag hashed from the serialised body, and/or set last_modified = True and override :meth:get_last_modified. The body is built, then downgraded to a 304 if the client is current.

  • Manual / cheap — compare a cheaply obtained validator (a version column, updated_at) inside the view and short-circuit before building the body, returning :meth:not_modified::

    conditional_requests = True # so the 304 is documented in OpenAPI

    async def retrieve(self, item_id: int) -> Item | Response: item = await self.get_item(item_id) if self.not_modified_since(item.updated_at): return self.not_modified(last_modified=item.updated_at) return item

or, for versioned models, with an ETag::

  async def retrieve(self, item_id: int) -> Item | Response:
      item = await self.get_item(item_id)
      etag = f'"{item.version}"'
      if self.etag_matches(etag):
          return self.not_modified(etag=etag)
      return item
Source code in fastapi_views/views/mixins.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
class ConditionalMixin:
    """ETag / ``Last-Modified`` validators and ``304 Not Modified`` handling.

    Two ways to opt in:

    * **Automatic** — set ``etag = True`` for a strong ETag hashed from the
      serialised body, and/or set ``last_modified = True`` and override
      :meth:`get_last_modified`. The body is built, then downgraded to a ``304``
      if the client is current.
    * **Manual / cheap** — compare a cheaply obtained validator (a version
      column, ``updated_at``) inside the view and short-circuit before building
      the body, returning :meth:`not_modified`::

          conditional_requests = True  # so the 304 is documented in OpenAPI

          async def retrieve(self, item_id: int) -> Item | Response:
              item = await self.get_item(item_id)
              if self.not_modified_since(item.updated_at):
                  return self.not_modified(last_modified=item.updated_at)
              return item

      or, for versioned models, with an ETag::

          async def retrieve(self, item_id: int) -> Item | Response:
              item = await self.get_item(item_id)
              etag = f'"{item.version}"'
              if self.etag_matches(etag):
                  return self.not_modified(etag=etag)
              return item
    """

    request: Request
    response: Response
    etag: bool = False
    last_modified: bool = False
    conditional_requests: bool = False

    def finalize_response(self, response: Response) -> Response:
        return self.make_conditional(response)

    def get_etag(self, body: bytes) -> str | None:
        if self.etag:
            return hashlib.blake2b(body, digest_size=16).hexdigest()
        return None

    @property
    def if_none_match(self) -> str | None:
        """The request's ``If-None-Match`` validator, if sent."""
        return self.request.headers.get("if-none-match")

    @property
    def if_modified_since(self) -> datetime | None:
        """The request's parsed ``If-Modified-Since`` timestamp, if sent."""
        value = self.request.headers.get("if-modified-since")
        return _parse_http_date(value) if value is not None else None

    def etag_matches(self, etag: str) -> bool:
        """Whether ``If-None-Match`` matches ``etag`` (handles ``*`` and lists).

        ``etag`` may be a raw value (e.g. ``str(version)``); it is quoted to a
        valid entity-tag before comparison.
        """
        if_none_match = self.if_none_match
        return if_none_match is not None and _etag_matches(
            if_none_match, _format_etag(etag)
        )

    def not_modified_since(self, last_modified: datetime) -> bool:
        """Whether ``last_modified`` is not newer than ``If-Modified-Since``."""
        since = self.if_modified_since
        return since is not None and _to_utc_seconds(last_modified) <= since

    def not_modified(
        self,
        *,
        etag: str | None = None,
        last_modified: datetime | None = None,
    ) -> Response:
        """Build a ``304 Not Modified`` response, echoing any given validators.

        Return this from a view to skip building and serialising the body once
        you have determined the client's cached copy is still current.
        """
        headers: dict[str, str] = {}
        if etag is not None:
            headers["etag"] = _format_etag(etag)
        if last_modified is not None:
            headers["last-modified"] = format_datetime(
                _to_utc_seconds(last_modified), usegmt=True
            )
        return Response(status_code=HTTP_304_NOT_MODIFIED, headers=headers)

    def set_etag(self, etag: str) -> None:
        """Send ``etag`` as the ``ETag`` header on this request's response.

        A raw value (e.g. ``str(version)``) is quoted to a valid entity-tag.
        """
        self.response.headers["etag"] = _format_etag(etag)

    def set_last_modified(self, last_modified: datetime) -> None:
        """Send ``last_modified`` as the ``Last-Modified`` header on the response."""
        self.response.headers["last-modified"] = format_datetime(
            _to_utc_seconds(last_modified), usegmt=True
        )

    def check_etag(self, etag: str) -> Response | None:
        """Return a ``304`` when the client's copy matches ``etag``.

        Otherwise stamp ``etag`` on the upcoming response and return ``None``,
        so ``return self.check_etag(tag) or item`` skips serialising the body
        when the client is current and sends the validator on the body response.
        """
        if self.etag_matches(etag):
            return self.not_modified(etag=etag)
        self.set_etag(etag)
        return None

    def check_last_modified(self, last_modified: datetime) -> Response | None:
        """``Last-Modified`` counterpart of :meth:`check_etag`."""
        if self.not_modified_since(last_modified):
            return self.not_modified(last_modified=last_modified)
        self.set_last_modified(last_modified)
        return None

    def get_last_modified(self) -> datetime | None:
        return None

    @classmethod
    def supports_conditional_requests(cls) -> bool:
        """Whether the view emits validators (an ETag or ``Last-Modified``)."""
        return cls.etag or cls.last_modified or cls.conditional_requests

    @classmethod
    def _conditional_response_headers(cls) -> dict[str, Any]:
        schema = ConditionalHeaders.get_openapi_headers()
        headers: dict[str, Any] = {}
        if cls.etag or cls.conditional_requests:
            headers["ETag"] = schema["ETag"]
        if cls.last_modified or cls.conditional_requests:
            headers["Last-Modified"] = schema["Last-Modified"]
        return headers

    @classmethod
    def get_conditional_responses(
        cls,
        *,
        action: Action | None = None,  # noqa: ARG003
        status_code: int | None = None,
        methods: Sequence[str] | None = None,
    ) -> dict[int | str, dict[str, Any]]:
        """Document validator headers and a ``304 Not Modified`` response.

        ``ETag`` / ``Last-Modified`` are documented on the success response and,
        for safe methods, a ``304`` is added — but only when the view actually
        emits a validator, so views that set none are not documented with
        headers or a ``304`` they will never return.
        """
        headers = cls._conditional_response_headers()
        if not headers:
            return {}
        responses: dict[int | str, dict[str, Any]] = {}
        if status_code is not None:
            responses[status_code] = {"headers": headers}
        if methods and any(method in _SAFE_METHODS for method in methods):
            responses[HTTP_304_NOT_MODIFIED] = {
                "description": "Not Modified",
                "headers": headers,
            }
        return responses

    def make_conditional(self, response: Response) -> Response:
        """Attach validators and downgrade to 304 when the client is current."""
        if not isinstance(response.body, bytes) or not (
            _SUCCESS_MIN <= response.status_code < _SUCCESS_MAX
        ):
            return response

        etag = self.get_etag(response.body)
        if etag is not None:
            etag = _format_etag(etag)
        last_modified = self.get_last_modified() if self.last_modified else None
        headers = response.headers
        if etag:
            headers["etag"] = etag
        if last_modified is not None:
            headers["last-modified"] = format_datetime(
                _to_utc_seconds(last_modified), usegmt=True
            )

        if self.request.method in _SAFE_METHODS and self._is_not_modified(
            etag, last_modified
        ):
            carried = {
                name: headers[name] for name in _REVALIDATION_HEADERS if name in headers
            }
            return Response(status_code=HTTP_304_NOT_MODIFIED, headers=carried)
        return response

    def _is_not_modified(
        self, etag: str | None, last_modified: datetime | None
    ) -> bool:
        if self.if_none_match is not None:
            # RFC 7232: If-Modified-Since is ignored when If-None-Match is present.
            return etag is not None and self.etag_matches(etag)
        return last_modified is not None and self.not_modified_since(last_modified)

if_none_match: str | None property

The request's If-None-Match validator, if sent.

if_modified_since: datetime | None property

The request's parsed If-Modified-Since timestamp, if sent.

etag_matches(etag: str) -> bool

Whether If-None-Match matches etag (handles * and lists).

etag may be a raw value (e.g. str(version)); it is quoted to a valid entity-tag before comparison.

Source code in fastapi_views/views/mixins.py 
246
247
248
249
250
251
252
253
254
255
def etag_matches(self, etag: str) -> bool:
    """Whether ``If-None-Match`` matches ``etag`` (handles ``*`` and lists).

    ``etag`` may be a raw value (e.g. ``str(version)``); it is quoted to a
    valid entity-tag before comparison.
    """
    if_none_match = self.if_none_match
    return if_none_match is not None and _etag_matches(
        if_none_match, _format_etag(etag)
    )

not_modified_since(last_modified: datetime) -> bool

Whether last_modified is not newer than If-Modified-Since.

Source code in fastapi_views/views/mixins.py 
257
258
259
260
def not_modified_since(self, last_modified: datetime) -> bool:
    """Whether ``last_modified`` is not newer than ``If-Modified-Since``."""
    since = self.if_modified_since
    return since is not None and _to_utc_seconds(last_modified) <= since

not_modified(*, etag: str | None = None, last_modified: datetime | None = None) -> Response

Build a 304 Not Modified response, echoing any given validators.

Return this from a view to skip building and serialising the body once you have determined the client's cached copy is still current.

Source code in fastapi_views/views/mixins.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
def not_modified(
    self,
    *,
    etag: str | None = None,
    last_modified: datetime | None = None,
) -> Response:
    """Build a ``304 Not Modified`` response, echoing any given validators.

    Return this from a view to skip building and serialising the body once
    you have determined the client's cached copy is still current.
    """
    headers: dict[str, str] = {}
    if etag is not None:
        headers["etag"] = _format_etag(etag)
    if last_modified is not None:
        headers["last-modified"] = format_datetime(
            _to_utc_seconds(last_modified), usegmt=True
        )
    return Response(status_code=HTTP_304_NOT_MODIFIED, headers=headers)

set_etag(etag: str) -> None

Send etag as the ETag header on this request's response.

A raw value (e.g. str(version)) is quoted to a valid entity-tag.

Source code in fastapi_views/views/mixins.py 
282
283
284
285
286
287
def set_etag(self, etag: str) -> None:
    """Send ``etag`` as the ``ETag`` header on this request's response.

    A raw value (e.g. ``str(version)``) is quoted to a valid entity-tag.
    """
    self.response.headers["etag"] = _format_etag(etag)

set_last_modified(last_modified: datetime) -> None

Send last_modified as the Last-Modified header on the response.

Source code in fastapi_views/views/mixins.py 
289
290
291
292
293
def set_last_modified(self, last_modified: datetime) -> None:
    """Send ``last_modified`` as the ``Last-Modified`` header on the response."""
    self.response.headers["last-modified"] = format_datetime(
        _to_utc_seconds(last_modified), usegmt=True
    )

check_etag(etag: str) -> Response | None

Return a 304 when the client's copy matches etag.

Otherwise stamp etag on the upcoming response and return None, so return self.check_etag(tag) or item skips serialising the body when the client is current and sends the validator on the body response.

Source code in fastapi_views/views/mixins.py 
295
296
297
298
299
300
301
302
303
304
305
def check_etag(self, etag: str) -> Response | None:
    """Return a ``304`` when the client's copy matches ``etag``.

    Otherwise stamp ``etag`` on the upcoming response and return ``None``,
    so ``return self.check_etag(tag) or item`` skips serialising the body
    when the client is current and sends the validator on the body response.
    """
    if self.etag_matches(etag):
        return self.not_modified(etag=etag)
    self.set_etag(etag)
    return None

check_last_modified(last_modified: datetime) -> Response | None

Last-Modified counterpart of :meth:check_etag.

Source code in fastapi_views/views/mixins.py 
307
308
309
310
311
312
def check_last_modified(self, last_modified: datetime) -> Response | None:
    """``Last-Modified`` counterpart of :meth:`check_etag`."""
    if self.not_modified_since(last_modified):
        return self.not_modified(last_modified=last_modified)
    self.set_last_modified(last_modified)
    return None

supports_conditional_requests() -> bool classmethod

Whether the view emits validators (an ETag or Last-Modified).

Source code in fastapi_views/views/mixins.py 
317
318
319
320
@classmethod
def supports_conditional_requests(cls) -> bool:
    """Whether the view emits validators (an ETag or ``Last-Modified``)."""
    return cls.etag or cls.last_modified or cls.conditional_requests

get_conditional_responses(*, action: Action | None = None, status_code: int | None = None, methods: Sequence[str] | None = None) -> dict[int | str, dict[str, Any]] classmethod

Document validator headers and a 304 Not Modified response.

ETag / Last-Modified are documented on the success response and, for safe methods, a 304 is added — but only when the view actually emits a validator, so views that set none are not documented with headers or a 304 they will never return.

Source code in fastapi_views/views/mixins.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
@classmethod
def get_conditional_responses(
    cls,
    *,
    action: Action | None = None,  # noqa: ARG003
    status_code: int | None = None,
    methods: Sequence[str] | None = None,
) -> dict[int | str, dict[str, Any]]:
    """Document validator headers and a ``304 Not Modified`` response.

    ``ETag`` / ``Last-Modified`` are documented on the success response and,
    for safe methods, a ``304`` is added — but only when the view actually
    emits a validator, so views that set none are not documented with
    headers or a ``304`` they will never return.
    """
    headers = cls._conditional_response_headers()
    if not headers:
        return {}
    responses: dict[int | str, dict[str, Any]] = {}
    if status_code is not None:
        responses[status_code] = {"headers": headers}
    if methods and any(method in _SAFE_METHODS for method in methods):
        responses[HTTP_304_NOT_MODIFIED] = {
            "description": "Not Modified",
            "headers": headers,
        }
    return responses

make_conditional(response: Response) -> Response

Attach validators and downgrade to 304 when the client is current.

Source code in fastapi_views/views/mixins.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
def make_conditional(self, response: Response) -> Response:
    """Attach validators and downgrade to 304 when the client is current."""
    if not isinstance(response.body, bytes) or not (
        _SUCCESS_MIN <= response.status_code < _SUCCESS_MAX
    ):
        return response

    etag = self.get_etag(response.body)
    if etag is not None:
        etag = _format_etag(etag)
    last_modified = self.get_last_modified() if self.last_modified else None
    headers = response.headers
    if etag:
        headers["etag"] = etag
    if last_modified is not None:
        headers["last-modified"] = format_datetime(
            _to_utc_seconds(last_modified), usegmt=True
        )

    if self.request.method in _SAFE_METHODS and self._is_not_modified(
        etag, last_modified
    ):
        carried = {
            name: headers[name] for name in _REVALIDATION_HEADERS if name in headers
        }
        return Response(status_code=HTTP_304_NOT_MODIFIED, headers=carried)
    return response

The validator headers documented in OpenAPI come from this model; ConditionalMixin contributes only the ones the view can actually emit.

fastapi_views.views.mixins.ConditionalHeaders

Bases: ResponseHeaders

Validator headers attached to conditional (304-capable) responses.

Source code in fastapi_views/views/mixins.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class ConditionalHeaders(ResponseHeaders):
    """Validator headers attached to conditional (``304``-capable) responses."""

    etag: str | None = Field(
        default=None,
        alias="ETag",
        description="Validator for the returned representation",
    )
    last_modified: str | None = Field(
        default=None,
        alias="Last-Modified",
        description="Time the representation was last modified",
        json_schema_extra={"format": "http-date"},
    )

Middleware and backends

fastapi_views.cache.middleware.CacheMiddleware

ASGI middleware that writes cache entries and injects cache headers.

On a cache miss the :func:~fastapi_views.cache.view.use_cache decorator stores a :class:_CacheContext on the shared ASGI scope. This middleware reads it back to inject the cache headers into the outgoing response and, once the full body has been sent, persist it to the backend::

app.add_middleware(CacheMiddleware, backend=InMemoryCache())

This is the single place cache entries are written. What gets stored is the body the context carries if the view recorded one (see :class:_CacheContext), otherwise the successful response's outgoing body.

Source code in fastapi_views/cache/middleware.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class CacheMiddleware:
    """ASGI middleware that writes cache entries and injects cache headers.

    On a cache miss the :func:`~fastapi_views.cache.view.use_cache` decorator
    stores a :class:`_CacheContext` on the shared ASGI ``scope``. This middleware
    reads it back to inject the cache headers into the outgoing response and,
    once the full body has been sent, persist it to the backend::

        app.add_middleware(CacheMiddleware, backend=InMemoryCache())

    This is the single place cache entries are written. What gets stored is the
    body the context carries if the view recorded one (see
    :class:`_CacheContext`), otherwise the successful response's outgoing body.
    """

    def __init__(self, app: ASGIApp, *, backend: CacheBackend | None = None) -> None:
        self.app = app
        if backend is not None:
            cache.init_backend(backend)

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        body_chunks: list[bytes] = []
        cacheable = False

        async def send_wrapper(message: Any) -> None:
            nonlocal cacheable

            ctx: _CacheContext | None = scope.get(_CACHE_SCOPE_KEY)
            if ctx is None:
                await send(message)
                return

            if message["type"] == "http.response.start":
                cacheable = message["status"] < _STATUS_CACHEABLE_MAX
                headers = MutableHeaders(scope=message)
                for name, value in ctx.headers.items():
                    headers[name] = value
            elif message["type"] == "http.response.body" and (
                chunk := message.get("body", b"")
            ):
                body_chunks.append(chunk)

            await send(message)

            if message["type"] == "http.response.body" and not message.get(
                "more_body", False
            ):
                body = ctx.resolve_body(body_chunks, cacheable=cacheable)
                if body:
                    await cache.set(ctx.key, body, ttl=ctx.ttl)

        await self.app(scope, receive, send_wrapper)

fastapi_views.cache.cache is the shared Cache instance the views, the middleware and the @cache decorator all use; CacheMiddleware(backend=...) (or cache.init_backend(...)) installs its backend.

fastapi_views.cache.cache.Cache

Source code in fastapi_views/cache/cache.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class Cache:
    def __init__(self, backend: CacheBackend | None = None) -> None:
        self._backend: CacheBackend | None = backend

    @property
    def backend(self) -> CacheBackend:
        if self._backend is None:
            raise ValueError("Cache backend not set")
        return self._backend

    def init_backend(self, backend: CacheBackend) -> None:
        self._backend = backend

    async def get(self, key: KeyT) -> EncodableT | None:
        return await self.backend.get(key)

    async def set(self, key: KeyT, value: EncodableT, ttl: int | None = None) -> None:
        return await self.backend.set(key, value, ttl=ttl)

    async def delete(self, key: KeyT) -> None:
        await self.backend.delete(key)

    async def pop(self, key: KeyT) -> EncodableT | None:
        return await self.backend.pop(key)

    def _format_key(
        self, key: Callable[..., KeyT] | KeyT, *args: Any, **kwargs: Any
    ) -> KeyT:
        if callable(key):
            return key(*args, **kwargs)
        if isinstance(key, str) and re.match(_KEY_PATTERN, key):
            return key.format(*args, **kwargs)
        return key

    def __call__(
        self,
        key: KeyT | Callable[..., KeyT],
        ttl: int | None = None,
    ) -> AsyncDecorator:

        def decorator(
            func: Callable[P, Awaitable[T]],
        ) -> Callable[P, Awaitable[T]]:
            return_type = _resolve_return_type(func)
            adapter = _get_type_adapter(return_type)

            @wraps(func)
            async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
                cache_key = self._format_key(key, *args, **kwargs)
                raw = await self.get(cache_key)
                if raw is None:
                    result = await func(*args, **kwargs)
                    await self.set(cache_key, adapter.dump_json(result), ttl=ttl)
                    return result
                return adapter.validate_json(raw)

            return wrapper

        return decorator

backend: CacheBackend property

init_backend(backend: CacheBackend) -> None

Source code in fastapi_views/cache/cache.py 
57
58
def init_backend(self, backend: CacheBackend) -> None:
    self._backend = backend

get(key: KeyT) -> EncodableT | None async

Source code in fastapi_views/cache/cache.py 
60
61
async def get(self, key: KeyT) -> EncodableT | None:
    return await self.backend.get(key)

set(key: KeyT, value: EncodableT, ttl: int | None = None) -> None async

Source code in fastapi_views/cache/cache.py 
63
64
async def set(self, key: KeyT, value: EncodableT, ttl: int | None = None) -> None:
    return await self.backend.set(key, value, ttl=ttl)

delete(key: KeyT) -> None async

Source code in fastapi_views/cache/cache.py 
66
67
async def delete(self, key: KeyT) -> None:
    await self.backend.delete(key)

pop(key: KeyT) -> EncodableT | None async

Source code in fastapi_views/cache/cache.py 
69
70
async def pop(self, key: KeyT) -> EncodableT | None:
    return await self.backend.pop(key)

__call__(key: KeyT | Callable[..., KeyT], ttl: int | None = None) -> AsyncDecorator

Source code in fastapi_views/cache/cache.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __call__(
    self,
    key: KeyT | Callable[..., KeyT],
    ttl: int | None = None,
) -> AsyncDecorator:

    def decorator(
        func: Callable[P, Awaitable[T]],
    ) -> Callable[P, Awaitable[T]]:
        return_type = _resolve_return_type(func)
        adapter = _get_type_adapter(return_type)

        @wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
            cache_key = self._format_key(key, *args, **kwargs)
            raw = await self.get(cache_key)
            if raw is None:
                result = await func(*args, **kwargs)
                await self.set(cache_key, adapter.dump_json(result), ttl=ttl)
                return result
            return adapter.validate_json(raw)

        return wrapper

    return decorator

Keys and values are str | bytes (KeyT / EncodableT in fastapi_views.cache.backends).

fastapi_views.cache.backends.abc.CacheBackend

Bases: ABC

Source code in fastapi_views/cache/backends/abc.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class CacheBackend(ABC):
    @abstractmethod
    async def get(self, key: KeyT) -> EncodableT | None:
        raise NotImplementedError

    @abstractmethod
    async def set(self, key: KeyT, value: EncodableT, ttl: int | None = None) -> None:
        raise NotImplementedError

    @abstractmethod
    async def delete(self, key: KeyT) -> None:
        raise NotImplementedError

    @abstractmethod
    async def pop(self, key: KeyT) -> EncodableT | None:
        raise NotImplementedError

fastapi_views.cache.backends.memory.InMemoryCache

Bases: CacheBackend

Source code in fastapi_views/cache/backends/memory.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
class InMemoryCache(CacheBackend):
    def __init__(self, default_ttl: int | None = None) -> None:
        self._default_ttl = default_ttl
        self._data: dict[KeyT, ExpiringItem] = {}

    async def get(self, key: KeyT) -> EncodableT | None:
        item = self._data.get(key)
        if item is None:
            return None
        if item.expires_at and time.monotonic() > item.expires_at:
            self._data.pop(key, None)
            return None
        return item.value

    async def set(self, key: KeyT, value: EncodableT, ttl: int | None = None) -> None:
        ttl = ttl or self._default_ttl
        expires_at = (time.monotonic() + ttl) if ttl else None
        self._data[key] = ExpiringItem(value, expires_at)

    async def delete(self, key: KeyT) -> None:
        self._data.pop(key, None)

    async def pop(self, key: KeyT) -> EncodableT | None:
        item = self._data.pop(key, None)
        if item is None:
            return None
        if item.expires_at and time.monotonic() > item.expires_at:
            return None
        return item.value

fastapi_views.cache.backends.redis.RedisCache

Bases: CacheBackend

Source code in fastapi_views/cache/backends/redis.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class RedisCache(CacheBackend):
    def __init__(self, client: Redis) -> None:
        self.redis = client

    async def get(self, key: KeyT) -> EncodableT | None:
        return await self.redis.get(key)

    async def set(self, key: KeyT, value: EncodableT, ttl: int | None = None) -> None:
        await self.redis.set(key, value, ex=ttl)

    async def delete(self, key: KeyT) -> None:
        await self.redis.delete(key)

    async def pop(self, key: KeyT) -> EncodableT | None:
        return await self.redis.getdel(key)