Trixy7/trixy_core/utils/decorators/caching.py

# -*- coding: utf-8 -*-
"""
Caching-Decorators mit TTL und LRU-Unterstützung.

Bietet flexible Caching-Optionen für Funktionsergebnisse mit
Zeit- und Größenlimits.
"""

from __future__ import annotations

import asyncio
import functools
import hashlib
import json
import threading
import time
from collections import OrderedDict
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Callable, Generic, Hashable, ParamSpec, TypeVar


P = ParamSpec("P")
R = TypeVar("R")


@dataclass
class CacheEntry(Generic[R]):
    """
    Ein Eintrag im Cache.

    Speichert den Wert zusammen mit Metadaten wie Erstellungszeit
    und Ablaufzeit.
    """

    value: R
    created_at: float = field(default_factory=time.time)
    expires_at: float | None = None
    hit_count: int = 0
    last_accessed: float = field(default_factory=time.time)

    def is_expired(self) -> bool:
        """Prüft, ob der Eintrag abgelaufen ist."""
        if self.expires_at is None:
            return False
        return time.time() > self.expires_at

    def access(self) -> R:
        """Markiert den Eintrag als zugegriffen und gibt den Wert zurück."""
        self.hit_count += 1
        self.last_accessed = time.time()
        return self.value


@dataclass
class CacheStats:
    """
    Statistiken für einen Cache.

    Bietet Einblick in die Cache-Nutzung und Effektivität.
    """

    hits: int = 0
    misses: int = 0
    size: int = 0
    max_size: int = 0
    evictions: int = 0
    expired: int = 0
    created_at: datetime = field(default_factory=datetime.now)

    @property
    def total_requests(self) -> int:
        """Gesamtzahl der Anfragen."""
        return self.hits + self.misses

    @property
    def hit_rate(self) -> float:
        """Cache-Trefferquote in Prozent."""
        if self.total_requests == 0:
            return 0.0
        return (self.hits / self.total_requests) * 100

    @property
    def miss_rate(self) -> float:
        """Cache-Fehlschlagquote in Prozent."""
        return 100.0 - self.hit_rate

    def record_hit(self) -> None:
        """Zeichnet einen Cache-Hit auf."""
        self.hits += 1

    def record_miss(self) -> None:
        """Zeichnet einen Cache-Miss auf."""
        self.misses += 1

    def record_eviction(self) -> None:
        """Zeichnet eine Eviction auf."""
        self.evictions += 1

    def record_expiration(self) -> None:
        """Zeichnet eine Expiration auf."""
        self.expired += 1

    def reset(self) -> None:
        """Setzt alle Statistiken zurück."""
        self.hits = 0
        self.misses = 0
        self.evictions = 0
        self.expired = 0

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert die Statistiken in ein Dictionary."""
        return {
            "hits": self.hits,
            "misses": self.misses,
            "total_requests": self.total_requests,
            "hit_rate": self.hit_rate,
            "size": self.size,
            "max_size": self.max_size,
            "evictions": self.evictions,
            "expired": self.expired,
        }


class LRUCache(Generic[R]):
    """
    Thread-sicherer LRU-Cache mit TTL-Unterstützung.

    Implementiert Least-Recently-Used Eviction-Strategie mit
    optionaler Zeit-basierter Expiration.
    """

    def __init__(
        self,
        max_size: int = 128,
        ttl: float | None = None,
    ) -> None:
        """
        Initialisiert den Cache.

        Args:
            max_size: Maximale Anzahl von Einträgen.
            ttl: Time-to-Live in Sekunden (None = kein Ablauf).
        """
        self._cache: OrderedDict[Hashable, CacheEntry[R]] = OrderedDict()
        self._max_size = max_size
        self._ttl = ttl
        self._lock = threading.RLock()
        self._stats = CacheStats(max_size=max_size)

    @property
    def stats(self) -> CacheStats:
        """Gibt die Cache-Statistiken zurück."""
        with self._lock:
            self._stats.size = len(self._cache)
            return self._stats

    def _make_key(self, args: tuple, kwargs: dict) -> str:
        """
        Erstellt einen Cache-Schlüssel aus Funktionsargumenten.

        Args:
            args: Positionsargumente.
            kwargs: Schlüsselwortargumente.

        Returns:
            Hash-String als Schlüssel.
        """
        key_data = {
            "args": args,
            "kwargs": sorted(kwargs.items()),
        }

        try:
            key_str = json.dumps(key_data, sort_keys=True, default=str)
        except (TypeError, ValueError):
            # Fallback für nicht-serialisierbare Objekte
            key_str = str(key_data)

        return hashlib.md5(key_str.encode()).hexdigest()

    def get(self, key: Hashable) -> tuple[bool, R | None]:
        """
        Holt einen Wert aus dem Cache.

        Args:
            key: Cache-Schlüssel.

        Returns:
            Tupel (gefunden, Wert).
        """
        with self._lock:
            if key not in self._cache:
                self._stats.record_miss()
                return False, None

            entry = self._cache[key]

            # Expiration prüfen
            if entry.is_expired():
                del self._cache[key]
                self._stats.record_expiration()
                self._stats.record_miss()
                return False, None

            # An das Ende verschieben (LRU)
            self._cache.move_to_end(key)
            self._stats.record_hit()

            return True, entry.access()

    def set(self, key: Hashable, value: R) -> None:
        """
        Speichert einen Wert im Cache.

        Args:
            key: Cache-Schlüssel.
            value: Zu speichernder Wert.
        """
        with self._lock:
            # Eviction falls nötig
            while len(self._cache) >= self._max_size:
                self._cache.popitem(last=False)
                self._stats.record_eviction()

            # Eintrag erstellen
            expires_at = time.time() + self._ttl if self._ttl else None
            entry = CacheEntry(value=value, expires_at=expires_at)
            self._cache[key] = entry

    def invalidate(self, key: Hashable) -> bool:
        """
        Invalidiert einen Cache-Eintrag.

        Args:
            key: Cache-Schlüssel.

        Returns:
            True wenn der Eintrag existierte.
        """
        with self._lock:
            if key in self._cache:
                del self._cache[key]
                return True
            return False

    def clear(self) -> int:
        """
        Löscht alle Cache-Einträge.

        Returns:
            Anzahl der gelöschten Einträge.
        """
        with self._lock:
            count = len(self._cache)
            self._cache.clear()
            self._stats.reset()
            return count

    def cleanup_expired(self) -> int:
        """
        Entfernt abgelaufene Einträge.

        Returns:
            Anzahl der entfernten Einträge.
        """
        with self._lock:
            expired_keys = [
                key for key, entry in self._cache.items()
                if entry.is_expired()
            ]

            for key in expired_keys:
                del self._cache[key]
                self._stats.record_expiration()

            return len(expired_keys)

    def __len__(self) -> int:
        """Gibt die aktuelle Cache-Größe zurück."""
        with self._lock:
            return len(self._cache)

    def __contains__(self, key: Hashable) -> bool:
        """Prüft, ob ein Schlüssel im Cache ist."""
        with self._lock:
            if key not in self._cache:
                return False
            entry = self._cache[key]
            return not entry.is_expired()


# Globale Cache-Registry
_caches: dict[str, LRUCache] = {}


def get_cache(name: str) -> LRUCache | None:
    """
    Gibt einen Cache anhand seines Namens zurück.

    Args:
        name: Name des Caches.

    Returns:
        Der Cache oder None.
    """
    return _caches.get(name)


def cache_clear(name: str | None = None) -> int:
    """
    Löscht einen oder alle Caches.

    Args:
        name: Optional spezifischer Cache-Name.

    Returns:
        Anzahl der gelöschten Einträge.
    """
    if name:
        cache = _caches.get(name)
        if cache:
            return cache.clear()
        return 0

    total = 0
    for cache in _caches.values():
        total += cache.clear()
    return total


def cached(
    ttl: float | None = None,
    max_size: int = 128,
    key_func: Callable[..., Hashable] | None = None,
    cache_name: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Decorator für Funktions-Caching mit TTL und LRU.

    Args:
        ttl: Time-to-Live in Sekunden (None = kein Ablauf).
        max_size: Maximale Cache-Größe.
        key_func: Optionale Funktion zur Schlüssel-Generierung.
        cache_name: Optionaler Name für den Cache (für Zugriff).

    Returns:
        Decorator-Funktion.

    Example:
        @cached(ttl=300, max_size=100)
        def expensive_calculation(x, y):
            return x ** y

        @cached(key_func=lambda user_id: user_id)
        def get_user(user_id):
            return database.get(user_id)
    """
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        name = cache_name or func.__qualname__
        cache: LRUCache[R] = LRUCache(max_size=max_size, ttl=ttl)
        _caches[name] = cache

        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            # Schlüssel generieren
            if key_func:
                key = key_func(*args, **kwargs)
            else:
                key = cache._make_key(args, kwargs)

            # Cache-Lookup
            found, value = cache.get(key)
            if found:
                return value  # type: ignore

            # Wert berechnen und cachen
            result = func(*args, **kwargs)
            cache.set(key, result)
            return result

        # Cache-Zugriff über die Funktion
        wrapper.cache = cache  # type: ignore
        wrapper.cache_clear = cache.clear  # type: ignore
        wrapper.cache_stats = lambda: cache.stats  # type: ignore

        return wrapper

    return decorator


def async_cached(
    ttl: float | None = None,
    max_size: int = 128,
    key_func: Callable[..., Hashable] | None = None,
    cache_name: str | None = None,
) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """
    Async-Decorator für Funktions-Caching.

    Wie @cached, aber für async Funktionen.

    Args:
        ttl: Time-to-Live in Sekunden.
        max_size: Maximale Cache-Größe.
        key_func: Optionale Funktion zur Schlüssel-Generierung.
        cache_name: Optionaler Name für den Cache.

    Returns:
        Decorator-Funktion.

    Example:
        @async_cached(ttl=60)
        async def fetch_data(url):
            async with aiohttp.ClientSession() as session:
                async with session.get(url) as response:
                    return await response.json()
    """
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        name = cache_name or func.__qualname__
        cache: LRUCache[R] = LRUCache(max_size=max_size, ttl=ttl)
        _caches[name] = cache

        # Lock für async-sichere Operationen
        _async_lock = asyncio.Lock()

        @functools.wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            if key_func:
                key = key_func(*args, **kwargs)
            else:
                key = cache._make_key(args, kwargs)

            # Erst synchron prüfen
            found, value = cache.get(key)
            if found:
                return value  # type: ignore

            # Async-Lock für die Berechnung
            async with _async_lock:
                # Nochmal prüfen (könnte sich geändert haben)
                found, value = cache.get(key)
                if found:
                    return value  # type: ignore

                result = await func(*args, **kwargs)
                cache.set(key, result)
                return result

        wrapper.cache = cache  # type: ignore
        wrapper.cache_clear = cache.clear  # type: ignore
        wrapper.cache_stats = lambda: cache.stats  # type: ignore

        return wrapper

    return decorator


def memoize(func: Callable[P, R]) -> Callable[P, R]:
    """
    Einfacher Memoization-Decorator ohne TTL oder Größenlimit.

    Cacht Ergebnisse permanent für die Lebensdauer des Prozesses.

    Args:
        func: Die zu cachende Funktion.

    Returns:
        Gecachte Funktion.

    Example:
        @memoize
        def fibonacci(n):
            if n < 2:
                return n
            return fibonacci(n - 1) + fibonacci(n - 2)
    """
    cache: dict[Hashable, R] = {}

    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        # Einfacher Schlüssel für hashbare Argumente
        key = (args, tuple(sorted(kwargs.items())))

        if key in cache:
            return cache[key]

        result = func(*args, **kwargs)
        cache[key] = result
        return result

    wrapper.cache = cache  # type: ignore
    wrapper.cache_clear = cache.clear  # type: ignore

    return wrapper