api/core/caching.py

import time
from functools import wraps
from typing import Dict, Any, Callable

_cache: Dict[str, Any] = {}

CACHE_DURATION = 900  # 15 minutes - default
LOGIN_CACHE_DURATION = 3600  # 1 hour - user login sessions and auth data
USER_DATA_CACHE_DURATION = 1800  # 30 minutes - user profile data
WATCHLIST_STATS_CACHE_DURATION = 600  # 10 minutes - watchlist statistics


def cache_result(duration: int = CACHE_DURATION) -> Callable:
    """
    Decorator to cache function results with configurable duration.
    
    Args:
        duration: Cache duration in seconds (default: 15 minutes)
    
    Returns:
        Decorated function with caching capability
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs):
            cache_key = f"{func.__module__}.{func.__name__}:{str(args)}:{str(sorted(kwargs.items()))}"
            
            # Check if cached result exists and is still valid
            if cache_key in _cache:
                cached_data, timestamp = _cache[cache_key]
                if time.time() - timestamp < duration:
                    return cached_data
            
            # Execute function and cache result
            result = func(*args, **kwargs)
            _cache[cache_key] = (result, time.time())
            return result
        return wrapper
    return decorator


def cache_user_data(duration: int = USER_DATA_CACHE_DURATION) -> Callable:
    """Cache user profile data with longer TTL."""
    return cache_result(duration)


def cache_login_data(duration: int = LOGIN_CACHE_DURATION) -> Callable:
    """Cache login session data with even longer TTL."""
    return cache_result(duration)


def cache_watchlist_stats(duration: int = WATCHLIST_STATS_CACHE_DURATION) -> Callable:
    """Cache watchlist statistics with moderate TTL."""
    return cache_result(duration)


def clear_user_cache(user_id: int) -> None:
    """
    Clear all cache entries related to a specific user_id.
    
    Args:
        user_id: The user ID whose cache entries should be cleared
    """
    global _cache
    user_id_str = str(user_id)
    keys_to_remove = [key for key in _cache if user_id_str in key]
    for key in keys_to_remove:
        del _cache[key]


def clear_old_cache(max_age: int = 1800) -> int:
    """
    Clear cache entries older than the specified age.
    
    Args:
        max_age: Maximum age in seconds (default: 30 minutes)
    
    Returns:
        Number of cache entries cleared
    """
    global _cache
    current_time = time.time()
    keys_to_remove = []
    
    for key, (data, timestamp) in _cache.items():
        if current_time - timestamp > max_age:
            keys_to_remove.append(key)
    
    for key in keys_to_remove:
        del _cache[key]
    
    return len(keys_to_remove)


def get_cache_stats() -> Dict[str, Any]:
    """
    Get statistics about the current cache state.
    
    Returns:
        Dictionary containing cache statistics
    """
    current_time = time.time()
    total_entries = len(_cache)
    
    if total_entries == 0:
        return {
            "total_entries": 0,
            "oldest_entry_age": 0,
            "newest_entry_age": 0,
            "average_age": 0
        }
    
    ages = [current_time - timestamp for _, timestamp in _cache.values()]
    
    return {
        "total_entries": total_entries,
        "oldest_entry_age": max(ages),
        "newest_entry_age": min(ages),
        "average_age": sum(ages) / len(ages)
    }