Source code for runnel.record

from pydantic import BaseModel

from runnel.exceptions import Misconfigured

[docs]class Record(BaseModel): def __init__(self, *args, **kwargs): """ This class is used to specify structured data types to store in streams. It is a specialized version of pydantic.BaseModel, see for more information. Examples -------- >>> class Order(Record): ... order_id: int ... created_at: datetime ... amount: int ... item_ids: List[int] Records can be nested arbitrarily, e.g. `items: List[Item]` where `Item` is another record. They will be serialized (according to the serializer/compressor settings) and stored as arbitrary bytes as a single value in a Redis stream entry. Alternatively, you can use the native Redis stream key/value pairs by setting primitive=True, e.g.: >>> class UserAction(Record, primitive=True): ... user_id: int ... type: str This allows you to benefit from optimisations such as delta compression (see, at the cost of not supporting nested values. Only int, float, bool, str, and bytes are allowed. """ super().__init__(*args, **kwargs) def __init_subclass__(cls, *, primitive=False, **kwargs): super().__init_subclass__(**kwargs) cls._primitive = primitive if primitive: allowed = {int, float, bool, str, bytes} if not set(cls.__annotations__.values()) <= allowed: raise Misconfigured(f"Primitive record fields must be in {allowed}")