model.py

from __future__ import annotations

import datetime
import decimal
import enum
import pathlib
import uuid
from abc import ABCMeta
from collections.abc import Callable as abcCallable
from types import FunctionType
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    ClassVar,
    Dict,
    FrozenSet,
    Iterable,
    List,
    Optional,
    Set,
    Tuple,
    Type,
    TypeVar,
    Union,
    cast,
    no_type_check,
)

import bson
import pydantic
import pymongo
from pydantic import TypeAdapter, ValidationError
from pydantic._internal._decorators import PydanticDescriptorProxy
from pydantic.fields import Field as PDField
from pydantic.fields import FieldInfo as PDFieldInfo
from pydantic.main import BaseModel
from pydantic_core import InitErrorDetails
from typing_extensions import Literal, deprecated

from odmantic.bson import (
    _BSON_SUBSTITUTED_FIELDS,
    BaseBSONModel,
    ObjectId,
    _get_bson_serializer,
)
from odmantic.config import ODMConfigDict, validate_config
from odmantic.exceptions import (
    DocumentParsingError,
    IncorrectGenericEmbeddedModelValue,
    KeyNotFoundInDocumentError,
    ReferencedDocumentNotFoundError,
)
from odmantic.field import (
    Field,
    FieldProxy,
    ODMBaseField,
    ODMBaseIndexableField,
    ODMEmbedded,
    ODMEmbeddedGeneric,
    ODMField,
    ODMFieldInfo,
    ODMReference,
)
from odmantic.index import Index, ODMBaseIndex, ODMSingleFieldIndex
from odmantic.reference import ODMReferenceInfo
from odmantic.typing import (
    GenericAlias,
    dataclass_transform,
    get_args,
    get_first_type_argument_subclassing,
    get_origin,
    is_classvar,
    is_type_argument_subclass,
    lenient_issubclass,
    resolve_annotations,
)
from odmantic.utils import (
    Undefined,
    is_dunder,
    raise_on_invalid_collection_name,
    raise_on_invalid_key_name,
    to_snake_case,
)

if TYPE_CHECKING:
    from odmantic.typing import (
        AbstractSetIntStr,
        DictStrAny,
        IncEx,
        MappingIntStrAny,
        ReprArgs,
    )


UNTOUCHED_TYPES = (
    FunctionType,
    property,
    classmethod,
    staticmethod,
    PydanticDescriptorProxy,
)


def should_touch_field(value: Any = None, type_: Optional[Type] = None) -> bool:
    return not (
        lenient_issubclass(type_, UNTOUCHED_TYPES)
        or isinstance(value, UNTOUCHED_TYPES)
        or (type_ is not None and is_classvar(type_))
    )


def find_duplicate_key(fields: Iterable[ODMBaseField]) -> Optional[str]:
    seen: Set[str] = set()
    for f in fields:
        if f.key_name in seen:
            return f.key_name
        seen.add(f.key_name)
    return None


_IMMUTABLE_TYPES = (
    type(None),
    bool,
    int,
    float,
    str,
    bytes,
    tuple,
    frozenset,
    datetime.date,
    datetime.time,
    datetime.datetime,
    datetime.timedelta,
    enum.Enum,
    decimal.Decimal,
    pathlib.Path,
    uuid.UUID,
    pydantic.BaseModel,
    bson.ObjectId,
    bson.Decimal128,
    decimal.Decimal,
)


def is_type_mutable(type_: Type) -> bool:
    type_origin: Optional[Type] = getattr(type_, "__origin__", None)
    if type_origin is not None:
        type_args: Tuple[Type, ...] = getattr(type_, "__args__", ())
        for type_arg in type_args:
            if type_arg is ...:  # Handle tuple definition
                continue
            if lenient_issubclass(type_origin, Iterable) and lenient_issubclass(
                type_arg, EmbeddedModel
            ):  # Handle nested embedded models
                return True
            if is_type_mutable(type_arg):
                return True
        if type_origin is Union:
            return False
        return not lenient_issubclass(type_origin, _IMMUTABLE_TYPES)
    else:
        is_immutable = type_ is None or (
            lenient_issubclass(type_, _IMMUTABLE_TYPES)
            and not lenient_issubclass(type_, EmbeddedModel)
        )
        return not is_immutable


def is_type_forbidden(t: Type) -> bool:
    if t is Callable or t is abcCallable:
        # Callable type require a special treatment since typing.Callable is not a class
        return True
    return False


def validate_type(type_: Type) -> Type:
    if not should_touch_field(type_=type_) or lenient_issubclass(
        type_, (Model, EmbeddedModel)
    ):
        return type_
    if is_type_forbidden(type_):
        raise TypeError(f"{type_} fields are not supported")

    subst_type = _BSON_SUBSTITUTED_FIELDS.get(type_)
    if subst_type is not None:
        return subst_type

    type_origin: Optional[Type] = get_origin(type_)
    if type_origin is not None and type_origin is not Literal:
        type_args: Tuple[Type, ...] = get_args(type_)
        new_arg_types = tuple(validate_type(subtype) for subtype in type_args)
        # FIXME: remove this hack when a better solution to handle dynamic
        # generics is found
        # https://github.com/pydantic/pydantic/issues/8354
        if type_origin is Union:
            # as new_arg_types is a tuple, we can directly create a matching Union
            # instance, instead of hacking our way around it:
            # https://stackoverflow.com/a/72884529/3784643
            type_ = Union[new_arg_types]  # type: ignore
        else:
            type_ = GenericAlias(type_origin, new_arg_types)  # type: ignore
    return type_


class BaseModelMetaclass(pydantic._internal._model_construction.ModelMetaclass):
    @staticmethod
    def __validate_cls_namespace__(  # noqa C901
        name: str, namespace: Dict[str, Any]
    ) -> None:
        """Validate the class name space in place"""
        annotations = resolve_annotations(
            namespace.get("__annotations__", {}), namespace.get("__module__")
        )
        config = validate_config(namespace.get("model_config", ODMConfigDict()), name)
        odm_fields: Dict[str, ODMBaseField] = {}
        references: List[str] = []
        bson_serializers: Dict[str, Callable[[Any], Any]] = {}
        mutable_fields: Set[str] = set()

        # Make sure all fields are defined with type annotation
        for field_name, value in namespace.items():
            if (
                should_touch_field(value=value)
                and not is_dunder(field_name)
                and field_name not in annotations
                and not field_name.startswith("model_")
            ):
                raise TypeError(
                    f"field {field_name} is defined without type annotation"
                )

        # Validate fields types and substitute bson fields
        for field_name, field_type in annotations.items():
            if not is_dunder(field_name) and should_touch_field(type_=field_type):
                substituted_type = validate_type(field_type)
                annotations[field_name] = substituted_type
                # Handle BSON serialized fields after substitution to allow some
                # builtin substitutions
                bson_serializer = _get_bson_serializer(substituted_type)
                if bson_serializer is not None:
                    bson_serializers[field_name] = bson_serializer

        # Validate fields
        for field_name, field_type in annotations.items():
            value = namespace.get(field_name, Undefined)

            if is_dunder(field_name) or not should_touch_field(value, field_type):
                continue  # pragma: no cover
                # https://github.com/nedbat/coveragepy/issues/198

            if isinstance(value, PDFieldInfo):
                raise TypeError("please use odmantic.Field instead of pydantic.Field")

            if is_type_mutable(field_type):
                mutable_fields.add(field_name)

            if lenient_issubclass(field_type, EmbeddedModel):
                if isinstance(value, ODMFieldInfo):
                    namespace[field_name] = value.pydantic_field_info
                    key_name = (
                        value.key_name if value.key_name is not None else field_name
                    )
                    primary_field = value.primary_field
                    index = value.index
                    unique = value.unique
                else:
                    key_name = field_name
                    primary_field = False
                    index = False
                    unique = False

                odm_fields[field_name] = ODMEmbedded(
                    primary_field=primary_field,
                    model=field_type,
                    key_name=key_name,
                    model_config=config,
                    index=index,
                    unique=unique,
                )

            elif is_type_argument_subclass(field_type, EmbeddedModel):
                if isinstance(value, ODMFieldInfo):
                    if value.primary_field:
                        raise TypeError(
                            "Declaring a generic type of embedded models as a primary "
                            f"field is not allowed: {field_name} in {name}"
                        )
                    namespace[field_name] = value.pydantic_field_info
                    key_name = (
                        value.key_name if value.key_name is not None else field_name
                    )
                    index = value.index
                    unique = value.unique
                else:
                    key_name = field_name
                    index = False
                    unique = False
                model = get_first_type_argument_subclassing(field_type, EmbeddedModel)
                assert model is not None
                if len(model.__references__) > 0:
                    raise TypeError(
                        "Declaring a generic type of embedded models containing "
                        f"references is not allowed: {field_name} in {name}"
                    )
                generic_origin = get_origin(field_type)
                assert generic_origin is not None
                odm_fields[field_name] = ODMEmbeddedGeneric(
                    model=model,
                    generic_origin=generic_origin,
                    key_name=key_name,
                    model_config=config,
                    index=index,
                    unique=unique,
                )

            elif lenient_issubclass(field_type, Model):
                if not isinstance(value, ODMReferenceInfo):
                    raise TypeError(
                        f"cannot define a reference {field_name} (in {name}) without"
                        " a Reference assigned to it"
                    )
                key_name = value.key_name if value.key_name is not None else field_name
                raise_on_invalid_key_name(key_name)
                odm_fields[field_name] = ODMReference(
                    model=field_type, key_name=key_name, model_config=config
                )
                references.append(field_name)
                del namespace[field_name]  # Remove default ODMReferenceInfo value
            else:
                if isinstance(value, ODMFieldInfo):
                    key_name = (
                        value.key_name if value.key_name is not None else field_name
                    )
                    raise_on_invalid_key_name(key_name)
                    odm_fields[field_name] = ODMField(
                        primary_field=value.primary_field,
                        key_name=key_name,
                        model_config=config,
                        index=value.index,
                        unique=value.unique,
                    )
                    namespace[field_name] = value.pydantic_field_info

                elif value is Undefined:
                    odm_fields[field_name] = ODMField(
                        primary_field=False, key_name=field_name, model_config=config
                    )

                else:
                    try:
                        TypeAdapter(field_type).validate_python(value)
                    except ValidationError:
                        raise TypeError(
                            f"Unhandled field definition {name}: {repr(field_type)}"
                            f" = {repr(value)}"
                        )
                    odm_fields[field_name] = ODMField(
                        primary_field=False, key_name=field_name, model_config=config
                    )

        # NOTE: Duplicate key detection make sur that at most one primary key is
        # defined
        duplicate_key = find_duplicate_key(odm_fields.values())
        if duplicate_key is not None:
            raise TypeError(f"Duplicated key_name: {duplicate_key} in {name}")

        namespace["__annotations__"] = annotations
        namespace["__odm_fields__"] = odm_fields
        namespace["__references__"] = tuple(references)
        namespace["__bson_serializers__"] = bson_serializers
        namespace["__mutable_fields__"] = frozenset(mutable_fields)
        namespace["model_config"] = config

    @no_type_check
    def __new__(
        mcs,
        name: str,
        bases: Tuple[type, ...],
        namespace: Dict[str, Any],
        **kwargs: Any,
    ):
        is_custom_cls = namespace.get(
            "__module__"
        ) != "odmantic.model" and namespace.get("__qualname__") not in (
            "_BaseODMModel",
            "Model",
            "EmbeddedModel",
        )

        if is_custom_cls:
            # Handle calls from pydantic.main.create_model (used internally by FastAPI)
            patched_bases = []
            for b in bases:
                if hasattr(b, "__pydantic_model__"):
                    patched_bases.append(b.__pydantic_model__)
                else:
                    patched_bases.append(b)
            bases = tuple(patched_bases)
            # Nullify unset docstring (to avoid getting the docstrings from the parent
            # classes)
            if namespace.get("__doc__", None) is None:
                namespace["__doc__"] = ""

        cls = super().__new__(mcs, name, bases, namespace, **kwargs)

        if is_custom_cls:
            config: ODMConfigDict = namespace["model_config"]
            # Patch Model related fields to build a "pure" pydantic model
            odm_fields: Dict[str, ODMBaseField] = namespace["__odm_fields__"]
            for field_name, field in odm_fields.items():
                if isinstance(field, (ODMReference, ODMEmbedded)):
                    namespace["__annotations__"][field_name] = (
                        field.model.__pydantic_model__
                    )
            # Build the pydantic model
            pydantic_cls = (
                pydantic._internal._model_construction.ModelMetaclass.__new__(
                    mcs,
                    f"{name}.__pydantic_model__",
                    (BaseBSONModel,),
                    namespace,
                    **kwargs,
                )
            )
            # Change the title to generate clean JSON schemas from this "pure" model
            if config["title"] is None:
                pydantic_cls.model_config["title"] = name
            cls.__pydantic_model__ = pydantic_cls

            for name, field in cls.__odm_fields__.items():
                field.bind_pydantic_field(cls.model_fields[name])
                setattr(cls, name, FieldProxy(parent=None, field=field))

        return cls


@dataclass_transform(kw_only_default=True, field_specifiers=(Field, ODMFieldInfo))
class ModelMetaclass(BaseModelMetaclass):
    @no_type_check
    def __new__(  # noqa C901
        mcs,
        name: str,
        bases: Tuple[type, ...],
        namespace: Dict[str, Any],
        **kwargs: Any,
    ):
        if namespace.get("__module__") != "odmantic.model" and namespace.get(
            "__qualname__"
        ) not in ("_BaseODMModel", "Model"):
            mcs.__validate_cls_namespace__(name, namespace)
            config: ODMConfigDict = namespace["model_config"]
            primary_field: Optional[str] = None
            odm_fields: Dict[str, ODMBaseField] = namespace["__odm_fields__"]

            for field_name, field in odm_fields.items():
                if isinstance(field, ODMField) and field.primary_field:
                    primary_field = field_name
                    break

            if primary_field is None:
                if "id" in odm_fields:
                    raise TypeError(
                        "can't automatically generate a primary field since an 'id' "
                        "field already exists"
                    )
                primary_field = "id"
                odm_fields["id"] = ODMField(
                    primary_field=True, key_name="_id", model_config=config
                )
                namespace["id"] = PDField(default_factory=ObjectId)
                namespace["__annotations__"]["id"] = ObjectId

            namespace["__primary_field__"] = primary_field

            if config["collection"] is not None:
                collection_name = config["collection"]
            else:
                cls_name = name
                if cls_name.endswith("Model"):
                    # TODO document this
                    cls_name = cls_name[:-5]  # Strip Model in the class name
                collection_name = to_snake_case(cls_name)
            raise_on_invalid_collection_name(collection_name, cls_name=name)
            namespace["__collection__"] = collection_name

        return super().__new__(mcs, name, bases, namespace, **kwargs)

    def __pos__(cls) -> str:
        return cast(str, getattr(cls, "__collection__"))


@dataclass_transform(kw_only_default=True, field_specifiers=(Field, ODMFieldInfo))
class EmbeddedModelMetaclass(BaseModelMetaclass):
    @no_type_check
    def __new__(
        mcs,
        name: str,
        bases: Tuple[type, ...],
        namespace: Dict[str, Any],
        **kwargs: Any,
    ):
        if namespace.get("__module__") != "odmantic.model" and namespace.get(
            "__qualname__"
        ) not in ("_BaseODMModel", "EmbeddedModel"):
            mcs.__validate_cls_namespace__(name, namespace)
            odm_fields: Dict[str, ODMBaseField] = namespace["__odm_fields__"]
            for field in odm_fields.values():
                if isinstance(field, ODMField) and field.primary_field:
                    raise TypeError(
                        f"cannot define a primary field in {name} embedded document"
                    )

        return super().__new__(mcs, name, bases, namespace, **kwargs)


BaseT = TypeVar("BaseT", bound="_BaseODMModel")


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


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


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.
    """