redis_cachedb

langroid/cachedb/redis_cachedb.py

RedisCacheConfig

Bases: BaseModel

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]:
            logger.warning(
                """REDIS_PASSWORD, REDIS_HOST, REDIS_PORT not set in .env file,
                using fake redis client"""
            )
            self.pool = fakeredis.FakeStrictRedis()  # type: ignore
        else:
            self.pool = redis.ConnectionPool(  # type: ignore
                host=redis_host,
                port=redis_port,
                password=redis_password,
                max_connections=50,
                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