Storing Pydantic objects in Redis

Safir provides a PydanticRedisStorage class that can conveniently store Pydantic objects in Redis. The advantage of using Pydantic models with Redis is that data usage is validated during developed and validated at runtime. Safir also provides a subclass, EncryptedPydanticRedisStorage, that encrypts data before storing it in Redis.

Important

To use safir.redis, you must install Safir with its redis extra: that is, safir[redis].

Basic usage

PydanticRedisStorage works with an asyncio Redis client (redis.asyncio.client.Redis) from redis-py. You supply the storage class type as a parameter. All objects stored in the storage must be instances of this type, and objects in Redis will be parsed and validated as this type. If you need to store multiple types of Pydantic objects in Redis, you can create separate instances of PydanticRedisStorage for each type.

Here is a basic set up:

import redis.asyncio as redis
from pydantic import BaseModel, Field
from safir.redis import PydanticRedisStorage


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


redis_client = redis.Redis("redis://localhost:6379/0")
mymodel_storage = PydanticRedisStorage(MyModel, redis_client)

Use the store method to store a Pydantic object in Redis with a specific key:

await mymodel_storage.store("people:1", MyModel(id=1, name="Drew"))
await mymodel_storage.store("people:2", MyModel(id=2, name="Blake"))

You can get objects back by their key:

drew = await mymodel_storage.get("people:1")
assert drew.id == 1
assert drew.name == "Drew"

You can scan for all keys that match a pattern with the scan method:

async for key in mymodel_storage.scan("people:*"):
    m = await mymodel_storage.get(key)
    print(m)

You can also delete objects by key using the delete method:

await mymodel_storage.delete("people:1")

It’s also possible to delete all objects at once with keys that match a pattern using the delete_all method:

await mymodel_storage.delete_all("people:*")

Encrypting data with EncryptedPydanticRedisStorage

EncryptedPydanticRedisStorage is a subclass of PydanticRedisStorage that encrypts data before storing it in Redis. It also decrypts data after retrieving it from Redis (assuming the key is correct).

To use EncryptedPydanticRedisStorage you must provide a Fernet key. A convenient way to generate a Fernet key is with the cryptography.fernet.Fernet.generate_key function from the cryptography Python package:

from cryptography.fernet import Fernet

print(Fernet.generate_key().decode())

Conventionally, you’ll store this key in a persistent secret store, such as 1Password or Vault (see the Phalanx documentation) and then make this key available to your application through an environment variable to your configuration class. Then pass the key’s value to EncryptedPydanticRedisStorage with the encryption_key parameter:

from safir.redis import EncryptedPydanticRedisStorage

redis_client = redis.Redis(config.redis_url)
mymodel_storage = EncryptedPydanticRedisStorage(
    datatype=MyModel,
    redis=redis_client,
    encryption_key=config.encryption_key,
)

Once set up, you can interact with this storage class exactly like PydanticRedisStorage, except that all data is encrypted in Redis.

Multi-tentant storage with key prefixes

PydanticRedisStorage and EncryptedPydanticRedisStorage both allow you to specify a prefix string that is automatically applied to the keys of an objects stored through that class. Once set, your application does not need to worry about consistently using this prefix.

A common use case for a key prefix is if multiple stores share the same Redis database. Each PydanticRedisStorage instance works with a specific Pydantic model type, so if your application needs to store multiple types of objects in Redis, you can use multiple instances of PydanticRedisStorage with different key prefixes.

class PetModel(BaseModel):
    id: int
    name: str
    age: int


class CustomerModel(BaseModel):
    id: int
    name: str
    email: str


redis_client = redis.Redis(config.redis_url)

pet_store = PydanticRedisStorage(
    datatype=PetModel,
    redis=redis_client,
    key_prefix="pet:",
)
customer_store = PydanticRedisStorage(
    datatype=CustomerModel,
    redis=redis_client,
    key_prefix="customer:",
)