cbh/api/common/services/cache/cache.py

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,
    )


@background_task()
async def invalidate_line_check_cache() -> None:
    await settings.DB_CLIENT.caches.delete_many({"namespace": "linecheck"})