67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
"""Shared Redis read-through for small configuration catalogues.
|
|
|
|
Callers authorize before entering this helper and include any user/query scope
|
|
in the key. Only successful reads are stored; Redis failures keep the original
|
|
database/file-backed path available.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Awaitable, Callable
|
|
from typing import Any
|
|
|
|
from fastapi import Request
|
|
from pydantic import BaseModel, ValidationError
|
|
|
|
from app.gateway import api_read_cache
|
|
|
|
|
|
def catalog_key(*parts: Any) -> str:
|
|
return json.dumps(parts, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
|
|
|
|
|
|
async def cached_catalog[T](
|
|
request: Request,
|
|
collection: str,
|
|
key: str,
|
|
load: Callable[[], Awaitable[T]],
|
|
model: type[BaseModel] | None = None,
|
|
) -> T:
|
|
config = getattr(request.app.state, "config", None)
|
|
version = await api_read_cache.get_cache_version(config, collection)
|
|
cached = await api_read_cache.get_cached_json(config, collection, key, version=version)
|
|
|
|
def decode(value):
|
|
try:
|
|
return model.model_validate(value) if model is not None else value
|
|
except (ValidationError, TypeError):
|
|
# A stale schema or malformed entry must not break the read path.
|
|
return None
|
|
|
|
if cached is not None:
|
|
decoded = decode(cached)
|
|
if decoded is not None:
|
|
return decoded
|
|
cached, fill_token = await api_read_cache.wait_for_cached_json_or_claim_fill(
|
|
config, collection, key, version=version,
|
|
)
|
|
try:
|
|
if cached is not None:
|
|
decoded = decode(cached)
|
|
if decoded is not None:
|
|
return decoded
|
|
value = await load()
|
|
await api_read_cache.set_cached_json(config, collection, key, value, expected_version=version)
|
|
return value
|
|
finally:
|
|
await api_read_cache.release_cache_fill(config, collection, key, fill_token)
|
|
|
|
|
|
async def invalidate_catalog(request: Request, collection: str) -> None:
|
|
# Writes while caching is disabled must also invalidate old generations,
|
|
# otherwise re-enabling the collection could resurrect stale values.
|
|
await api_read_cache.invalidate_cached_collection(
|
|
getattr(request.app.state, "config", None), collection, force=True,
|
|
)
|