Skip to content

redis_cachedb

langroid/cachedb/redis_cachedb.py

RedisCacheConfig

Bases: CacheDBConfig

Configuration model for RedisCache.

RedisCache(config)

Bases: CacheDB

Redis implementation of the CacheDB.

Parameters:

Name Type Description Default
config RedisCacheConfig

The configuration to use.

required
Source code in langroid/cachedb/redis_cachedb.py
def __init__(self, config: RedisCacheConfig):
    """
    Initialize a RedisCache with the given config.

    Args:
        config (RedisCacheConfig): The configuration to use.
    """
    self.config = config
    load_dotenv()

    if self.config.fake:
        self.pool = fakeredis.FakeStrictRedis()  # type: ignore
    else:
        redis_password = os.getenv("REDIS_PASSWORD")
        redis_host = os.getenv("REDIS_HOST")
        redis_port = os.getenv("REDIS_PORT")
        if None in [redis_password, redis_host, redis_port]:
            if not RedisCache._warned_password:
                logger.warning(
                    """REDIS_PASSWORD, REDIS_HOST, REDIS_PORT not set in .env file,
                    using fake redis client"""
                )
                RedisCache._warned_password = True
            self.pool = fakeredis.FakeStrictRedis()  # type: ignore
        else:
            self.pool = redis.ConnectionPool(  # type: ignore
                host=redis_host,
                port=redis_port,
                password=redis_password,
                max_connections=500,
                socket_timeout=5,
                socket_keepalive=True,
                retry_on_timeout=True,
                health_check_interval=30,
            )

redis_client()

Cleanly open and close a redis client, avoids max clients exceeded error

Source code in langroid/cachedb/redis_cachedb.py
@contextmanager  # type: ignore
def redis_client(self) -> AbstractContextManager[T]:  # type: ignore
    """Cleanly open and close a redis client, avoids max clients exceeded error"""
    if isinstance(self.pool, fakeredis.FakeStrictRedis):
        yield self.pool
    else:
        client: T = redis.Redis(connection_pool=self.pool)
        try:
            yield client
        finally:
            client.close()

clear()

Clear keys from current db.

Source code in langroid/cachedb/redis_cachedb.py
def clear(self) -> None:
    """Clear keys from current db."""
    with self.redis_client() as client:  # type: ignore
        client.flushdb()

clear_all()

Clear all keys from all dbs.

Source code in langroid/cachedb/redis_cachedb.py
def clear_all(self) -> None:
    """Clear all keys from all dbs."""
    with self.redis_client() as client:  # type: ignore
        client.flushall()

store(key, value)

Store a value associated with a key.

Parameters:

Name Type Description Default
key str

The key under which to store the value.

required
value Any

The value to store.

required
Source code in langroid/cachedb/redis_cachedb.py
def store(self, key: str, value: Any) -> None:
    """
    Store a value associated with a key.

    Args:
        key (str): The key under which to store the value.
        value (Any): The value to store.
    """
    with self.redis_client() as client:  # type: ignore
        try:
            client.set(key, json.dumps(value))
        except redis.exceptions.ConnectionError:
            logger.warning("Redis connection error, not storing key/value")
            return None

retrieve(key)

Retrieve the value associated with a key.

Parameters:

Name Type Description Default
key str

The key to retrieve the value for.

required

Returns:

Type Description
Dict[str, Any] | str | None

dict|str|None: The value associated with the key.

Source code in langroid/cachedb/redis_cachedb.py
def retrieve(self, key: str) -> Dict[str, Any] | str | None:
    """
    Retrieve the value associated with a key.

    Args:
        key (str): The key to retrieve the value for.

    Returns:
        dict|str|None: The value associated with the key.
    """
    with self.redis_client() as client:  # type: ignore
        try:
            value = client.get(key)
        except redis.exceptions.ConnectionError:
            logger.warning("Redis connection error, returning None")
            return None
        return json.loads(value) if value else None

delete_keys(keys)

Delete the keys from the cache.

Parameters:

Name Type Description Default
keys List[str]

The keys to delete.

required
Source code in langroid/cachedb/redis_cachedb.py
def delete_keys(self, keys: List[str]) -> None:
    """
    Delete the keys from the cache.

    Args:
        keys (List[str]): The keys to delete.
    """
    with self.redis_client() as client:  # type: ignore
        try:
            client.delete(*keys)
        except redis.exceptions.ConnectionError:
            logger.warning("Redis connection error, not deleting keys")
            return None

delete_keys_pattern(pattern)

Delete the keys matching the pattern from the cache.

Parameters:

Name Type Description Default
prefix str

The pattern to match.

required
Source code in langroid/cachedb/redis_cachedb.py
def delete_keys_pattern(self, pattern: str) -> None:
    """
    Delete the keys matching the pattern from the cache.

    Args:
        prefix (str): The pattern to match.
    """
    with self.redis_client() as client:  # type: ignore
        try:
            keys = client.keys(pattern)
            if len(keys) > 0:
                client.delete(*keys)
        except redis.exceptions.ConnectionError:
            logger.warning("Redis connection error, not deleting keys")
            return None