80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from collections.abc import Mapping, Sequence
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from fastapi import Request, Response
|
|
from fastapi.encoders import jsonable_encoder
|
|
|
|
|
|
def _strip_keys(value: Any, excluded_keys: set[str]) -> Any:
|
|
if isinstance(value, Mapping):
|
|
return {
|
|
key: _strip_keys(item, excluded_keys)
|
|
for key, item in value.items()
|
|
if key not in excluded_keys
|
|
}
|
|
if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
|
|
return [_strip_keys(item, excluded_keys) for item in value]
|
|
return value
|
|
|
|
|
|
def build_etag(payload: Any, *, exclude_keys: set[str] | None = None) -> str:
|
|
encoded = jsonable_encoder(payload)
|
|
if exclude_keys:
|
|
encoded = _strip_keys(encoded, exclude_keys)
|
|
serialized = json.dumps(encoded, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
|
digest = hashlib.sha1(serialized.encode("utf-8")).hexdigest()
|
|
return f'"{digest}"'
|
|
|
|
|
|
def is_matching_etag(request: Request, etag: str) -> bool:
|
|
header = request.headers.get("if-none-match")
|
|
if not header:
|
|
return False
|
|
|
|
for candidate in header.split(","):
|
|
if candidate.strip() in {etag, "*"}:
|
|
return True
|
|
return False
|
|
|
|
|
|
def apply_cache_headers(
|
|
response: Response,
|
|
*,
|
|
cache_control: str,
|
|
etag: str | None = None,
|
|
last_modified: datetime | None = None,
|
|
) -> None:
|
|
response.headers["Cache-Control"] = cache_control
|
|
response.headers["Vary"] = "Cookie, Authorization"
|
|
if etag is not None:
|
|
response.headers["ETag"] = etag
|
|
if last_modified is not None:
|
|
response.headers["Last-Modified"] = last_modified.strftime("%a, %d %b %Y %H:%M:%S GMT")
|
|
|
|
|
|
def set_no_store(response: Response) -> None:
|
|
apply_cache_headers(response, cache_control="no-store")
|
|
|
|
|
|
def set_private_revalidate(response: Response, *, etag: str | None = None, last_modified: datetime | None = None) -> None:
|
|
apply_cache_headers(
|
|
response,
|
|
cache_control="private, max-age=0, must-revalidate",
|
|
etag=etag,
|
|
last_modified=last_modified,
|
|
)
|
|
|
|
|
|
def set_public_immutable(response: Response, *, etag: str | None = None, last_modified: datetime | None = None) -> None:
|
|
apply_cache_headers(
|
|
response,
|
|
cache_control="public, max-age=31536000, immutable",
|
|
etag=etag,
|
|
last_modified=last_modified,
|
|
)
|