​
RedisStorage

class RedisStorage(BaseKeyValueStorage):

A concrete implementation of the :obj:BaseCacheStorage using Redis as the backend. This is suitable for distributed cache systems that require persistence and high availability.

​
init

def __init__(
    self,
    sid: str,
    url: str = 'redis://localhost:6379',
    loop: Optional[asyncio.AbstractEventLoop] = None,
    **kwargs
):

Initializes the RedisStorage instance with the provided URL and options.

Parameters:

  • sid (str): The ID for the storage instance to identify the record space.
  • url (str): The URL for connecting to the Redis server. **kwargs: Additional keyword arguments for Redis client configuration.

​
enter

def __enter__(self):

​
exit

def __exit__(
    self,
    exc_type,
    exc,
    tb
):

​
_create_client

def _create_client(self, **kwargs):

Creates the Redis client with the provided URL and options.

​
client

def client(self):

Returns:

redis.asyncio.Redis: The Redis client instance.

​
save

def save(
    self,
    records: List[Dict[str, Any]],
    expire: Optional[int] = None
):

Saves a batch of records to the key-value storage system.

​
load

def load(self):

Returns:

List[Dict[str, Any]]: A list of dictionaries, where each dictionary represents a stored record.

​
clear

def clear(self):

Removes all records from the key-value storage system.

​
_run_async

def _run_async(self, coro):