Skip to content

beanie.odm.documents

Document

class Document(
    BaseModel, 
    UpdateMethods, 
    FindInterface, 
    AggregateInterface, 
    OtherGettersInterface)

Document Mapping class.

Fields:

  • id - MongoDB document ObjectID "_id" field. Mapped to the PydanticObjectId class

Inherited from:

Document.get

 | @classmethod
 | async get(cls: Type["DocType"], document_id: PydanticObjectId, session: Optional[ClientSession] = None, ignore_cache: bool = False, fetch_links: bool = False, **pymongo_kwargs, ,) -> Optional["DocType"]

Get document by id, returns None if document does not exist

Arguments:

:param **pymongo_kwargs: pymongo native parameters for find operation - document_id: PydanticObjectId - document id - session: Optional[ClientSession] - pymongo session - ignore_cache: bool - ignore cache (if it is turned on)

Returns:

Union["Document", None]

Document.insert

 | @wrap_with_actions(EventTypes.INSERT)
 | @save_state_after
 | @swap_revision_after
 | @validate_self_before
 | async insert(*, link_rule: WriteRules = WriteRules.DO_NOTHING, session: Optional[ClientSession] = None, skip_actions: Optional[List[Union[ActionDirections, str]]] = None) -> DocType

Insert the document (self) to the collection

Returns:

Document

Document.create

 | async create(session: Optional[ClientSession] = None) -> DocType

The same as self.insert()

Returns:

Document

Document.insert_one

 | @classmethod
 | async insert_one(cls: Type[DocType], document: DocType, session: Optional[ClientSession] = None, bulk_writer: "BulkWriter" = None, link_rule: WriteRules = WriteRules.DO_NOTHING) -> Optional[DocType]

Insert one document to the collection

Arguments:

  • document: Document - document to insert
  • session: ClientSession - pymongo session
  • bulk_writer: "BulkWriter" - Beanie bulk writer
  • link_rule: InsertRules - hot to manage link fields

Returns:

DocType

Document.insert_many

 | @classmethod
 | async insert_many(cls: Type[DocType], documents: List[DocType], session: Optional[ClientSession] = None, link_rule: WriteRules = WriteRules.DO_NOTHING, **pymongo_kwargs, ,) -> InsertManyResult

Insert many documents to the collection

Arguments:

  • documents: List["Document"] - documents to insert
  • session: ClientSession - pymongo session
  • link_rule: InsertRules - how to manage link fields

Returns:

InsertManyResult

Document.replace

 | @wrap_with_actions(EventTypes.REPLACE)
 | @save_state_after
 | @swap_revision_after
 | @validate_self_before
 | async replace(ignore_revision: bool = False, session: Optional[ClientSession] = None, bulk_writer: Optional[BulkWriter] = None, link_rule: WriteRules = WriteRules.DO_NOTHING, skip_actions: Optional[List[Union[ActionDirections, str]]] = None) -> DocType

Fully update the document in the database

Arguments:

Used when revision based protection is turned on.
  • session: Optional[ClientSession] - pymongo session.
  • ignore_revision: bool - do force replace.
  • bulk_writer: "BulkWriter" - Beanie bulk writer

Returns:

self

Document.save

 | async save(session: Optional[ClientSession] = None, link_rule: WriteRules = WriteRules.DO_NOTHING, **kwargs, ,) -> DocType

Update an existing model in the database or insert it if it does not yet exist.

Arguments:

  • session: Optional[ClientSession] - pymongo session.

Returns:

None

Document.save_changes

 | @saved_state_needed
 | @wrap_with_actions(EventTypes.SAVE_CHANGES)
 | @validate_self_before
 | async save_changes(ignore_revision: bool = False, session: Optional[ClientSession] = None, bulk_writer: Optional[BulkWriter] = None, skip_actions: Optional[List[Union[ActionDirections, str]]] = None) -> None

Save changes. State management usage must be turned on

Arguments:

  • ignore_revision: bool - ignore revision id, if revision is turned on
  • bulk_writer: "BulkWriter" - Beanie bulk writer

Returns:

None

Document.replace_many

 | @classmethod
 | async replace_many(cls: Type[DocType], documents: List[DocType], session: Optional[ClientSession] = None) -> None

Replace list of documents

Arguments:

  • documents: List["Document"]
  • session: Optional[ClientSession] - pymongo session.

Returns:

None

Document.update

 | @save_state_after
 | async update(*args, *, ignore_revision: bool = False, session: Optional[ClientSession] = None, bulk_writer: Optional[BulkWriter] = None, **pymongo_kwargs, ,) -> None

Partially update the document in the database

Arguments:

:param *pymongo_kwargs: pymongo native parameters for update operation - args: Union[dict, Mapping] - the modifications to apply. - session: ClientSession - pymongo session. - ignore_revision: bool - force update. Will update even if revision id is not the same, as stored - bulk_writer: "BulkWriter" - Beanie bulk writer

Returns:

None

Document.update_all

 | @classmethod
 | update_all(cls, *args: Union[dict, Mapping], *, session: Optional[ClientSession] = None, bulk_writer: Optional[BulkWriter] = None, **pymongo_kwargs, ,) -> UpdateMany

Partially update all the documents

Arguments:

:param *pymongo_kwargs: pymongo native parameters for find operation - args: Union[dict, Mapping] - the modifications to apply. - session: ClientSession - pymongo session. - bulk_writer: "BulkWriter" - Beanie bulk writer

Returns:

UpdateMany query

Document.delete

 | @wrap_with_actions(EventTypes.DELETE)
 | async delete(session: Optional[ClientSession] = None, bulk_writer: Optional[BulkWriter] = None, link_rule: DeleteRules = DeleteRules.DO_NOTHING, skip_actions: Optional[List[Union[ActionDirections, str]]] = None, **pymongo_kwargs, ,) -> Optional[DeleteResult]

Delete the document

Arguments:

:param **pymongo_kwargs: pymongo native parameters for delete operation - session: Optional[ClientSession] - pymongo session. - bulk_writer: "BulkWriter" - Beanie bulk writer - link_rule: DeleteRules - rules for link fields

Returns:

Optional[DeleteResult] - pymongo DeleteResult instance.

Document.delete_all

 | @classmethod
 | async delete_all(cls, session: Optional[ClientSession] = None, bulk_writer: Optional[BulkWriter] = None, **pymongo_kwargs, ,) -> Optional[DeleteResult]

Delete all the documents

Arguments:

:param **pymongo_kwargs: pymongo native parameters for delete operation - session: Optional[ClientSession] - pymongo session. - bulk_writer: "BulkWriter" - Beanie bulk writer

Returns:

Optional[DeleteResult] - pymongo DeleteResult instance.

Document.use_state_management

 | @classmethod
 | use_state_management(cls) -> bool

Is state management turned on

Returns:

bool

Document.state_management_replace_objects

 | @classmethod
 | state_management_replace_objects(cls) -> bool

Should objects be replaced when using state management

Returns:

bool

Document.get_saved_state

 | get_saved_state() -> Optional[Dict[str, Any]]

Saved state getter. It is protected property.

Returns:

Optional[Dict[str, Any]] - saved state

Document.init_cache

 | @classmethod
 | init_cache(cls) -> None

Init model's cache

Returns:

None

Document.init_fields

 | @classmethod
 | init_fields(cls) -> None

Init class fields

Returns:

None

Document.init_settings

 | @classmethod
 | async init_settings(cls, database: AsyncIOMotorDatabase, allow_index_dropping: bool) -> None

Init document settings (collection and models)

Arguments:

  • database: AsyncIOMotorDatabase - motor database
  • allow_index_dropping: bool

Returns:

None

Document.init_actions

 | @classmethod
 | init_actions(cls)

Init event-based actions

Document.init_model

 | @classmethod
 | async init_model(cls, database: AsyncIOMotorDatabase, allow_index_dropping: bool) -> None

Init wrapper

Arguments:

  • database: AsyncIOMotorDatabase
  • allow_index_dropping: bool

Returns:

None

Document.get_settings

 | @classmethod
 | get_settings(cls) -> DocumentSettings

Get document settings, which was created on the initialization step

Returns:

DocumentSettings class

Document.inspect_collection

 | @classmethod
 | async inspect_collection(cls, session: Optional[ClientSession] = None) -> InspectionResult

Check, if documents, stored in the MongoDB collection are compatible with the Document schema

Returns:

InspectionResult

Document.dict

 | dict(*, include: Union["AbstractSetIntStr", "MappingIntStrAny"] = None, exclude: Union["AbstractSetIntStr", "MappingIntStrAny"] = None, by_alias: bool = False, skip_defaults: bool = None, exclude_hidden: bool = True, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) -> "DictStrAny"

Overriding of the respective method from Pydantic Hides fields, marked as "hidden