Usage¶

radish requires that you declare schemas for your cached resources using Pydantic. This ensures a consistent interface, making your code easier to grasp, and allowing static type checkers like mypy to spot errors without executing your code.

Setting up your interface¶

You can configure your interface to Redis by setting a number of Resource() attributes on a subclass of Interface.

from datetime import datetime
from typing import List, Tuple

from pydantic import BaseModel
import radish


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


class Order(BaseModel):
    id: int
    item_id: int
    timestamp: datetime
    customer: Customer


class Radish(radish.Interface):
    customers = radish.Resource(Customer, key="id")
    orders = radish.Resource(Order, key="id")

In the above example, there are two models, Customer and Order. Each will be stored in the default Redis database (0), and keyed on their ID fields.

Note

Multiple resources can use the same Redis database, as the keys will be prefixed with the model name.

If you want to store the same model in the same database under different contexts, you will need to manually set the namespace to avoid conflicts:

class Person(BaseModel):
    id: int
    email: str

class Radish(radish.Interface):
    users = radish.Resource(Person, key="id", prefix="user")
    administrators = radish.Resource(Person, key="id", prefix="administrator")

Alternatively, you can store resources in different databases:

class Person(BaseModel):
    id: int
    email: str

class Radish(radish.Interface):
    users = radish.Resource(Person, key="id", db=0)
    administrators = radish.Resource(Person, key="id", db=1)

Warning

Redis support for multiple databases is deprecated and not supported in Redis Cluster. It’s recommended that you do not specify the database on your resources, unless you absolutely must.

Connecting to Redis¶

To connect to Redis, use the Interface as an asynchronous context manager:

async with Radish(address="redis://redis") as cache:
    ...

The Interface passes its arguments through to aioredis.create_redis_pool. You can override the create_redis_pool by passing the connection_factory option:

async with Radish(connection_factory=custom_redis_pool, address="redis://redis") as cache:
    ...

Caching records¶

To store a record in the cache, you can pass it to the save() method of its corresponding manager:

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

user = User(id=1, name="Bob")

async with Radish(address="redis://redis") as cache:
    await cache.users.save(user)

You can save multiple records with save():

bob = User(id=1, name="Bob")
fred = User(id=2, name="Bob")

async with Radish(address="redis://redis") as cache:
    await cache.users.save(fred, bob)

By default, save() will update existing records:

async with Radish(address="redis://redis") as cache:
    user: User = await cache.users.get(1)
    user.name = "Fred"
    await cache.users.save(user)

But this behaviour can be disabled:

async with Radish(address="redis://redis") as cache:
    await cache.users.save(user, allow_update=False)

You can also set how long you want to keep the cached record for (in seconds):

async with Radish(address="redis://redis") as cache:
    await cache.users.save(user, expire=15.0)

The create() method provides a shorthand for initialising the model instance and caching at the same time:

async with Radish(address="redis://redis") as cache:
    user: User = await cache.users.create(id=1, name="frank")

You can set a record to expire or delete it directly, using the expire() and delete() methods.

async with Radish(address="redis://redis") as cache:
    await cache.users.expire(user1, 15.0)
    await cache.users.delete(user2)

Retrieving from the cache¶

The get() method allows you to retrieve a record by ID:

async with Radish(address="redis://redis") as cache:
    user = await cache.users.get(1)

If you aren’t sure whether the record exists or not, you can set a default:

async with Radish(address="redis://redis") as cache:
    user = await cache.users.get(1, None)

You can also pass a model instance directly to get() to find the current cached version:

async with Radish(address="redis://redis") as cache:
    cached_user = await cache.users.get(user)

The resource manager can be treated as an asynchronous iterable over all records:

async with Radish(address="redis://redis") as cache:
    all_users = [user async for user in cache.users]

And the filter method allows you to find particular records:

async with Radish(address="redis://redis") as cache:
    async for user in cache.users.filter(name="fred"):
        print(user)

See filter() for more ways to filter on record fields.

Note

Filtering is done on the client-side, and so an iteration over a filter will still retrieve every record from the cache under the hood. This is due to how scan operations work in Redis.

Radish

Navigation

Related Topics