"""Database-backed response cache with stale fallback support.""" from __future__ import annotations import hashlib import os import re from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Any from sqlalchemy import create_engine, text from sqlalchemy.engine import Engine from app.core.config import settings from app.utils.serialization import dumps_json, loads_json, to_jsonable UTC = timezone.utc TABLE_NAME_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$") @dataclass(frozen=True) class CacheEntry: key: str namespace: str payload: Any source: str created_at: datetime expires_at: datetime row_count: int | None @property def expired(self) -> bool: return self.expires_at <= datetime.now(UTC) class ResponseCache: def __init__(self, database_url: str, enabled: bool = True) -> None: self.enabled = enabled self.database_url = self._normalize_database_url(database_url) self.table_name = self._validate_table_name(settings.cache_table_name) self.engine: Engine | None = None if self.enabled: self._init_engine() def _normalize_database_url(self, database_url: str) -> str: if database_url.startswith("mysql://"): return database_url.replace("mysql://", "mysql+pymysql://", 1) return database_url def _validate_table_name(self, table_name: str) -> str: value = table_name.strip() if not TABLE_NAME_PATTERN.fullmatch(value): raise ValueError(f"Invalid CACHE_TABLE_NAME: {table_name!r}") return value def _init_engine(self) -> None: if self.database_url.startswith("sqlite:///"): path = self.database_url.removeprefix("sqlite:///") dirname = os.path.dirname(path) if dirname: os.makedirs(dirname, exist_ok=True) connect_args: dict[str, Any] = {} if settings.database_ssl and self.database_url.startswith("mysql+pymysql://"): connect_args["ssl"] = {"ssl": True} self.engine = create_engine( self.database_url, pool_pre_ping=True, future=True, connect_args=connect_args, ) with self.engine.begin() as conn: conn.execute( text( f""" CREATE TABLE IF NOT EXISTS {self.table_name} ( cache_key VARCHAR(191) PRIMARY KEY, namespace VARCHAR(96) NOT NULL, request_hash VARCHAR(64) NOT NULL, payload_json LONGTEXT NOT NULL, source_name VARCHAR(160) NOT NULL, row_count INTEGER NULL, created_at DATETIME NOT NULL, expires_at DATETIME NOT NULL ) """ ) ) def make_key(self, namespace: str, params: dict[str, Any]) -> str: canonical = dumps_json( { "version": settings.cache_schema_version, "namespace": namespace, "params": to_jsonable(params), } ) digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() return f"{namespace}:{digest}" def get(self, key: str, allow_expired: bool = False) -> CacheEntry | None: if not self.enabled or self.engine is None: return None with self.engine.begin() as conn: row = conn.execute( text( f""" SELECT cache_key, namespace, payload_json, source_name, row_count, created_at, expires_at FROM {self.table_name} WHERE cache_key = :key """ ), {"key": key}, ).mappings().first() if not row: return None entry = CacheEntry( key=str(row["cache_key"]), namespace=str(row["namespace"]), payload=loads_json(str(row["payload_json"])), source=str(row["source_name"]), row_count=row["row_count"], created_at=self._as_utc(row["created_at"]), expires_at=self._as_utc(row["expires_at"]), ) if entry.expired and not allow_expired: return None return entry def set( self, key: str, namespace: str, payload: Any, source: str, ttl_seconds: int, row_count: int | None = None, ) -> CacheEntry | None: if not self.enabled or self.engine is None: return None now = datetime.now(UTC) expires_at = now + timedelta(seconds=ttl_seconds) request_hash = key.rsplit(":", 1)[-1] with self.engine.begin() as conn: conn.execute( text( f""" DELETE FROM {self.table_name} WHERE cache_key = :key """ ), {"key": key}, ) conn.execute( text( f""" INSERT INTO {self.table_name} ( cache_key, namespace, request_hash, payload_json, source_name, row_count, created_at, expires_at ) VALUES ( :key, :namespace, :request_hash, :payload_json, :source_name, :row_count, :created_at, :expires_at ) """ ), { "key": key, "namespace": namespace, "request_hash": request_hash, "payload_json": dumps_json(payload), "source_name": source, "row_count": row_count, "created_at": now.replace(tzinfo=None), "expires_at": expires_at.replace(tzinfo=None), }, ) return CacheEntry(key, namespace, payload, source, now, expires_at, row_count) def purge_expired(self) -> int: if not self.enabled or self.engine is None: return 0 now = datetime.now(UTC).replace(tzinfo=None) with self.engine.begin() as conn: result = conn.execute( text(f"DELETE FROM {self.table_name} WHERE expires_at < :now"), {"now": now}, ) return int(result.rowcount or 0) def purge_all(self) -> int: if not self.enabled or self.engine is None: return 0 with self.engine.begin() as conn: result = conn.execute(text(f"DELETE FROM {self.table_name}")) return int(result.rowcount or 0) def stats(self) -> dict[str, Any]: if not self.enabled or self.engine is None: return {"enabled": False} with self.engine.begin() as conn: total = conn.execute(text(f"SELECT COUNT(*) FROM {self.table_name}")).scalar_one() expired = conn.execute( text(f"SELECT COUNT(*) FROM {self.table_name} WHERE expires_at < :now"), {"now": datetime.now(UTC).replace(tzinfo=None)}, ).scalar_one() return { "enabled": True, "backend": self.database_url.split(":", 1)[0], "total_entries": int(total), "expired_entries": int(expired), } def _as_utc(self, value: Any) -> datetime: if isinstance(value, datetime): if value.tzinfo is None: return value.replace(tzinfo=UTC) return value.astimezone(UTC) return datetime.fromisoformat(str(value)).replace(tzinfo=UTC) cache = ResponseCache(settings.database_url, enabled=settings.cache_enabled)