leocrm/app/core/cache.py

"""Redis cache wrapper."""

from __future__ import annotations

import json
from typing import Any

import redis.asyncio as aioredis

from app.core.auth import get_redis


def get_cache() -> aioredis.Redis:
    """Get the cache Redis client (delegates to the global ``get_redis()`` singleton)."""
    return get_redis()


async def cache_get(key: str) -> Any | None:
    """Get a value from cache."""
    r = get_cache()
    raw = await r.get(key)
    if raw is None:
        return None
    return json.loads(raw)


async def cache_set(key: str, value: Any, ttl: int = 300) -> None:
    """Set a value in cache with TTL."""
    r = get_cache()
    await r.setex(key, ttl, json.dumps(value))


async def cache_delete(key: str) -> None:
    """Delete a key from cache."""
    r = get_cache()
    await r.delete(key)


async def cache_flush_pattern(pattern: str) -> None:
    """Delete all keys matching a pattern."""
    r = get_cache()
    keys = await r.keys(pattern)
    if keys:
        await r.delete(*keys)