EncryptedPydanticRedisStorage#

class safir.redis.EncryptedPydanticRedisStorage(*, datatype, redis, encryption_key, key_prefix='')#

Bases: PydanticRedisStorage[S]

A Pydantic-based Redis store that encrypts data.

Parameters:

  • datatype (type[TypeVar(S, bound= BaseModel)]) – The class of object being stored (a Pydantic model).

  • redis (Redis) – A Redis client configured to talk to the backend store.

  • encryption_key (str) – Encryption key. Must be a ~cryptography.fernet.Fernet key. Generate a key with Fernet.generate_key().decode().

  • key_prefix (str, default: '') – A prefix to prepend to all Redis keys. Setting a prefix ensures that all objects stored through this class share the same redis key prefix. This is useful if multiple storage classes share the same Redis database. The keys that you use to store objects (e.g. the store method) and the key you get back from the scan method do not include this prefix.

Methods Summary

delete(key)

Delete a stored object.

delete_all(pattern)

Delete all stored objects.

get(key)

Retrieve a stored object.

scan(pattern)

Scan Redis for a given key pattern, returning each key.

store(key, obj[, lifetime])

Store an object.

Methods Documentation

async delete(key)#

Delete a stored object.

Parameters:

key (str) – The key to delete.

Returns:

True if the key was found and deleted, False otherwise.

Return type:

bool

async delete_all(pattern)#

Delete all stored objects.

Parameters:

pattern (str) – Glob pattern matching the keys to purge, such as oidc:*.

Return type:

None

async get(key)#

Retrieve a stored object.

Parameters:

key (str) – The key for the object.

Returns:

The deserialized object or None if no such object could be found.

Return type:

Any or None

Raises:

DeserializeError – Raised if the stored object could not be decrypted or deserialized.

async scan(pattern)#

Scan Redis for a given key pattern, returning each key.

Parameters:

pattern (str) – Key pattern to scan for.

Yields:

str – Each key matching that pattern.

Return type:

AsyncIterator[str]

async store(key, obj, lifetime=None)#

Store an object.

Parameters:

  • key (str) – The key for the object.

  • obj (TypeVar(S, bound= BaseModel)) – The object to store.

  • lifetime (Optional[int], default: None) – The object lifetime in seconds. The object should expire from the data store after that many seconds after the current time. Pass None if the object should not expire.

Return type:

None