Repository: Significant-Gravitas/Auto-GPT
File: autogpt_platform/backend/backend/util/cache.py (Python)

Code before:
import time
from dataclasses import dataclass
from functools import wraps
from typing import Any, Callable, ParamSpec, Protocol, TypeVar, cast, runtime_checkable

from redis import ConnectionPool, Redis

from backend.util.retry import conn_retry
from backend.util.settings import Settings

P = ParamSpec("P")
R = TypeVar("R")
R_co = TypeVar("R_co", covariant=True)
T = TypeVar("T")

logger = logging.getLogger(__name__)
settings = Settings()

# RECOMMENDED REDIS CONFIGURATION FOR PRODUCTION:
# Configure Redis with the following settings for optimal caching performance:
#   maxmemory-policy allkeys-lru    # Evict least recently used keys when memory limit reached
#   maxmemory 2gb                   # Set memory limit (adjust based on your needs)
#   save ""                         # Disable persistence if using Redis purely for caching

# Create a dedicated Redis connection pool for caching (binary mode for pickle)
_cache_pool: ConnectionPool | None = None


@conn_retry("Redis", "Acquiring cache connection pool")
def _get_cache_pool() -> ConnectionPool:
    """Get or create a connection pool for cache operations."""
    global _cache_pool
    if _cache_pool is None:
        _cache_pool = ConnectionPool(
            host=settings.config.redis_host,
            port=settings.config.redis_port,
            password=settings.config.redis_password or None,
            decode_responses=False,  # Binary mode for pickle
            max_connections=50,
            socket_keepalive=True,
            socket_connect_timeout=5,
            retry_on_timeout=True,
        )
    return _cache_pool


redis = Redis(connection_pool=_get_cache_pool())


@dataclass
class CachedValue:

Review comment (performance):
nit: this is slightly confusing when used like this in a `cache` module; I'd recommend `import functools` + `@functools.cache`

Code after:
import time
from dataclasses import dataclass
from functools import cache, wraps
from typing import Any, Callable, ParamSpec, Protocol, TypeVar, cast, runtime_checkable

from redis import ConnectionPool, Redis

from backend.util.retry import conn_retry
from backend.util.settings import Settings

P = ParamSpec("P")
R = TypeVar("R")
R_co = TypeVar("R_co", covariant=True)
T = TypeVar("T")

logger = logging.getLogger(__name__)
settings = Settings()

# RECOMMENDED REDIS CONFIGURATION FOR PRODUCTION:
# Configure Redis with the following settings for optimal caching performance:
#   maxmemory-policy allkeys-lru    # Evict least recently used keys when memory limit reached
#   maxmemory 2gb                   # Set memory limit (adjust based on your needs)
#   save ""                         # Disable persistence if using Redis purely for caching


@cache
def _get_cache_pool() -> ConnectionPool:
    """Get or create a connection pool for cache operations (lazy, thread-safe)."""
    return ConnectionPool(
        host=settings.config.redis_host,
        port=settings.config.redis_port,
        password=settings.config.redis_password or None,
        decode_responses=False,  # Binary mode for pickle
        max_connections=50,
        socket_keepalive=True,
        socket_connect_timeout=5,
        retry_on_timeout=True,
    )


@cache
@conn_retry("Redis", "Acquiring cache connection")
def _get_redis() -> Redis:
    """
    Get the lazily-initialized Redis client for shared cache operations.
    Uses @cache for thread-safe singleton behavior - connection is only
    established when first accessed, allowing services that only use
    in-memory caching to work without Redis configuration.
    """
    r = Redis(connection_pool=_get_cache_pool())
    r.ping()  # Verify connection