Spaces:
Sleeping
Sleeping
| import hashlib | |
| from datetime import datetime, timedelta | |
| from typing import Any | |
| from pydantic import BaseModel | |
| from cbh.api.common.models import CacheEntryModel | |
| from cbh.core.config import settings | |
| from cbh.core.wrappers import background_task | |
| def build_cache_key(*parts: Any) -> str: | |
| raw = "|".join("" if part is None else str(part) for part in parts) | |
| return hashlib.sha256(raw.encode("utf-8")).hexdigest() | |
| def _serialize(value: Any) -> Any: | |
| if isinstance(value, BaseModel): | |
| return value.model_dump(mode="json") | |
| if isinstance(value, list): | |
| return [_serialize(item) for item in value] | |
| if isinstance(value, dict): | |
| return {key: _serialize(item) for key, item in value.items()} | |
| return value | |
| async def get_cache(namespace: str, key: str) -> Any | None: | |
| entry = await settings.DB_CLIENT.caches.find_one({"namespace": namespace, "key": key}) | |
| if not entry: | |
| return None | |
| deleted_at = entry.get("deletedAt") | |
| if isinstance(deleted_at, datetime) and deleted_at < datetime.now(): | |
| return None | |
| return entry.get("value") | |
| async def set_cache( | |
| namespace: str, | |
| key: str, | |
| value: Any, | |
| ttl_seconds: int | None = None, | |
| ) -> None: | |
| ttl = ttl_seconds if ttl_seconds is not None else settings.LINE_CHECK_CACHE_TTL_SECONDS | |
| document = CacheEntryModel( | |
| namespace=namespace, | |
| key=key, | |
| value=_serialize(value), | |
| ).to_mongo() | |
| document["deletedAt"] = datetime.now() + timedelta(seconds=ttl) | |
| await settings.DB_CLIENT.caches.update_one( | |
| {"namespace": namespace, "key": key}, | |
| {"$set": document}, | |
| upsert=True, | |
| ) | |
| async def invalidate_line_check_cache() -> None: | |
| await settings.DB_CLIENT.caches.delete_many({"namespace": "linecheck"}) | |