API Reference

Radish Interface

class radish.interface.Interface(*, connection_factory=<function create_redis_pool>, **redis_settings)

A specification of resources to be managed in Redis. Should be subclassed.

Resources are declared by setting Resource() instance as class attributes on subclasses of Interface:

class User(BaseModel):
    id: int
    name: str

class Redis(radish.Interface):
    users = radish.Resource(User, key="id", db=0)

Connection to Redis can then be made by using the Interface subclass as an asynchronous context manager:

async with Redis("redis://redis") as redis:
    user = await redis.users.create(id=1, name="bob")

You are free to define custom methods and properties on the subclass, and these will all be available on the yielded context:

class Redis(radish.Interface):
    users = radish.Resource(User, key="id", db=0)

    async def user_list(self) -> List[User]:
        return [user async for user in self.users]

async with Redis("redis://redis") as redis:
    user_list = await redis.user_list()

However the class attribute _meta is reserved, and will raise RadishError if set.

class radish.interface.InterfaceMeta

Metaclass for the Interface class.

Collects resource manager descriptors assigned as class variables.

Radish Resource

radish.resource.Resource(model, key, db=0, prefix=None)

Declare a new resource type on an Interface subclass.

Usage:

class User(BaseModel):
    id: int
    name: str

class Redis(radish.Interface):
    users = radish.Resource(User, key="id", db=0)

This creates a record manager on the Interface subclass, which allows to save, retrieve and iterate records of this type.

Record types are differentiated by namespacing the key. For example:

await redis.users.create(id=1, name="bob")

This would store a serialized user instance at key "User-1" in the Redis database. This is hidden internally if your only interface to Redis is via radish, but is important to note if the cached data must be accessed elsewhere.

By default the namespace is taken from the model’s class name. If this would cause conflicts then the prefix can be set explicitly as follows:

class Redis(radish.Interface):
    users = radish.Resource(User, key="id", db=0, prefix="_radish_user")

Parameters

• model (Type[~Model]) – Subclass of pydantic.BaseModel which described the resource stored on this manager.

• key (Union[str, Callable[[~Model], SupportsStr]]) – Either the attribute of the model instance used to generate the key in Redis, or a function which will generate a key from an instance.

• db (int) – The Redis database in which to store the resource records.

• prefix (Optional[str]) – The prefix used to namespace the resource in Redis.

Return type

_ResourceManager[~Model]

Returns

The descriptor for the resource manager.

When using the Interface as a client, declared resources behave as resource managers with access to the methods declared on _ResourceManager.

class radish.resource._ResourceManager(descriptor, connection_factory=<function create_redis_pool>, **connection_kwargs)

A manager for interacting with a particular resource type in Redis.

Should not be instantiated directly, but instead declared via Resource() on a Interface subclass.

async create(*args, **kwargs)

Create a new record in Redis, and return the model instance.

Usage:

user: User = await redis.users.create(id=1, name="bob")

Arguments are passed directly through to the resource model to create the instance.

Raises

RadishError: if a record already exists with this identifier.

Return type

~Model

Returns

The newly created and cached model instance.

async save(*instances, allow_update=True, expire=None)

Store one or more model instances in the Redis cache.

await redis.users.save(User(id=1, name="bob"), User(id=2, name="frank"))

Parameters

• instances (~Model) – The set of model instances to store in the cache.

• allow_update (bool) – Whether to allow updates to existing records.

• expire (Optional[SupportsFloat]) – The number of seconds in which to expire te records.

Return type

None

async get(instance, default=<object object>)

Retrieve a record from the cache.

Either accepts lookup by instance ID:

user = await redis.users.get(request.data["id"])

Or by instance itself:

user = await redis.users.get(user)

Default values can be provided for the case where the record does not exist:

if not await redis.users.get(user, None):
    await redis.users.save(user)

Parameters

• instance (Union[~Model, SupportsStr]) – Either the key to look up a record, or a model instance itself.

• default – Optional default value to return if the record does not exist. Follows the same semantics as getattr.

Return type

~Model

Returns

The cached model instance, or default if provided and the record is not found.

Raises

RadishKeyError if no record is found and no default is provided.

async delete(instance)

Delete a record from the cache.

await redis.users.delete(bad_user)

Parameters

instance (Union[~Model, SupportsStr]) – Either the key of the record to delete, or a model instance itself.

Raises

RadishKeyError if no matching record exists.

Return type

None

async expire(instance, expire)

Set a record to expire.

await redis.users.expire(old_user, 10.0)

Parameters

• instance (Union[~Model, SupportsStr]) – Either the key of the record to expire, or a model instance itself.

• expire (SupportsFloat) – The number of seconds in which to expire the record.

Raises

RadishKeyError if no matching record exists.

Return type

None

__aiter__()

Iterate all cached records for this resource type.

async for user in redis.users:
    await send(user.id)

Return type

AsyncIterator[~Model]

filter(**filter_kwargs)

Asynchronously iterate over records matching given criteria.

Usage:

from radish.filter import like, within, contains

# Exact field matching:
results = [result async for result in redis.users.filter(name="bob")]

# String matching (UNIX glob):
results = [result async for result in redis.users.filter(name=like("bob *"))]

# String matching (Regular expression):
results = [
    result
    async for result in redis.users.filter(
        name=like("^[Bb]ob.*$"), regex=True
    )
]

# Container field matching:
results = [
    result
    async for result in redis.users.filter(friends=contains(source_user))
]

# Field within container:
results = [
    result
    async for result in redis.users.filter(age=within(range(25, 30)))
]

# Combining filters can be achieved using the & and | operators:
results = [
    result
    async for result in redis.users.filter(
        age=within(range(0, 10)) | within(range(70, 100))
    )
]

Parameters

filter_kwargs (Any) – The keys should match the resource model’s fields and the values can either be values for exact matching, or be one of the filter terms provided in radish.filter (see above).

Return type

AsyncIterator[~Model]

Returns

An async iterator over the matching records.