Document

bunnet.odm.documents

Document

class Document(LazyModel, SettersInterface, InheritanceInterface,
               FindInterface, AggregateInterface, OtherGettersInterface)

Document Mapping class.

Fields:

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

Document.get

@classmethod
def get(cls: Type["DocType"],
        document_id: Any,
        session: Optional[ClientSession] = None,
        ignore_cache: bool = False,
        fetch_links: bool = False,
        with_children: bool = False,
        nesting_depth: Optional[int] = None,
        nesting_depths_per_field: Optional[Dict[str, int]] = None,
        **pymongo_kwargs) -> Optional["DocType"]

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

Arguments:

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

Returns:

Union["Document", None]

Document.sync

def sync(merge_strategy: MergeStrategy = MergeStrategy.remote)

Sync the document with the database

Arguments:

• merge_strategy: MergeStrategy - how to merge the document

Returns:

None

Document.insert

@wrap_with_actions(EventTypes.INSERT)
@save_state_after
@validate_self_before
def 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

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

The same as self.insert()

Returns:

Document

Document.insert_one

@classmethod
def insert_one(
        cls: Type[DocType],
        document: DocType,
        session: Optional[ClientSession] = None,
        bulk_writer: Optional["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" - bunnet bulk writer
• link_rule: InsertRules - hot to manage link fields

Returns:

DocType

Document.insert_many

@classmethod
def insert_many(cls: Type[DocType],
                documents: Iterable[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
@validate_self_before
def 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:

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

Returns:

self

Document.save

@wrap_with_actions(EventTypes.SAVE)
@save_state_after
@validate_self_before
def save(session: Optional[ClientSession] = None,
         link_rule: WriteRules = WriteRules.DO_NOTHING,
         ignore_revision: bool = False,
         **kwargs) -> None

Update an existing model in the database or

insert it if it does not yet exist.

Arguments:

• session: Optional[ClientSession] - pymongo session.
• link_rule: WriteRules - rules how to deal with links on writing
• ignore_revision: bool - do force save.

Returns:

None

Document.save_changes

@saved_state_needed
@wrap_with_actions(EventTypes.SAVE_CHANGES)
@validate_self_before
def 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" - bunnet bulk writer

Returns:

None

Document.replace_many

@classmethod
def 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

@wrap_with_actions(EventTypes.UPDATE)
@save_state_after
def update(*args,
           ignore_revision: bool = False,
           session: Optional[ClientSession] = None,
           bulk_writer: Optional[BulkWriter] = None,
           skip_actions: Optional[List[Union[ActionDirections, str]]] = None,
           skip_sync: Optional[bool] = None,
           **pymongo_kwargs) -> DocType

Partially update the document in the database

Arguments:

• 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" - bunnet bulk writer
• **pymongo_kwargs: pymongo native parameters for update operation

Returns:

None

Document.update_all

@classmethod
def 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:

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

Returns:

UpdateMany query

Document.set

def set(expression: Dict[Union[ExpressionField, str], Any],
        session: Optional[ClientSession] = None,
        bulk_writer: Optional[BulkWriter] = None,
        skip_sync: Optional[bool] = None,
        **kwargs)

Set values

Example:

class Sample(Document):
    one: int

Document.find(Sample.one == 1).set({Sample.one: 100})

Uses Set operator

Arguments:

• expression: Dict[Union[ExpressionField, str], Any] - keys and values to set
• session: Optional[ClientSession] - pymongo session
• bulk_writer: Optional[BulkWriter] - bulk writer
• skip_sync: bool - skip doc syncing. Available for the direct instances only

Returns:

self

Document.current_date

def current_date(expression: Dict[Union[ExpressionField, str], Any],
                 session: Optional[ClientSession] = None,
                 bulk_writer: Optional[BulkWriter] = None,
                 skip_sync: Optional[bool] = None,
                 **kwargs)

Set current date

Uses CurrentDate operator

Arguments:

• expression: Dict[Union[ExpressionField, str], Any]
• session: Optional[ClientSession] - pymongo session
• bulk_writer: Optional[BulkWriter] - bulk writer
• skip_sync: bool - skip doc syncing. Available for the direct instances only

Returns:

self

Document.inc

def inc(expression: Dict[Union[ExpressionField, str], Any],
        session: Optional[ClientSession] = None,
        bulk_writer: Optional[BulkWriter] = None,
        skip_sync: Optional[bool] = None,
        **kwargs)

Increment

Example:

class Sample(Document):
    one: int

Document.find(Sample.one == 1).inc({Sample.one: 100})

Uses Inc operator

Arguments:

• expression: Dict[Union[ExpressionField, str], Any]
• session: Optional[ClientSession] - pymongo session
• bulk_writer: Optional[BulkWriter] - bulk writer
• skip_sync: bool - skip doc syncing. Available for the direct instances only

Returns:

self

Document.delete

@wrap_with_actions(EventTypes.DELETE)
def 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:

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

Returns:

Optional[DeleteResult] - pymongo DeleteResult instance.

Document.delete_all

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

Delete all the documents

Arguments:

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

Returns:

Optional[DeleteResult] - pymongo DeleteResult instance.

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) -> DocumentSettings

Get document settings, which was created on

the initialization step

Returns:

DocumentSettings class

Document.inspect_collection

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

Check, if documents, stored in the MongoDB collection

are compatible with the Document schema

Returns:

InspectionResult