beanis.odm.documents
Document
class Document(LazyModel, SettersInterface, InheritanceInterface,
OtherGettersInterface)
Document Mapping class for Redis.
Uses Redis Hashes for storage by default, with support for secondary indexes, TTL, and batch operations.
Document.get
@classmethod
async def get(cls: Type["DocType"], document_id: Any) -> Optional["DocType"]
Get document by id, returns None if document does not exist
Arguments:
document_id: str - document idReturns:
Union["Document", None]
Document.exists
@classmethod
async def exists(cls: Type["DocType"], document_id: Any) -> bool
Check if a document exists by ID
Arguments:
document_id: str - document idReturns:
bool
Document.insert
@wrap_with_actions(EventTypes.INSERT)
async def insert(ttl: Optional[int] = None, skip_actions=None) -> DocType
Insert the document (self) to Redis
Arguments:
ttl: Optional[int] - TTL in secondsskip_actions: Optional[List] - actions to skipReturns:
Document
Document.insert_one
@classmethod
async def insert_one(cls: Type[DocType],
document: DocType,
ttl: Optional[int] = None) -> Optional[DocType]
Insert one document to Redis
Arguments:
document: Document - document to insertttl: Optional[int] - TTL in secondsReturns:
DocType
Document.insert_many
@classmethod
async def insert_many(cls: Type[DocType],
documents: Iterable[DocType],
ttl: Optional[int] = None) -> List[DocType]
Insert many documents to Redis using pipeline
Arguments:
documents: List["Document"] - documents to insertttl: Optional[int] - TTL in seconds for all documentsReturns:
List[DocType]
Document.get_many
@classmethod
async def get_many(cls: Type[DocType],
document_ids: List[Any]) -> List[Optional[DocType]]
Get many documents by IDs using pipeline
Optimized with msgspec (2x faster than orjson) and model_construct() (skip validation)
Performance: 3-4x faster than orjson + model_validate approach
Arguments:
document_ids: List[str] - list of document IDsReturns:
List[Optional[DocType]]
Document.save
def save() -> DocType
Update an existing model in Redis or insert it if it does not yet exist.
Returns:
Document
Document.update
@wrap_with_actions(EventTypes.UPDATE)
async def update(skip_actions=None, **fields) -> DocType
Update specific fields of the document
Arguments:
skip_actions: Optional[List] - actions to skipfields: Field names and values to updateReturns:
Document
Document.get_field
async def get_field(field_name: str) -> Any
Get a specific field value from Redis without loading the entire document
Arguments:
field_name: Name of the fieldReturns:
Field value
Document.set_field
async def set_field(field_name: str, value: Any) -> None
Set a specific field value in Redis
Arguments:
field_name: Name of the fieldvalue: Value to set
Document.increment_field
async def increment_field(field_name: str,
amount: Union[int, float] = 1) -> Union[int, float]
Increment a numeric field atomically
Arguments:
field_name: Name of the fieldamount: Amount to increment byReturns:
New value
Document.set_ttl
async def set_ttl(seconds: int) -> bool
Set TTL (time to live) for this document
Arguments:
seconds: TTL in secondsReturns:
bool - True if TTL was set
Document.get_ttl
async def get_ttl() -> Optional[int]
Get the remaining TTL for this document
Returns:
Optional[int] - TTL in seconds, -1 if no TTL, -2 if key doesn't exist
Document.persist
async def persist() -> bool
Remove TTL from this document (make it persistent)
Returns:
bool - True if TTL was removed
Document.delete_self
@wrap_with_actions(EventTypes.DELETE)
async def delete_self(skip_actions=None)
Delete the document
Document.delete
@classmethod
async def delete(cls, document_id)
Delete a document by ID
Arguments:
document_id: str - document id
Document.delete_many
@classmethod
async def delete_many(cls, document_ids: List[Any]) -> int
Delete many documents by IDs
Arguments:
document_ids: List[str] - list of document IDsReturns:
int - number of documents deleted
Document.delete_all
@classmethod
async def delete_all(cls) -> int
Delete all documents of this class
Returns:
int - number of documents deleted
Document.count
@classmethod
async def count(cls) -> int
Count all documents of this class
Returns:
int - number of documents
Document.find
@classmethod
async def find(cls: Type[DocType], **filters) -> List[DocType]
Find documents by indexed fields
Examples: # Exact match on indexed field products = await Product.find(category="electronics")
# Range query on numeric indexed field products = await Product.find(price__gte=10, price__lte=100)Arguments:
filters: Field filters (supports __gte, __lte for numeric fields)Returns:
List[DocType]
Document.all
@classmethod
async def all(cls: Type[DocType],
skip: int = 0,
limit: Optional[int] = None,
sort_desc: bool = False) -> List[DocType]
Get all documents of this class
Arguments:
skip: Number of documents to skiplimit: Maximum number of documents to returnsort_desc: Sort by insertion time descendingReturns:
List[DocType]
Document.use_state_management
@classmethod
def use_state_management(cls) -> bool
Is state management turned on
Returns:
bool
Document.state_management_save_previous
@classmethod
def state_management_save_previous(cls) -> bool
Should we save the previous state after a commit to database
Returns:
bool
Document.state_management_replace_objects
@classmethod
def state_management_replace_objects(cls) -> bool
Should objects be replaced when using state management
Returns:
bool
Document.get_saved_state
def get_saved_state() -> Optional[Dict[str, Any]]
Saved state getter. It is protected property.
Returns:
Optional[Dict[str, Any]] - saved state
Document.get_previous_saved_state
def get_previous_saved_state() -> Optional[Dict[str, Any]]
Previous state getter. It is a protected property.
Returns:
Optional[Dict[str, Any]] - previous state
Document.get_settings
@classmethod
def get_settings(cls) -> ItemSettings
Get document settings, which was created on
the initialization step
Returns:
ItemSettings class