odmantic.model

Bases: BaseModel

Base class for Model and EmbeddedModel.

Warning

This internal class should never be instanciated directly.

Source code in odmantic/model.py

class _BaseODMModel(pydantic.BaseModel, metaclass=ABCMeta):
    """Base class for [Model][odmantic.model.Model] and
    [EmbeddedModel][odmantic.model.EmbeddedModel].

    !!! warning
        This internal class should never be instanciated directly.
    """

    if TYPE_CHECKING:
        __odm_fields__: ClassVar[Dict[str, ODMBaseField]] = {}
        __bson_serializers__: ClassVar[Dict[str, Callable[[Any], Any]]] = {}
        __mutable_fields__: ClassVar[FrozenSet[str]] = frozenset()
        __references__: ClassVar[Tuple[str, ...]] = ()
        __pydantic_model__: ClassVar[Type[BaseBSONModel]]
        # __fields_modified__ is not a ClassVar but this allows to hide this field from
        # the dataclass transform generated constructor
        __fields_modified__: ClassVar[Set[str]] = set()
        model_config: ClassVar[ODMConfigDict]

    __slots__ = ("__fields_modified__",)

    def __init__(self, **data: Any):
        super().__init__(**data)
        object.__setattr__(self, "__fields_modified__", set(self.__odm_fields__.keys()))

    @classmethod
    # TODO: rename to model_validate
    def validate(cls: Type[BaseT], value: Any) -> BaseT:
        if isinstance(value, cls):
            # Do not copy the object as done in pydantic
            # This enable to keep the same python object
            return value
        return super().model_validate(value)

    def __repr_args__(self) -> "ReprArgs":
        # Place the id field first in the repr string
        args = list(super().__repr_args__())
        id_arg = next((arg for arg in args if arg[0] == "id"), None)
        if id_arg is None:
            return args
        args.remove(id_arg)
        args = [id_arg] + args
        return args

    @deprecated(
        "copy is deprecated, please use model_copy instead",
    )
    def copy(
        self: BaseT,
        *,
        include: Union["AbstractSetIntStr", "MappingIntStrAny", None] = None,
        exclude: Union["AbstractSetIntStr", "MappingIntStrAny", None] = None,
        update: Dict[str, Any] | None = None,
        deep: bool = False,
    ) -> BaseT:
        if include is not None or exclude is not None:
            raise NotImplementedError(
                "copy with include or exclude is not supported anymore, "
                "please use `model_copy` instead"
            )
        return self.model_copy(update=update, deep=deep)

    def model_copy(
        self: BaseT,
        *,
        update: Optional["DictStrAny"] = None,
        deep: bool = False,
    ) -> BaseT:
        """Duplicate a model, optionally choose which fields to change.

        Danger:
            The data is not validated before creating the new model: **you should trust
            this data**.

        Arguments:
            update: values to change/add in the new model.
            deep: set to `True` to make a deep copy of the model

        Returns:
            new model instance

        """
        copied = super().model_copy(update=update, deep=deep)
        copied._post_copy_update()
        return copied

    def _post_copy_update(self: BaseT) -> None:
        """Recursively update internal fields of the copied model after it has been
        copied.

        Set them as if they were modified to make sure they are saved in the database.
        """
        object.__setattr__(self, "__fields_modified__", set(self.model_fields))
        for field_name, field in self.__odm_fields__.items():
            if isinstance(field, ODMEmbedded):
                value = getattr(self, field_name)
                value._post_copy_update()

    @deprecated(
        "update is deprecated, please use model_update instead",
    )
    def update(
        self,
        patch_object: Union[BaseModel, Dict[str, Any]],
        *,
        include: "IncEx" = None,
        exclude: "IncEx" = None,
        exclude_unset: bool = True,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
    ) -> None:
        self.model_update(
            patch_object,
            include=include,
            exclude=exclude,
            exclude_unset=exclude_unset,
            exclude_defaults=exclude_defaults,
            exclude_none=exclude_none,
        )

    def model_update(
        self,
        patch_object: Union[BaseModel, Dict[str, Any]],
        *,
        include: "IncEx" = None,
        exclude: "IncEx" = None,
        exclude_unset: bool = True,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
    ) -> None:
        """Update instance fields from a Pydantic model or a dictionary.

        If a pydantic model is provided, only the **fields set** will be
        applied by default.

        Args:
            patch_object: object containing the values to update
            include: fields to include from the `patch_object` (include all fields if
                `None`)
            exclude: fields to exclude from the `patch_object`, this takes
                precedence over include
            exclude_unset: only update fields explicitly set in the patch object (only
                applies to Pydantic models)
            exclude_defaults: only update fields that are different from their default
                value in the patch object (only applies to Pydantic models)
            exclude_none: only update fields different from None in the patch object
                (only applies to Pydantic models)

        Raises:
            ValidationError: the modifications would make the instance invalid

        <!--
        #noqa: DAR402 ValidationError
        -->
        """
        if isinstance(patch_object, BaseModel):
            patch_dict = patch_object.model_dump(
                include=include,
                exclude=exclude,
                exclude_unset=exclude_unset,
                exclude_defaults=exclude_defaults,
                exclude_none=exclude_none,
            )
        else:
            odm_fields = set(self.__odm_fields__.keys())
            patch_dict = {}
            for k, v in patch_object.items():
                if include is not None and k not in include:
                    continue
                if exclude is not None and k in exclude:
                    continue
                if k not in odm_fields:
                    continue
                patch_dict[k] = v
        patched_instance_dict = {**self.model_dump(), **patch_dict}
        # FIXME: improve performance by only running updated field validators and then
        # model validators
        patched_instance = self.validate(patched_instance_dict)
        for name, new_value in patched_instance.__dict__.items():
            if self.__dict__[name] != new_value:
                # Manually change the field to avoid running the validators again
                self.__dict__[name] = new_value
                self.model_fields_set.add(name)
                self.__fields_modified__.add(name)

    def __setattr__(self, name: str, value: Any) -> None:
        super().__setattr__(name, value)
        self.__fields_modified__.add(name)

    @deprecated(
        "doc is deprecated, please use model_dump_doc instead",
    )
    def doc(self, include: Optional["AbstractSetIntStr"] = None) -> Dict[str, Any]:
        return self.model_dump_doc(include=include)

    def model_dump_doc(
        self, include: Optional["AbstractSetIntStr"] = None
    ) -> Dict[str, Any]:
        """Generate a document (BSON) representation of the instance (as a dictionary).

        Args:
            include: field that should be included; if None, every fields will be
                included

        Returns:
            the document associated to the instance
        """
        raw_doc = self.model_dump()
        doc = self.__doc(raw_doc, type(self), include)
        return doc

    def __doc(  # noqa C901 # TODO: refactor document generation
        self,
        raw_doc: Dict[str, Any],
        model: Type["_BaseODMModel"],
        include: Optional["AbstractSetIntStr"] = None,
    ) -> Dict[str, Any]:
        doc: Dict[str, Any] = {}
        for field_name, field in model.__odm_fields__.items():
            if include is not None and field_name not in include:
                continue
            if isinstance(field, ODMReference):
                doc[field.key_name] = raw_doc[field_name][field.model.__primary_field__]
            elif isinstance(field, ODMEmbedded):
                doc[field.key_name] = self.__doc(raw_doc[field_name], field.model, None)
            elif isinstance(field, ODMEmbeddedGeneric):
                if field.generic_origin is dict:
                    doc[field.key_name] = {
                        item_key: self.__doc(item_value, field.model)
                        for item_key, item_value in raw_doc[field_name].items()
                    }
                else:
                    doc[field.key_name] = [
                        self.__doc(item, field.model) for item in raw_doc[field_name]
                    ]
            elif field_name in model.__bson_serializers__:
                doc[field.key_name] = model.__bson_serializers__[field_name](
                    raw_doc[field_name]
                )
            else:
                doc[field.key_name] = raw_doc[field_name]

        if model.model_config["extra"] == "allow":
            # raw_doc is indexed by field name so we compare against odm field names
            extras = set(raw_doc.keys()) - set(self.__odm_fields__.keys())
            for extra in extras:
                value = raw_doc[extra]
                subst_type = validate_type(type(value))
                bson_serializer = _get_bson_serializer(subst_type)
                if bson_serializer is not None:
                    doc[extra] = bson_serializer(value)
                else:
                    doc[extra] = value
        return doc

    @classmethod
    @deprecated(
        "parse_doc is deprecated, please use model_validate_doc instead",
    )
    def parse_doc(cls: Type[BaseT], raw_doc: Dict) -> BaseT:
        return cls.model_validate_doc(raw_doc)

    @classmethod
    def model_validate_doc(cls: Type[BaseT], raw_doc: Dict) -> BaseT:
        """Parse a BSON document into an instance of the Model

        Args:
            raw_doc: document to parse (as Dict)

        Raises:
            DocumentParsingError: the specified document is invalid

        Returns:
            an instance of the Model class this method is called on.
        """
        errors, obj = cls._parse_doc_to_obj(raw_doc)
        if len(errors) > 0:
            raise DocumentParsingError(
                errors=errors,
                model=cls,
            )
        try:
            instance = cls.model_validate(obj)
        except ValidationError as e:
            raise DocumentParsingError(
                errors=e.errors(),  # type: ignore
                model=cls,
            )

        return instance

    @classmethod
    def _parse_doc_to_obj(  # noqa C901 # TODO: refactor document parsing
        cls: Type[BaseT], raw_doc: Dict, base_loc: Tuple[str, ...] = ()
    ) -> Tuple[List[InitErrorDetails], Dict[str, Any]]:
        errors: List[InitErrorDetails] = []
        obj: Dict[str, Any] = {}
        for field_name, field in cls.__odm_fields__.items():
            if isinstance(field, ODMReference):
                sub_doc = raw_doc.get(field.key_name)
                if sub_doc is None:
                    errors.append(
                        InitErrorDetails(
                            type=ReferencedDocumentNotFoundError(field.key_name),
                            loc=base_loc + (field_name,),
                            input=raw_doc,
                        )
                    )

                else:
                    sub_errors, sub_obj = field.model._parse_doc_to_obj(
                        sub_doc, base_loc=base_loc + (field_name,)
                    )
                    errors.extend(sub_errors)
                    obj[field_name] = sub_obj
            elif isinstance(field, ODMEmbedded):
                value = raw_doc.get(field.key_name, Undefined)
                if value is not Undefined:
                    sub_errors, value = field.model._parse_doc_to_obj(
                        value, base_loc=base_loc + (field_name,)
                    )
                    errors.extend(sub_errors)
                else:
                    if not field.is_required_in_doc():
                        value = field.get_default_importing_value()
                    if value is Undefined:
                        errors.append(
                            InitErrorDetails(
                                type=KeyNotFoundInDocumentError(field.key_name),
                                loc=base_loc + (field_name,),
                                input=raw_doc,
                            )
                        )

                obj[field_name] = value
            elif isinstance(field, ODMEmbeddedGeneric):
                value = Undefined
                raw_value = raw_doc.get(field.key_name, Undefined)
                if raw_value is not Undefined:
                    if isinstance(raw_value, list) and (
                        field.generic_origin is list
                        or field.generic_origin is tuple
                        or field.generic_origin is set
                    ):
                        value = []
                        for i, item in enumerate(raw_value):
                            sub_errors, item = field.model._parse_doc_to_obj(
                                item, base_loc=base_loc + (field_name, f"[{i}]")
                            )
                            if len(sub_errors) > 0:
                                errors.extend(sub_errors)
                            else:
                                value.append(item)
                        obj[field_name] = value
                    elif isinstance(raw_value, dict) and field.generic_origin is dict:
                        value = {}
                        for item_key, item_value in raw_value.items():
                            sub_errors, item_value = field.model._parse_doc_to_obj(
                                item_value,
                                base_loc=base_loc + (field_name, f'["{item_key}"]'),
                            )
                            if len(sub_errors) > 0:
                                errors.extend(sub_errors)
                            else:
                                value[item_key] = item_value
                        obj[field_name] = value
                    else:
                        errors.append(
                            InitErrorDetails(
                                type=IncorrectGenericEmbeddedModelValue(raw_value),
                                loc=base_loc + (field_name,),
                                input=raw_doc,
                            )
                        )

                else:
                    if not field.is_required_in_doc():
                        value = field.get_default_importing_value()
                    if value is Undefined:
                        errors.append(
                            InitErrorDetails(
                                type=KeyNotFoundInDocumentError(field.key_name),
                                loc=base_loc + (field_name,),
                                input=raw_doc,
                            )
                        )
                    else:
                        obj[field_name] = value
            else:
                field = cast(ODMField, field)
                value = raw_doc.get(field.key_name, Undefined)
                if value is Undefined and not field.is_required_in_doc():
                    value = field.get_default_importing_value()

                if value is Undefined:
                    errors.append(
                        InitErrorDetails(
                            type=KeyNotFoundInDocumentError(field.key_name),
                            loc=base_loc + (field_name,),
                            input=raw_doc,
                        )
                    )
                else:
                    obj[field_name] = value

        if cls.model_config["extra"] == "allow":
            extras = set(raw_doc.keys()) - set(obj.keys())
            for extra in extras:
                obj[extra] = raw_doc[extra]

        return errors, obj

model_copy(*, update=None, deep=False)

Duplicate a model, optionally choose which fields to change.

Danger

The data is not validated before creating the new model: you should trust this data.

Parameters:

Name	Type	Description	Default
update	Optional['DictStrAny']	values to change/add in the new model.	None
deep	bool	set to True to make a deep copy of the model	False

Returns:

Type	Description
BaseT	new model instance

Source code in odmantic/model.py

def model_copy(
    self: BaseT,
    *,
    update: Optional["DictStrAny"] = None,
    deep: bool = False,
) -> BaseT:
    """Duplicate a model, optionally choose which fields to change.

    Danger:
        The data is not validated before creating the new model: **you should trust
        this data**.

    Arguments:
        update: values to change/add in the new model.
        deep: set to `True` to make a deep copy of the model

    Returns:
        new model instance

    """
    copied = super().model_copy(update=update, deep=deep)
    copied._post_copy_update()
    return copied

model_dump_doc(include=None)

Generate a document (BSON) representation of the instance (as a dictionary).

Parameters:

Name	Type	Description	Default
include	Optional['AbstractSetIntStr']	field that should be included; if None, every fields will be included	None

Returns:

Type	Description
Dict[str, Any]	the document associated to the instance

Source code in odmantic/model.py

def model_dump_doc(
    self, include: Optional["AbstractSetIntStr"] = None
) -> Dict[str, Any]:
    """Generate a document (BSON) representation of the instance (as a dictionary).

    Args:
        include: field that should be included; if None, every fields will be
            included

    Returns:
        the document associated to the instance
    """
    raw_doc = self.model_dump()
    doc = self.__doc(raw_doc, type(self), include)
    return doc

model_update(patch_object, *, include=None, exclude=None, exclude_unset=True, exclude_defaults=False, exclude_none=False)

Update instance fields from a Pydantic model or a dictionary.

If a pydantic model is provided, only the fields set will be applied by default.

Parameters:

Name	Type	Description	Default
patch_object	Union[BaseModel, Dict[str, Any]]	object containing the values to update	required
include	'IncEx'	fields to include from the patch_object (include all fields if None)	None
exclude	'IncEx'	fields to exclude from the patch_object, this takes precedence over include	None
exclude_unset	bool	only update fields explicitly set in the patch object (only applies to Pydantic models)	True
exclude_defaults	bool	only update fields that are different from their default value in the patch object (only applies to Pydantic models)	False
exclude_none	bool	only update fields different from None in the patch object (only applies to Pydantic models)	False

Raises:

Type	Description
ValidationError	the modifications would make the instance invalid

Source code in odmantic/model.py

def model_update(
    self,
    patch_object: Union[BaseModel, Dict[str, Any]],
    *,
    include: "IncEx" = None,
    exclude: "IncEx" = None,
    exclude_unset: bool = True,
    exclude_defaults: bool = False,
    exclude_none: bool = False,
) -> None:
    """Update instance fields from a Pydantic model or a dictionary.

    If a pydantic model is provided, only the **fields set** will be
    applied by default.

    Args:
        patch_object: object containing the values to update
        include: fields to include from the `patch_object` (include all fields if
            `None`)
        exclude: fields to exclude from the `patch_object`, this takes
            precedence over include
        exclude_unset: only update fields explicitly set in the patch object (only
            applies to Pydantic models)
        exclude_defaults: only update fields that are different from their default
            value in the patch object (only applies to Pydantic models)
        exclude_none: only update fields different from None in the patch object
            (only applies to Pydantic models)

    Raises:
        ValidationError: the modifications would make the instance invalid

    <!--
    #noqa: DAR402 ValidationError
    -->
    """
    if isinstance(patch_object, BaseModel):
        patch_dict = patch_object.model_dump(
            include=include,
            exclude=exclude,
            exclude_unset=exclude_unset,
            exclude_defaults=exclude_defaults,
            exclude_none=exclude_none,
        )
    else:
        odm_fields = set(self.__odm_fields__.keys())
        patch_dict = {}
        for k, v in patch_object.items():
            if include is not None and k not in include:
                continue
            if exclude is not None and k in exclude:
                continue
            if k not in odm_fields:
                continue
            patch_dict[k] = v
    patched_instance_dict = {**self.model_dump(), **patch_dict}
    # FIXME: improve performance by only running updated field validators and then
    # model validators
    patched_instance = self.validate(patched_instance_dict)
    for name, new_value in patched_instance.__dict__.items():
        if self.__dict__[name] != new_value:
            # Manually change the field to avoid running the validators again
            self.__dict__[name] = new_value
            self.model_fields_set.add(name)
            self.__fields_modified__.add(name)

model_validate_doc(raw_doc) classmethod

Parse a BSON document into an instance of the Model

Parameters:

Name	Type	Description	Default
raw_doc	Dict	document to parse (as Dict)	required

Raises:

Type	Description
DocumentParsingError	the specified document is invalid

Returns:

Type	Description
BaseT	an instance of the Model class this method is called on.

Source code in odmantic/model.py

@classmethod
def model_validate_doc(cls: Type[BaseT], raw_doc: Dict) -> BaseT:
    """Parse a BSON document into an instance of the Model

    Args:
        raw_doc: document to parse (as Dict)

    Raises:
        DocumentParsingError: the specified document is invalid

    Returns:
        an instance of the Model class this method is called on.
    """
    errors, obj = cls._parse_doc_to_obj(raw_doc)
    if len(errors) > 0:
        raise DocumentParsingError(
            errors=errors,
            model=cls,
        )
    try:
        instance = cls.model_validate(obj)
    except ValidationError as e:
        raise DocumentParsingError(
            errors=e.errors(),  # type: ignore
            model=cls,
        )

    return instance

Bases: _BaseODMModel

Class that can be extended to create an ODMantic Model.

Each model will be bound to a MongoDB collection. You can customize the collection name by setting the __collection__ class variable in the model classes.

Source code in odmantic/model.py

class Model(_BaseODMModel, metaclass=ModelMetaclass):
    """Class that can be extended to create an ODMantic Model.

    Each model will be bound to a MongoDB collection. You can customize the collection
    name by setting the `__collection__` class variable in the model classes.
    """

    if TYPE_CHECKING:
        __collection__: ClassVar[str] = ""
        __primary_field__: ClassVar[str] = ""

        id: Union[ObjectId, Any] = Field()  # TODO fix basic id field typing

    def __setattr__(self, name: str, value: Any) -> None:
        if name == self.__primary_field__:
            # TODO implement
            raise NotImplementedError(
                "Reassigning a new primary key is not supported yet"
            )
        super().__setattr__(name, value)

    @classmethod
    def __indexes__(cls) -> Tuple[Union[ODMBaseIndex, pymongo.IndexModel], ...]:
        indexes: List[Union[ODMBaseIndex, pymongo.IndexModel]] = []
        for field in cls.__odm_fields__.values():
            if isinstance(field, ODMBaseIndexableField) and (
                field.index or field.unique
            ):
                indexes.append(
                    ODMSingleFieldIndex(
                        key_name=field.key_name,
                        unique=field.unique,
                    )
                )
        get_indexes_from_config = cls.model_config["indexes"]
        if get_indexes_from_config is not None:
            for index in get_indexes_from_config():
                indexes.append(
                    index.to_odm_index() if isinstance(index, Index) else index
                )
        return tuple(indexes)

    @deprecated(
        "update is deprecated, please use model_update instead",
    )
    def update(
        self,
        patch_object: Union[BaseModel, Dict[str, Any]],
        *,
        include: "IncEx" = None,
        exclude: "IncEx" = None,
        exclude_unset: bool = True,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
    ) -> None:
        return self.model_update(
            patch_object,
            include=include,
            exclude=exclude,
            exclude_unset=exclude_unset,
            exclude_defaults=exclude_defaults,
            exclude_none=exclude_none,
        )

    def model_update(
        self,
        patch_object: Union[BaseModel, Dict[str, Any]],
        *,
        include: "IncEx" = None,
        exclude: "IncEx" = None,
        exclude_unset: bool = True,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
    ) -> None:
        is_primary_field_in_patch = (
            isinstance(patch_object, BaseModel)
            and self.__primary_field__ in patch_object.model_fields
        ) or (isinstance(patch_object, dict) and self.__primary_field__ in patch_object)
        if is_primary_field_in_patch:
            if (
                include is None
                and (exclude is None or self.__primary_field__ not in exclude)
            ) or (
                include is not None
                and self.__primary_field__ in include
                and (exclude is None or self.__primary_field__ not in exclude)
            ):
                raise ValueError(
                    "Updating the primary key is not supported. "
                    "See the copy method if you want to modify the primary field."
                )
        return super().model_update(
            patch_object,
            include=include,
            exclude=exclude,
            exclude_unset=exclude_unset,
            exclude_defaults=exclude_defaults,
            exclude_none=exclude_none,
        )

Bases: _BaseODMModel

Class that can be extended to create an ODMantic Embedded Model.

An embedded document cannot be persisted directly to the database but should be integrated in a regular ODMantic Model.

Source code in odmantic/model.py

class EmbeddedModel(_BaseODMModel, metaclass=EmbeddedModelMetaclass):
    """Class that can be extended to create an ODMantic Embedded Model.

    An embedded document cannot be persisted directly to the database but should be
    integrated in a regular ODMantic Model.
    """