locate-the-news / backend / app / services / cache.py
cache.py
Raw
"""
Cache Service - Simple in-memory cache for API responses.
Can be extended to use Redis later.
"""
from typing import Any, Optional
from datetime import datetime, timedelta
from functools import wraps

from ..utils.logger import setup_logger

logger = setup_logger(__name__)

# Simple in-memory cache
_cache: dict = {}
_cache_ttl: dict = {}


def get(key: str) -> Optional[Any]:
    """
    Retrieves a value from the cache if it exists and hasn't expired.
    
    Args:
        key: Cache key
        
    Returns:
        Cached value or None if not found or expired
    """
    if key not in _cache:
        return None
    
    # Check if expired
    if key in _cache_ttl:
        if datetime.now() > _cache_ttl[key]:
            # Expired, remove it
            del _cache[key]
            del _cache_ttl[key]
            return None
    
    return _cache[key]


def set(key: str, value: Any, ttl_seconds: int = 86400) -> None:
    """
    Stores a value in the cache with a TTL.
    
    Args:
        key: Cache key
        value: Value to cache
        ttl_seconds: Time to live in seconds (default: 24 hours)
    """
    _cache[key] = value
    _cache_ttl[key] = datetime.now() + timedelta(seconds=ttl_seconds)
    logger.debug(f"Cached value for key: {key} (TTL: {ttl_seconds}s)")


def clear(key: Optional[str] = None) -> None:
    """
    Clears cache entries.
    
    Args:
        key: Specific key to clear, or None to clear all
    """
    if key:
        if key in _cache:
            del _cache[key]
        if key in _cache_ttl:
            del _cache_ttl[key]
        logger.debug(f"Cleared cache for key: {key}")
    else:
        _cache.clear()
        _cache_ttl.clear()
        logger.debug("Cleared all cache")


def cached(ttl_seconds: int = 86400):
    """
    Decorator to cache function results.
    
    Usage:
        @cached(ttl_seconds=1800)
        def expensive_function(arg1, arg2):
            ...
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Create cache key from function name and arguments
            cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            
            # Try to get from cache
            cached_value = get(cache_key)
            if cached_value is not None:
                logger.debug(f"Cache hit for {func.__name__}")
                return cached_value
            
            # Call function and cache result
            result = func(*args, **kwargs)
            set(cache_key, result, ttl_seconds)
            return result
        
        return wrapper
    return decorator