mixins

Package contains functionalities divided by features. All mixins are combined into ModelTableProxy which is one of the parents of Model. The split into mixins was done to ease the maintainability of the proxy class, as it became quite complicated over time.

AliasMixin

Used to translate field names into database column names.

Source code in ormar\models\mixins\alias_mixin.py

class AliasMixin:
    """
    Used to translate field names into database column names.
    """

    if TYPE_CHECKING:  # pragma: no cover
        from ormar.models.ormar_config import OrmarConfig

        ormar_config: OrmarConfig

    @classmethod
    def get_column_alias(cls, field_name: str) -> str:
        """
        Returns db alias (column name in db) for given ormar field.
        For fields without alias field name is returned.
        :param field_name: name of the field to get alias from
        :type field_name: str
        :return: alias (db name) if set, otherwise passed name
        :rtype: str
        """
        field = cls.ormar_config.model_fields.get(field_name)
        return field.get_alias() if field is not None else field_name

    @classmethod
    def get_column_name_from_alias(cls, alias: str) -> str:
        """
        Returns ormar field name for given db alias (column name in db).
        If field do not have alias it's returned as is.
        :param alias:
        :type alias: str
        :return: field name if set, otherwise passed alias (db name)
        :rtype: str
        """
        for field_name, field in cls.ormar_config.model_fields.items():
            if field.get_alias() == alias:
                return field_name
        return alias  # if not found it's not an alias but actual name

    @classmethod
    def translate_columns_to_aliases(cls, new_kwargs: Dict) -> Dict:
        """
        Translates dictionary of model fields changing field names into aliases.
        If field has no alias the field name remains intact.
        Only fields present in the dictionary are translated.
        :param new_kwargs: dict with fields names and their values
        :type new_kwargs: Dict
        :return: dict with aliases and their values
        :rtype: Dict
        """
        for field_name, field in cls.ormar_config.model_fields.items():
            if field_name in new_kwargs:
                new_kwargs[field.get_alias()] = new_kwargs.pop(field_name)
        return new_kwargs

    @classmethod
    def translate_aliases_to_columns(cls, new_kwargs: Dict) -> Dict:
        """
        Translates dictionary of model fields changing aliases into field names.
        If field has no alias the alias is already a field name.
        Only fields present in the dictionary are translated.
        :param new_kwargs: dict with aliases and their values
        :type new_kwargs: Dict
        :return: dict with fields names and their values
        :rtype: Dict
        """
        for field_name, field in cls.ormar_config.model_fields.items():
            if field.get_alias() and field.get_alias() in new_kwargs:
                new_kwargs[field_name] = new_kwargs.pop(field.get_alias())
        return new_kwargs

get_column_alias(field_name) classmethod

Returns db alias (column name in db) for given ormar field. For fields without alias field name is returned.

Parameters:

Name	Type	Description	Default
field_name	str	name of the field to get alias from	required

Returns:

Type	Description
str	alias (db name) if set, otherwise passed name

Source code in ormar\models\mixins\alias_mixin.py

@classmethod
def get_column_alias(cls, field_name: str) -> str:
    """
    Returns db alias (column name in db) for given ormar field.
    For fields without alias field name is returned.
    :param field_name: name of the field to get alias from
    :type field_name: str
    :return: alias (db name) if set, otherwise passed name
    :rtype: str
    """
    field = cls.ormar_config.model_fields.get(field_name)
    return field.get_alias() if field is not None else field_name

get_column_name_from_alias(alias) classmethod

Returns ormar field name for given db alias (column name in db). If field do not have alias it's returned as is.

Parameters:

Name	Type	Description	Default
alias	str		required

Returns:

Type	Description
str	field name if set, otherwise passed alias (db name)

Source code in ormar\models\mixins\alias_mixin.py

@classmethod
def get_column_name_from_alias(cls, alias: str) -> str:
    """
    Returns ormar field name for given db alias (column name in db).
    If field do not have alias it's returned as is.
    :param alias:
    :type alias: str
    :return: field name if set, otherwise passed alias (db name)
    :rtype: str
    """
    for field_name, field in cls.ormar_config.model_fields.items():
        if field.get_alias() == alias:
            return field_name
    return alias  # if not found it's not an alias but actual name

translate_aliases_to_columns(new_kwargs) classmethod

Translates dictionary of model fields changing aliases into field names. If field has no alias the alias is already a field name. Only fields present in the dictionary are translated.

Parameters:

Name	Type	Description	Default
new_kwargs	Dict	dict with aliases and their values	required

Returns:

Type	Description
Dict	dict with fields names and their values

Source code in ormar\models\mixins\alias_mixin.py

@classmethod
def translate_aliases_to_columns(cls, new_kwargs: Dict) -> Dict:
    """
    Translates dictionary of model fields changing aliases into field names.
    If field has no alias the alias is already a field name.
    Only fields present in the dictionary are translated.
    :param new_kwargs: dict with aliases and their values
    :type new_kwargs: Dict
    :return: dict with fields names and their values
    :rtype: Dict
    """
    for field_name, field in cls.ormar_config.model_fields.items():
        if field.get_alias() and field.get_alias() in new_kwargs:
            new_kwargs[field_name] = new_kwargs.pop(field.get_alias())
    return new_kwargs

translate_columns_to_aliases(new_kwargs) classmethod

Translates dictionary of model fields changing field names into aliases. If field has no alias the field name remains intact. Only fields present in the dictionary are translated.

Parameters:

Name	Type	Description	Default
new_kwargs	Dict	dict with fields names and their values	required

Returns:

Type	Description
Dict	dict with aliases and their values

Source code in ormar\models\mixins\alias_mixin.py

@classmethod
def translate_columns_to_aliases(cls, new_kwargs: Dict) -> Dict:
    """
    Translates dictionary of model fields changing field names into aliases.
    If field has no alias the field name remains intact.
    Only fields present in the dictionary are translated.
    :param new_kwargs: dict with fields names and their values
    :type new_kwargs: Dict
    :return: dict with aliases and their values
    :rtype: Dict
    """
    for field_name, field in cls.ormar_config.model_fields.items():
        if field_name in new_kwargs:
            new_kwargs[field.get_alias()] = new_kwargs.pop(field_name)
    return new_kwargs

ExcludableMixin

Bases: RelationMixin

Used to include/exclude given set of fields on models during load and dict() calls.

Source code in ormar\models\mixins\excludable_mixin.py

class ExcludableMixin(RelationMixin):
    """
    Used to include/exclude given set of fields on models during load and dict() calls.
    """

    if TYPE_CHECKING:  # pragma: no cover
        from ormar import Model
        from ormar.models import ModelRow

    @staticmethod
    def get_child(
        items: Union[Set, Dict, None], key: Optional[str] = None
    ) -> Union[Set, Dict, None]:
        """
        Used to get nested dictionaries keys if they exists otherwise returns
        passed items.
        :param items: bag of items to include or exclude
        :type items:  Union[Set, Dict, None]
        :param key: name of the child to extract
        :type key: str
        :return: child extracted from items if exists
        :rtype: Union[Set, Dict, None]
        """
        if isinstance(items, dict):
            return items.get(key, {})
        return items

    @staticmethod
    def _populate_pk_column(
        model: Union[Type["Model"], Type["ModelRow"]],
        columns: List[str],
        use_alias: bool = False,
    ) -> List[str]:
        """
        Adds primary key column/alias (depends on use_alias flag) to list of
        column names that are selected.

        :param model: model on columns are selected
        :type model: Type["Model"]
        :param columns: list of columns names
        :type columns: List[str]
        :param use_alias: flag to set if aliases or field names should be used
        :type use_alias: bool
        :return: list of columns names with pk column in it
        :rtype: List[str]
        """
        pk_alias = (
            model.get_column_alias(model.ormar_config.pkname)
            if use_alias
            else model.ormar_config.pkname
        )
        if pk_alias not in columns:
            columns.append(pk_alias)
        return columns

    @classmethod
    def own_table_columns(
        cls,
        model: Union[Type["Model"], Type["ModelRow"]],
        excludable: ExcludableItems,
        alias: str = "",
        use_alias: bool = False,
        add_pk_columns: bool = True,
    ) -> List[str]:
        """
        Returns list of aliases or field names for given model.
        Aliases/names switch is use_alias flag.

        If provided only fields included in fields will be returned.
        If provided fields in exclude_fields will be excluded in return.

        Primary key field is always added and cannot be excluded (will be added anyway).

        :param add_pk_columns: flag if add primary key - always yes if ormar parses data
        :type add_pk_columns: bool
        :param alias: relation prefix
        :type alias: str
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param model: model on columns are selected
        :type model: Type["Model"]
        :param use_alias: flag if aliases or field names should be used
        :type use_alias: bool
        :return: list of column field names or aliases
        :rtype: List[str]
        """
        model_excludable = excludable.get(model_cls=model, alias=alias)  # type: ignore
        columns = [
            model.get_column_name_from_alias(col.name) if not use_alias else col.name
            for col in model.ormar_config.table.columns
        ]
        field_names = [
            model.get_column_name_from_alias(col.name)
            for col in model.ormar_config.table.columns
        ]
        if model_excludable.include:
            columns = [
                col
                for col, name in zip(columns, field_names)
                if model_excludable.is_included(name)
            ]
        if model_excludable.exclude:
            columns = [
                col
                for col, name in zip(columns, field_names)
                if not model_excludable.is_excluded(name)
            ]

        # always has to return pk column for ormar to work
        if add_pk_columns:
            columns = cls._populate_pk_column(
                model=model, columns=columns, use_alias=use_alias
            )

        return columns

    @classmethod
    def _update_excluded_with_related(cls, exclude: Union[Set, Dict, None]) -> Set:
        """
        Used during generation of the dict().
        To avoid cyclical references and max recurrence limit nested models have to
        exclude related models that are not mandatory.

        For a main model (not nested) only nullable related field names are added to
        exclusion, for nested models all related models are excluded.

        :param exclude: set/dict with fields to exclude
        :type exclude: Union[Set, Dict, None]
        :return: set or dict with excluded fields added.
        :rtype: Union[Set, Dict]
        """
        exclude = exclude or set()
        related_set = cls.extract_related_names()
        if isinstance(exclude, set):
            exclude = {s for s in exclude}
            exclude = exclude.union(related_set)
        elif isinstance(exclude, dict):
            # relations are handled in ormar - take only own fields (ellipsis in dict)
            exclude = {k for k, v in exclude.items() if v is Ellipsis}
            exclude = exclude.union(related_set)
        return exclude

    @classmethod
    def _update_excluded_with_pks_and_through(
        cls, exclude: Set, exclude_primary_keys: bool, exclude_through_models: bool
    ) -> Set:
        """
        Updates excluded names with name of pk column if exclude flag is set.

        :param exclude: set of names to exclude
        :type exclude: Set
        :param exclude_primary_keys: flag if the primary keys should be excluded
        :type exclude_primary_keys: bool
        :return: set updated with pk if flag is set
        :rtype: Set
        """
        if exclude_primary_keys:
            exclude.add(cls.ormar_config.pkname)
        if exclude_through_models:
            exclude = exclude.union(cls.extract_through_names())
        return exclude

    @classmethod
    def get_names_to_exclude(cls, excludable: ExcludableItems, alias: str) -> Set:
        """
        Returns a set of models field names that should be explicitly excluded
        during model initialization.

        Those fields will be set to None to avoid ormar/pydantic setting default
        values on them. They should be returned as None in any case.

        Used in parsing data from database rows that construct Models by initializing
        them with dicts constructed from those db rows.

        :param alias: alias of current relation
        :type alias: str
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :return: set of field names that should be excluded
        :rtype: Set
        """
        model = cast(Type["Model"], cls)
        model_excludable = excludable.get(model_cls=model, alias=alias)
        fields_names = cls.extract_db_own_fields()
        if model_excludable.include:
            fields_to_keep = model_excludable.include.intersection(fields_names)
        else:
            fields_to_keep = fields_names

        fields_to_exclude = fields_names - fields_to_keep

        if model_excludable.exclude:
            fields_to_exclude = fields_to_exclude.union(
                model_excludable.exclude.intersection(fields_names)
            )
        fields_to_exclude = fields_to_exclude - {cls.ormar_config.pkname}

        return fields_to_exclude

Model

Bases: ModelRow

Source code in ormar\models\model.py

class Model(ModelRow):
    __abstract__ = False
    if TYPE_CHECKING:  # pragma nocover
        ormar_config: OrmarConfig

    def __repr__(self) -> str:  # pragma nocover
        _repr = {
            k: getattr(self, k)
            for k, v in self.ormar_config.model_fields.items()
            if not v.skip_field
        }
        return f"{self.__class__.__name__}({str(_repr)})"

    async def upsert(self: T, **kwargs: Any) -> T:
        """
        Performs either a save or an update depending on the presence of the pk.
        If the pk field is filled it's an update, otherwise the save is performed.
        For save kwargs are ignored, used only in update if provided.

        :param kwargs: list of fields to update
        :type kwargs: Any
        :return: saved Model
        :rtype: Model
        """

        force_save = kwargs.pop("__force_save__", False)
        if force_save:
            expr = self.ormar_config.table.select().where(self.pk_column == self.pk)
            row = await self.ormar_config.database.fetch_one(expr)
            if not row:
                return await self.save()
            return await self.update(**kwargs)

        if not self.pk:
            return await self.save()
        return await self.update(**kwargs)

    async def save(self: T) -> T:
        """
        Performs a save of given Model instance.
        If primary key is already saved, db backend will throw integrity error.

        Related models are saved by pk number, reverse relation and many to many fields
        are not saved - use corresponding relations methods.

        If there are fields with server_default set and those fields
        are not already filled save will trigger also a second query
        to refreshed the fields populated server side.

        Does not recognize if model was previously saved.
        If you want to perform update or insert depending on the pk
        fields presence use upsert.

        Sends pre_save and post_save signals.

        Sets model save status to True.

        :return: saved Model
        :rtype: Model
        """
        await self.signals.pre_save.send(sender=self.__class__, instance=self)
        self_fields = self._extract_model_db_fields()

        if (
            not self.pk
            and self.ormar_config.model_fields[self.ormar_config.pkname].autoincrement
        ):
            self_fields.pop(self.ormar_config.pkname, None)
        self_fields = self.populate_default_values(self_fields)
        self.update_from_dict(
            {
                k: v
                for k, v in self_fields.items()
                if k not in self.extract_related_names()
            }
        )

        self_fields = self.translate_columns_to_aliases(self_fields)
        expr = self.ormar_config.table.insert()
        expr = expr.values(**self_fields)

        pk = await self.ormar_config.database.execute(expr)
        if pk and isinstance(pk, self.pk_type()):
            setattr(self, self.ormar_config.pkname, pk)

        self.set_save_status(True)
        # refresh server side defaults
        if any(
            field.server_default is not None
            for name, field in self.ormar_config.model_fields.items()
            if name not in self_fields
        ):
            await self.load()

        await self.signals.post_save.send(sender=self.__class__, instance=self)
        return self

    async def save_related(  # noqa: CCR001, CFQ002
        self,
        follow: bool = False,
        save_all: bool = False,
        relation_map: Optional[Dict] = None,
        exclude: Union[Set, Dict, None] = None,
        update_count: int = 0,
        previous_model: Optional["Model"] = None,
        relation_field: Optional["ForeignKeyField"] = None,
    ) -> int:
        """
        Triggers a upsert method on all related models
        if the instances are not already saved.
        By default saves only the directly related ones.

        If follow=True is set it saves also related models of related models.

        To not get stuck in an infinite loop as related models also keep a relation
        to parent model visited models set is kept.

        That way already visited models that are nested are saved, but the save do not
        follow them inside. So Model A -> Model B -> Model A -> Model C will save second
        Model A but will never follow into Model C.
        Nested relations of those kind need to be persisted manually.

        :param relation_field: field with relation leading to this model
        :type relation_field: Optional[ForeignKeyField]
        :param previous_model: previous model from which method came
        :type previous_model: Model
        :param exclude: items to exclude during saving of relations
        :type exclude: Union[Set, Dict]
        :param relation_map: map of relations to follow
        :type relation_map: Dict
        :param save_all: flag if all models should be saved or only not saved ones
        :type save_all: bool
        :param follow: flag to trigger deep save -
        by default only directly related models are saved
        with follow=True also related models of related models are saved
        :type follow: bool
        :param update_count: internal parameter for recursive calls -
        number of updated instances
        :type update_count: int
        :return: number of updated/saved models
        :rtype: int
        """
        relation_map = (
            relation_map
            if relation_map is not None
            else translate_list_to_dict(self._iterate_related_models())
        )
        if exclude and isinstance(exclude, Set):
            exclude = translate_list_to_dict(exclude)
        relation_map = subtract_dict(relation_map, exclude or {})

        if relation_map:
            fields_to_visit = {
                field
                for field in self.extract_related_fields()
                if field.name in relation_map
            }
            pre_save = {
                field
                for field in fields_to_visit
                if not field.virtual and not field.is_multi
            }

            update_count = await self._update_relation_list(
                fields_list=pre_save,
                follow=follow,
                save_all=save_all,
                relation_map=relation_map,
                update_count=update_count,
            )

            update_count = await self._upsert_model(
                instance=self,
                save_all=save_all,
                previous_model=previous_model,
                relation_field=relation_field,
                update_count=update_count,
            )

            post_save = fields_to_visit - pre_save

            update_count = await self._update_relation_list(
                fields_list=post_save,
                follow=follow,
                save_all=save_all,
                relation_map=relation_map,
                update_count=update_count,
            )

        else:
            update_count = await self._upsert_model(
                instance=self,
                save_all=save_all,
                previous_model=previous_model,
                relation_field=relation_field,
                update_count=update_count,
            )

        return update_count

    async def update(self: T, _columns: Optional[List[str]] = None, **kwargs: Any) -> T:
        """
        Performs update of Model instance in the database.
        Fields can be updated before or you can pass them as kwargs.

        Sends pre_update and post_update signals.

        Sets model save status to True.

        :param _columns: list of columns to update, if None all are updated
        :type _columns: List
        :raises ModelPersistenceError: If the pk column is not set

        :param kwargs: list of fields to update as field=value pairs
        :type kwargs: Any
        :return: updated Model
        :rtype: Model
        """
        if kwargs:
            self.update_from_dict(kwargs)

        if not self.pk:
            raise ModelPersistenceError(
                "You cannot update not saved model! Use save or upsert method."
            )

        await self.signals.pre_update.send(
            sender=self.__class__, instance=self, passed_args=kwargs
        )
        self_fields = self._extract_model_db_fields()
        self_fields.pop(self.get_column_name_from_alias(self.ormar_config.pkname))
        if _columns:
            self_fields = {k: v for k, v in self_fields.items() if k in _columns}
        if self_fields:
            self_fields = self.translate_columns_to_aliases(self_fields)
            expr = self.ormar_config.table.update().values(**self_fields)
            expr = expr.where(self.pk_column == getattr(self, self.ormar_config.pkname))

            await self.ormar_config.database.execute(expr)
        self.set_save_status(True)
        await self.signals.post_update.send(sender=self.__class__, instance=self)
        return self

    async def delete(self) -> int:
        """
        Removes the Model instance from the database.

        Sends pre_delete and post_delete signals.

        Sets model save status to False.

        Note it does not delete the Model itself (python object).
        So you can delete and later save (since pk is deleted no conflict will arise)
        or update and the Model will be saved in database again.

        :return: number of deleted rows (for some backends)
        :rtype: int
        """
        await self.signals.pre_delete.send(sender=self.__class__, instance=self)
        expr = self.ormar_config.table.delete()
        expr = expr.where(self.pk_column == (getattr(self, self.ormar_config.pkname)))
        result = await self.ormar_config.database.execute(expr)
        self.set_save_status(False)
        await self.signals.post_delete.send(sender=self.__class__, instance=self)
        return result

    async def load(self: T) -> T:
        """
        Allow to refresh existing Models fields from database.
        Be careful as the related models can be overwritten by pk_only models in load.
        Does NOT refresh the related models fields if they were loaded before.

        :raises NoMatch: If given pk is not found in database.

        :return: reloaded Model
        :rtype: Model
        """
        expr = self.ormar_config.table.select().where(self.pk_column == self.pk)
        row = await self.ormar_config.database.fetch_one(expr)
        if not row:  # pragma nocover
            raise NoMatch("Instance was deleted from database and cannot be refreshed")
        kwargs = dict(row)
        kwargs = self.translate_aliases_to_columns(kwargs)
        self.update_from_dict(kwargs)
        self.set_save_status(True)
        return self

    async def load_all(
        self: T,
        follow: bool = False,
        exclude: Union[List, str, Set, Dict, None] = None,
        order_by: Union[List, str, None] = None,
    ) -> T:
        """
        Allow to refresh existing Models fields from database.
        Performs refresh of the related models fields.

        By default, loads only self and the directly related ones.

        If follow=True is set it loads also related models of related models.

        To not get stuck in an infinite loop as related models also keep a relation
        to parent model visited models set is kept.

        That way already visited models that are nested are loaded, but the load do not
        follow them inside. So Model A -> Model B -> Model C -> Model A -> Model X
        will load second Model A but will never follow into Model X.
        Nested relations of those kind need to be loaded manually.

        :param order_by: columns by which models should be sorted
        :type order_by: Union[List, str]
        :raises NoMatch: If given pk is not found in database.

        :param exclude: related models to exclude
        :type exclude: Union[List, str, Set, Dict]
        :param follow: flag to trigger deep save -
        by default only directly related models are saved
        with follow=True also related models of related models are saved
        :type follow: bool
        :return: reloaded Model
        :rtype: Model
        """
        relations = list(self.extract_related_names())
        if follow:
            relations = self._iterate_related_models()
        queryset = self.__class__.objects
        if exclude:
            queryset = queryset.exclude_fields(exclude)
        if order_by:
            queryset = queryset.order_by(order_by)
        instance = await queryset.select_related(relations).get(pk=self.pk)
        self._orm.clear()
        self.update_from_dict(instance.model_dump())
        return self

delete() async

Removes the Model instance from the database.

Sends pre_delete and post_delete signals.

Sets model save status to False.

Note it does not delete the Model itself (python object). So you can delete and later save (since pk is deleted no conflict will arise) or update and the Model will be saved in database again.

Returns:

Type	Description
int	number of deleted rows (for some backends)

Source code in ormar\models\model.py

async def delete(self) -> int:
    """
    Removes the Model instance from the database.

    Sends pre_delete and post_delete signals.

    Sets model save status to False.

    Note it does not delete the Model itself (python object).
    So you can delete and later save (since pk is deleted no conflict will arise)
    or update and the Model will be saved in database again.

    :return: number of deleted rows (for some backends)
    :rtype: int
    """
    await self.signals.pre_delete.send(sender=self.__class__, instance=self)
    expr = self.ormar_config.table.delete()
    expr = expr.where(self.pk_column == (getattr(self, self.ormar_config.pkname)))
    result = await self.ormar_config.database.execute(expr)
    self.set_save_status(False)
    await self.signals.post_delete.send(sender=self.__class__, instance=self)
    return result

load() async

Allow to refresh existing Models fields from database. Be careful as the related models can be overwritten by pk_only models in load. Does NOT refresh the related models fields if they were loaded before.

Returns:

Type	Description
Model	reloaded Model

Raises:

Type	Description
NoMatch	If given pk is not found in database.

Source code in ormar\models\model.py

async def load(self: T) -> T:
    """
    Allow to refresh existing Models fields from database.
    Be careful as the related models can be overwritten by pk_only models in load.
    Does NOT refresh the related models fields if they were loaded before.

    :raises NoMatch: If given pk is not found in database.

    :return: reloaded Model
    :rtype: Model
    """
    expr = self.ormar_config.table.select().where(self.pk_column == self.pk)
    row = await self.ormar_config.database.fetch_one(expr)
    if not row:  # pragma nocover
        raise NoMatch("Instance was deleted from database and cannot be refreshed")
    kwargs = dict(row)
    kwargs = self.translate_aliases_to_columns(kwargs)
    self.update_from_dict(kwargs)
    self.set_save_status(True)
    return self

load_all(follow=False, exclude=None, order_by=None) async

Allow to refresh existing Models fields from database. Performs refresh of the related models fields.

By default, loads only self and the directly related ones.

If follow=True is set it loads also related models of related models.

To not get stuck in an infinite loop as related models also keep a relation to parent model visited models set is kept.

That way already visited models that are nested are loaded, but the load do not follow them inside. So Model A -> Model B -> Model C -> Model A -> Model X will load second Model A but will never follow into Model X. Nested relations of those kind need to be loaded manually.

Parameters:

Name	Type	Description	Default
order_by	Union[List, str, None]	columns by which models should be sorted	None
exclude	Union[List, str, Set, Dict, None]	related models to exclude	None
follow	bool	flag to trigger deep save - by default only directly related models are saved with follow=True also related models of related models are saved	False

Returns:

Type	Description
Model	reloaded Model

Raises:

Type	Description
NoMatch	If given pk is not found in database.

Source code in ormar\models\model.py

async def load_all(
    self: T,
    follow: bool = False,
    exclude: Union[List, str, Set, Dict, None] = None,
    order_by: Union[List, str, None] = None,
) -> T:
    """
    Allow to refresh existing Models fields from database.
    Performs refresh of the related models fields.

    By default, loads only self and the directly related ones.

    If follow=True is set it loads also related models of related models.

    To not get stuck in an infinite loop as related models also keep a relation
    to parent model visited models set is kept.

    That way already visited models that are nested are loaded, but the load do not
    follow them inside. So Model A -> Model B -> Model C -> Model A -> Model X
    will load second Model A but will never follow into Model X.
    Nested relations of those kind need to be loaded manually.

    :param order_by: columns by which models should be sorted
    :type order_by: Union[List, str]
    :raises NoMatch: If given pk is not found in database.

    :param exclude: related models to exclude
    :type exclude: Union[List, str, Set, Dict]
    :param follow: flag to trigger deep save -
    by default only directly related models are saved
    with follow=True also related models of related models are saved
    :type follow: bool
    :return: reloaded Model
    :rtype: Model
    """
    relations = list(self.extract_related_names())
    if follow:
        relations = self._iterate_related_models()
    queryset = self.__class__.objects
    if exclude:
        queryset = queryset.exclude_fields(exclude)
    if order_by:
        queryset = queryset.order_by(order_by)
    instance = await queryset.select_related(relations).get(pk=self.pk)
    self._orm.clear()
    self.update_from_dict(instance.model_dump())
    return self

save() async

Performs a save of given Model instance. If primary key is already saved, db backend will throw integrity error.

Related models are saved by pk number, reverse relation and many to many fields are not saved - use corresponding relations methods.

If there are fields with server_default set and those fields are not already filled save will trigger also a second query to refreshed the fields populated server side.

Does not recognize if model was previously saved. If you want to perform update or insert depending on the pk fields presence use upsert.

Sends pre_save and post_save signals.

Sets model save status to True.

Returns:

Type	Description
Model	saved Model

Source code in ormar\models\model.py

async def save(self: T) -> T:
    """
    Performs a save of given Model instance.
    If primary key is already saved, db backend will throw integrity error.

    Related models are saved by pk number, reverse relation and many to many fields
    are not saved - use corresponding relations methods.

    If there are fields with server_default set and those fields
    are not already filled save will trigger also a second query
    to refreshed the fields populated server side.

    Does not recognize if model was previously saved.
    If you want to perform update or insert depending on the pk
    fields presence use upsert.

    Sends pre_save and post_save signals.

    Sets model save status to True.

    :return: saved Model
    :rtype: Model
    """
    await self.signals.pre_save.send(sender=self.__class__, instance=self)
    self_fields = self._extract_model_db_fields()

    if (
        not self.pk
        and self.ormar_config.model_fields[self.ormar_config.pkname].autoincrement
    ):
        self_fields.pop(self.ormar_config.pkname, None)
    self_fields = self.populate_default_values(self_fields)
    self.update_from_dict(
        {
            k: v
            for k, v in self_fields.items()
            if k not in self.extract_related_names()
        }
    )

    self_fields = self.translate_columns_to_aliases(self_fields)
    expr = self.ormar_config.table.insert()
    expr = expr.values(**self_fields)

    pk = await self.ormar_config.database.execute(expr)
    if pk and isinstance(pk, self.pk_type()):
        setattr(self, self.ormar_config.pkname, pk)

    self.set_save_status(True)
    # refresh server side defaults
    if any(
        field.server_default is not None
        for name, field in self.ormar_config.model_fields.items()
        if name not in self_fields
    ):
        await self.load()

    await self.signals.post_save.send(sender=self.__class__, instance=self)
    return self

save_related(follow=False, save_all=False, relation_map=None, exclude=None, update_count=0, previous_model=None, relation_field=None) async

Triggers a upsert method on all related models if the instances are not already saved. By default saves only the directly related ones.

If follow=True is set it saves also related models of related models.

To not get stuck in an infinite loop as related models also keep a relation to parent model visited models set is kept.

That way already visited models that are nested are saved, but the save do not follow them inside. So Model A -> Model B -> Model A -> Model C will save second Model A but will never follow into Model C. Nested relations of those kind need to be persisted manually.

Parameters:

Name	Type	Description	Default
relation_field	Optional[ForeignKeyField]	field with relation leading to this model	None
previous_model	Optional[Model]	previous model from which method came	None
exclude	Union[Set, Dict, None]	items to exclude during saving of relations	None
relation_map	Optional[Dict]	map of relations to follow	None
save_all	bool	flag if all models should be saved or only not saved ones	False
follow	bool	flag to trigger deep save - by default only directly related models are saved with follow=True also related models of related models are saved	False
update_count	int	internal parameter for recursive calls - number of updated instances	0

Returns:

Type	Description
int	number of updated/saved models

Source code in ormar\models\model.py

async def save_related(  # noqa: CCR001, CFQ002
    self,
    follow: bool = False,
    save_all: bool = False,
    relation_map: Optional[Dict] = None,
    exclude: Union[Set, Dict, None] = None,
    update_count: int = 0,
    previous_model: Optional["Model"] = None,
    relation_field: Optional["ForeignKeyField"] = None,
) -> int:
    """
    Triggers a upsert method on all related models
    if the instances are not already saved.
    By default saves only the directly related ones.

    If follow=True is set it saves also related models of related models.

    To not get stuck in an infinite loop as related models also keep a relation
    to parent model visited models set is kept.

    That way already visited models that are nested are saved, but the save do not
    follow them inside. So Model A -> Model B -> Model A -> Model C will save second
    Model A but will never follow into Model C.
    Nested relations of those kind need to be persisted manually.

    :param relation_field: field with relation leading to this model
    :type relation_field: Optional[ForeignKeyField]
    :param previous_model: previous model from which method came
    :type previous_model: Model
    :param exclude: items to exclude during saving of relations
    :type exclude: Union[Set, Dict]
    :param relation_map: map of relations to follow
    :type relation_map: Dict
    :param save_all: flag if all models should be saved or only not saved ones
    :type save_all: bool
    :param follow: flag to trigger deep save -
    by default only directly related models are saved
    with follow=True also related models of related models are saved
    :type follow: bool
    :param update_count: internal parameter for recursive calls -
    number of updated instances
    :type update_count: int
    :return: number of updated/saved models
    :rtype: int
    """
    relation_map = (
        relation_map
        if relation_map is not None
        else translate_list_to_dict(self._iterate_related_models())
    )
    if exclude and isinstance(exclude, Set):
        exclude = translate_list_to_dict(exclude)
    relation_map = subtract_dict(relation_map, exclude or {})

    if relation_map:
        fields_to_visit = {
            field
            for field in self.extract_related_fields()
            if field.name in relation_map
        }
        pre_save = {
            field
            for field in fields_to_visit
            if not field.virtual and not field.is_multi
        }

        update_count = await self._update_relation_list(
            fields_list=pre_save,
            follow=follow,
            save_all=save_all,
            relation_map=relation_map,
            update_count=update_count,
        )

        update_count = await self._upsert_model(
            instance=self,
            save_all=save_all,
            previous_model=previous_model,
            relation_field=relation_field,
            update_count=update_count,
        )

        post_save = fields_to_visit - pre_save

        update_count = await self._update_relation_list(
            fields_list=post_save,
            follow=follow,
            save_all=save_all,
            relation_map=relation_map,
            update_count=update_count,
        )

    else:
        update_count = await self._upsert_model(
            instance=self,
            save_all=save_all,
            previous_model=previous_model,
            relation_field=relation_field,
            update_count=update_count,
        )

    return update_count

update(_columns=None, **kwargs) async

Performs update of Model instance in the database. Fields can be updated before or you can pass them as kwargs.

Sends pre_update and post_update signals.

Sets model save status to True.

Parameters:

Name	Type	Description	Default
_columns	Optional[List[str]]	list of columns to update, if None all are updated	None
kwargs	Any	list of fields to update as field=value pairs	{}

Returns:

Type	Description
Model	updated Model

Raises:

Type	Description
ModelPersistenceError	If the pk column is not set

Source code in ormar\models\model.py

async def update(self: T, _columns: Optional[List[str]] = None, **kwargs: Any) -> T:
    """
    Performs update of Model instance in the database.
    Fields can be updated before or you can pass them as kwargs.

    Sends pre_update and post_update signals.

    Sets model save status to True.

    :param _columns: list of columns to update, if None all are updated
    :type _columns: List
    :raises ModelPersistenceError: If the pk column is not set

    :param kwargs: list of fields to update as field=value pairs
    :type kwargs: Any
    :return: updated Model
    :rtype: Model
    """
    if kwargs:
        self.update_from_dict(kwargs)

    if not self.pk:
        raise ModelPersistenceError(
            "You cannot update not saved model! Use save or upsert method."
        )

    await self.signals.pre_update.send(
        sender=self.__class__, instance=self, passed_args=kwargs
    )
    self_fields = self._extract_model_db_fields()
    self_fields.pop(self.get_column_name_from_alias(self.ormar_config.pkname))
    if _columns:
        self_fields = {k: v for k, v in self_fields.items() if k in _columns}
    if self_fields:
        self_fields = self.translate_columns_to_aliases(self_fields)
        expr = self.ormar_config.table.update().values(**self_fields)
        expr = expr.where(self.pk_column == getattr(self, self.ormar_config.pkname))

        await self.ormar_config.database.execute(expr)
    self.set_save_status(True)
    await self.signals.post_update.send(sender=self.__class__, instance=self)
    return self

upsert(**kwargs) async

Performs either a save or an update depending on the presence of the pk. If the pk field is filled it's an update, otherwise the save is performed. For save kwargs are ignored, used only in update if provided.

Parameters:

Name	Type	Description	Default
kwargs	Any	list of fields to update	{}

Returns:

Type	Description
Model	saved Model

Source code in ormar\models\model.py

async def upsert(self: T, **kwargs: Any) -> T:
    """
    Performs either a save or an update depending on the presence of the pk.
    If the pk field is filled it's an update, otherwise the save is performed.
    For save kwargs are ignored, used only in update if provided.

    :param kwargs: list of fields to update
    :type kwargs: Any
    :return: saved Model
    :rtype: Model
    """

    force_save = kwargs.pop("__force_save__", False)
    if force_save:
        expr = self.ormar_config.table.select().where(self.pk_column == self.pk)
        row = await self.ormar_config.database.fetch_one(expr)
        if not row:
            return await self.save()
        return await self.update(**kwargs)

    if not self.pk:
        return await self.save()
    return await self.update(**kwargs)

ModelRow

Bases: NewBaseModel

Source code in ormar\models\model_row.py

class ModelRow(NewBaseModel):
    @classmethod
    def from_row(  # noqa: CFQ002
        cls,
        row: ResultProxy,
        source_model: Type["Model"],
        select_related: Optional[List] = None,
        related_models: Any = None,
        related_field: Optional["ForeignKeyField"] = None,
        excludable: Optional[ExcludableItems] = None,
        current_relation_str: str = "",
        proxy_source_model: Optional[Type["Model"]] = None,
        used_prefixes: Optional[List[str]] = None,
    ) -> Optional["Model"]:
        """
        Model method to convert raw sql row from database into ormar.Model instance.
        Traverses nested models if they were specified in select_related for query.

        Called recurrently and returns model instance if it's present in the row.
        Note that it's processing one row at a time, so if there are duplicates of
        parent row that needs to be joined/combined
        (like parent row in sql join with 2+ child rows)
        instances populated in this method are later combined in the QuerySet.
        Other method working directly on raw database results is in prefetch_query,
        where rows are populated in a different way as they do not have
        nested models in result.

        :param used_prefixes: list of already extracted prefixes
        :type used_prefixes: List[str]
        :param proxy_source_model: source model from which querysetproxy is constructed
        :type proxy_source_model: Optional[Type["ModelRow"]]
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param current_relation_str: name of the relation field
        :type current_relation_str: str
        :param source_model: model on which relation was defined
        :type source_model: Type[Model]
        :param row: raw result row from the database
        :type row: ResultProxy
        :param select_related: list of names of related models fetched from database
        :type select_related: List
        :param related_models: list or dict of related models
        :type related_models: Union[List, Dict]
        :param related_field: field with relation declaration
        :type related_field: ForeignKeyField
        :return: returns model if model is populated from database
        :rtype: Optional[Model]
        """
        item: Dict[str, Any] = {}
        select_related = select_related or []
        related_models = related_models or []
        table_prefix = ""
        used_prefixes = used_prefixes if used_prefixes is not None else []
        excludable = excludable or ExcludableItems()

        if select_related:
            related_models = group_related_list(select_related)

        if related_field:
            table_prefix = cls._process_table_prefix(
                source_model=source_model,
                current_relation_str=current_relation_str,
                related_field=related_field,
                used_prefixes=used_prefixes,
            )

        item = cls._populate_nested_models_from_row(
            item=item,
            row=row,
            related_models=related_models,
            excludable=excludable,
            current_relation_str=current_relation_str,
            source_model=source_model,  # type: ignore
            proxy_source_model=proxy_source_model,  # type: ignore
            table_prefix=table_prefix,
            used_prefixes=used_prefixes,
        )
        item = cls.extract_prefixed_table_columns(
            item=item, row=row, table_prefix=table_prefix, excludable=excludable
        )

        instance: Optional["Model"] = None
        if item.get(cls.ormar_config.pkname, None) is not None:
            item["__excluded__"] = cls.get_names_to_exclude(
                excludable=excludable, alias=table_prefix
            )
            instance = cast("Model", cls(**item))
            instance.set_save_status(True)
        return instance

    @classmethod
    def _process_table_prefix(
        cls,
        source_model: Type["Model"],
        current_relation_str: str,
        related_field: "ForeignKeyField",
        used_prefixes: List[str],
    ) -> str:
        """

        :param source_model: model on which relation was defined
        :type source_model: Type[Model]
        :param current_relation_str: current relation string
        :type current_relation_str: str
        :param related_field: field with relation declaration
        :type related_field: "ForeignKeyField"
        :param used_prefixes: list of already extracted prefixes
        :type used_prefixes: List[str]
        :return: table_prefix to use
        :rtype: str
        """
        if related_field.is_multi:
            previous_model = related_field.through
        else:
            previous_model = related_field.owner
        table_prefix = cls.ormar_config.alias_manager.resolve_relation_alias(
            from_model=previous_model, relation_name=related_field.name
        )
        if not table_prefix or table_prefix in used_prefixes:
            manager = cls.ormar_config.alias_manager
            table_prefix = manager.resolve_relation_alias_after_complex(
                source_model=source_model,
                relation_str=current_relation_str,
                relation_field=related_field,
            )
        used_prefixes.append(table_prefix)
        return table_prefix

    @classmethod
    def _populate_nested_models_from_row(  # noqa: CFQ002
        cls,
        item: dict,
        row: ResultProxy,
        source_model: Type["Model"],
        related_models: Any,
        excludable: ExcludableItems,
        table_prefix: str,
        used_prefixes: List[str],
        current_relation_str: Optional[str] = None,
        proxy_source_model: Optional[Type["Model"]] = None,
    ) -> dict:
        """
        Traverses structure of related models and populates the nested models
        from the database row.
        Related models can be a list if only directly related models are to be
        populated, converted to dict if related models also have their own related
        models to be populated.

        Recurrently calls from_row method on nested instances and create nested
        instances. In the end those instances are added to the final model dictionary.

        :param proxy_source_model: source model from which querysetproxy is constructed
        :type proxy_source_model: Optional[Type["ModelRow"]]
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param source_model: source model from which relation started
        :type source_model: Type[Model]
        :param current_relation_str: joined related parts into one string
        :type current_relation_str: str
        :param item: dictionary of already populated nested models, otherwise empty dict
        :type item: Dict
        :param row: raw result row from the database
        :type row: ResultProxy
        :param related_models: list or dict of related models
        :type related_models: Union[Dict, List]
        :return: dictionary with keys corresponding to model fields names
        and values are database values
        :rtype: Dict
        """

        for related in related_models:
            field = cls.ormar_config.model_fields[related]
            field = cast("ForeignKeyField", field)
            model_cls = field.to
            model_excludable = excludable.get(
                model_cls=cast(Type["Model"], cls), alias=table_prefix
            )
            if model_excludable.is_excluded(related):
                continue

            relation_str, remainder = cls._process_remainder_and_relation_string(
                related_models=related_models,
                current_relation_str=current_relation_str,
                related=related,
            )
            child = model_cls.from_row(
                row,
                related_models=remainder,
                related_field=field,
                excludable=excludable,
                current_relation_str=relation_str,
                source_model=source_model,
                proxy_source_model=proxy_source_model,
                used_prefixes=used_prefixes,
            )
            item[model_cls.get_column_name_from_alias(related)] = child
            if (
                field.is_multi
                and child
                and not model_excludable.is_excluded(field.through.get_name())
            ):
                cls._populate_through_instance(
                    row=row,
                    item=item,
                    related=related,
                    excludable=excludable,
                    child=child,
                    proxy_source_model=proxy_source_model,
                )

        return item

    @staticmethod
    def _process_remainder_and_relation_string(
        related_models: Union[Dict, List],
        current_relation_str: Optional[str],
        related: str,
    ) -> Tuple[str, Optional[Union[Dict, List]]]:
        """
        Process remainder models and relation string

        :param related_models: list or dict of related models
        :type related_models: Union[Dict, List]
        :param current_relation_str: current relation string
        :type current_relation_str: Optional[str]
        :param related: name of the relation
        :type related: str
        """
        relation_str = (
            "__".join([current_relation_str, related])
            if current_relation_str
            else related
        )

        remainder = None
        if isinstance(related_models, dict) and related_models[related]:
            remainder = related_models[related]
        return relation_str, remainder

    @classmethod
    def _populate_through_instance(  # noqa: CFQ002
        cls,
        row: ResultProxy,
        item: Dict,
        related: str,
        excludable: ExcludableItems,
        child: "Model",
        proxy_source_model: Optional[Type["Model"]],
    ) -> None:
        """
        Populates the through model on reverse side of current query.
        Normally it's child class, unless the query is from queryset.

        :param row: row from db result
        :type row: ResultProxy
        :param item: parent item dict
        :type item: Dict
        :param related: current relation name
        :type related: str
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param child: child item of parent
        :type child: "Model"
        :param proxy_source_model: source model from which querysetproxy is constructed
        :type proxy_source_model: Type["Model"]
        """
        through_name = cls.ormar_config.model_fields[related].through.get_name()
        through_child = cls._create_through_instance(
            row=row, related=related, through_name=through_name, excludable=excludable
        )

        if child.__class__ != proxy_source_model:
            setattr(child, through_name, through_child)
        else:
            item[through_name] = through_child
        child.set_save_status(True)

    @classmethod
    def _create_through_instance(
        cls,
        row: ResultProxy,
        through_name: str,
        related: str,
        excludable: ExcludableItems,
    ) -> "ModelRow":
        """
        Initialize the through model from db row.
        Excluded all relation fields and other exclude/include set in excludable.

        :param row: loaded row from database
        :type row: sqlalchemy.engine.ResultProxy
        :param through_name: name of the through field
        :type through_name: str
        :param related: name of the relation
        :type related: str
        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :return: initialized through model without relation
        :rtype: "ModelRow"
        """
        model_cls = cls.ormar_config.model_fields[through_name].to
        table_prefix = cls.ormar_config.alias_manager.resolve_relation_alias(
            from_model=cls, relation_name=related
        )
        # remove relations on through field
        model_excludable = excludable.get(model_cls=model_cls, alias=table_prefix)
        model_excludable.set_values(
            value=model_cls.extract_related_names(), is_exclude=True
        )
        child_dict = model_cls.extract_prefixed_table_columns(
            item={}, row=row, excludable=excludable, table_prefix=table_prefix
        )
        child_dict["__excluded__"] = model_cls.get_names_to_exclude(
            excludable=excludable, alias=table_prefix
        )
        child = model_cls(**child_dict)  # type: ignore
        return child

    @classmethod
    def extract_prefixed_table_columns(
        cls,
        item: dict,
        row: ResultProxy,
        table_prefix: str,
        excludable: ExcludableItems,
    ) -> Dict:
        """
        Extracts own fields from raw sql result, using a given prefix.
        Prefix changes depending on the table's position in a join.

        If the table is a main table, there is no prefix.
        All joined tables have prefixes to allow duplicate column names,
        as well as duplicated joins to the same table from multiple different tables.

        Extracted fields populates the related dict later used to construct a Model.

        Used in Model.from_row and PrefetchQuery._populate_rows methods.

        :param excludable: structure of fields to include and exclude
        :type excludable: ExcludableItems
        :param item: dictionary of already populated nested models, otherwise empty dict
        :type item: Dict
        :param row: raw result row from the database
        :type row: sqlalchemy.engine.result.ResultProxy
        :param table_prefix: prefix of the table from AliasManager
        each pair of tables have own prefix (two of them depending on direction) -
        used in joins to allow multiple joins to the same table.
        :type table_prefix: str
        :return: dictionary with keys corresponding to model fields names
        and values are database values
        :rtype: Dict
        """
        selected_columns = cls.own_table_columns(
            model=cls, excludable=excludable, alias=table_prefix, use_alias=False
        )

        column_prefix = table_prefix + "_" if table_prefix else ""
        for column in cls.ormar_config.table.columns:
            alias = cls.get_column_name_from_alias(column.name)
            if alias not in item and alias in selected_columns:
                prefixed_name = f"{column_prefix}{column.name}"
                item[alias] = row[prefixed_name]

        return item

extract_prefixed_table_columns(item, row, table_prefix, excludable) classmethod

Extracts own fields from raw sql result, using a given prefix. Prefix changes depending on the table's position in a join.

If the table is a main table, there is no prefix. All joined tables have prefixes to allow duplicate column names, as well as duplicated joins to the same table from multiple different tables.

Extracted fields populates the related dict later used to construct a Model.

Used in Model.from_row and PrefetchQuery._populate_rows methods.

Parameters:

Name	Type	Description	Default
excludable	ExcludableItems	structure of fields to include and exclude	required
item	dict	dictionary of already populated nested models, otherwise empty dict	required
row	Row	raw result row from the database	required
table_prefix	str	prefix of the table from AliasManager each pair of tables have own prefix (two of them depending on direction) - used in joins to allow multiple joins to the same table.	required

Returns:

Type	Description
Dict	dictionary with keys corresponding to model fields names and values are database values

Source code in ormar\models\model_row.py

@classmethod
def extract_prefixed_table_columns(
    cls,
    item: dict,
    row: ResultProxy,
    table_prefix: str,
    excludable: ExcludableItems,
) -> Dict:
    """
    Extracts own fields from raw sql result, using a given prefix.
    Prefix changes depending on the table's position in a join.

    If the table is a main table, there is no prefix.
    All joined tables have prefixes to allow duplicate column names,
    as well as duplicated joins to the same table from multiple different tables.

    Extracted fields populates the related dict later used to construct a Model.

    Used in Model.from_row and PrefetchQuery._populate_rows methods.

    :param excludable: structure of fields to include and exclude
    :type excludable: ExcludableItems
    :param item: dictionary of already populated nested models, otherwise empty dict
    :type item: Dict
    :param row: raw result row from the database
    :type row: sqlalchemy.engine.result.ResultProxy
    :param table_prefix: prefix of the table from AliasManager
    each pair of tables have own prefix (two of them depending on direction) -
    used in joins to allow multiple joins to the same table.
    :type table_prefix: str
    :return: dictionary with keys corresponding to model fields names
    and values are database values
    :rtype: Dict
    """
    selected_columns = cls.own_table_columns(
        model=cls, excludable=excludable, alias=table_prefix, use_alias=False
    )

    column_prefix = table_prefix + "_" if table_prefix else ""
    for column in cls.ormar_config.table.columns:
        alias = cls.get_column_name_from_alias(column.name)
        if alias not in item and alias in selected_columns:
            prefixed_name = f"{column_prefix}{column.name}"
            item[alias] = row[prefixed_name]

    return item

from_row(row, source_model, select_related=None, related_models=None, related_field=None, excludable=None, current_relation_str='', proxy_source_model=None, used_prefixes=None) classmethod

Model method to convert raw sql row from database into ormar.Model instance. Traverses nested models if they were specified in select_related for query.

Called recurrently and returns model instance if it's present in the row. Note that it's processing one row at a time, so if there are duplicates of parent row that needs to be joined/combined (like parent row in sql join with 2+ child rows) instances populated in this method are later combined in the QuerySet. Other method working directly on raw database results is in prefetch_query, where rows are populated in a different way as they do not have nested models in result.

Parameters:

Name	Type	Description	Default
used_prefixes	Optional[List[str]]	list of already extracted prefixes	None
proxy_source_model	Optional[Type[Model]]	source model from which querysetproxy is constructed	None
excludable	Optional[ExcludableItems]	structure of fields to include and exclude	None
current_relation_str	str	name of the relation field	''
source_model	Type[Model]	model on which relation was defined	required
row	Row	raw result row from the database	required
select_related	Optional[List]	list of names of related models fetched from database	None
related_models	Any	list or dict of related models	None
related_field	Optional[ForeignKeyField]	field with relation declaration	None

Returns:

Type	Description
Optional[Model]	returns model if model is populated from database

Source code in ormar\models\model_row.py

@classmethod
def from_row(  # noqa: CFQ002
    cls,
    row: ResultProxy,
    source_model: Type["Model"],
    select_related: Optional[List] = None,
    related_models: Any = None,
    related_field: Optional["ForeignKeyField"] = None,
    excludable: Optional[ExcludableItems] = None,
    current_relation_str: str = "",
    proxy_source_model: Optional[Type["Model"]] = None,
    used_prefixes: Optional[List[str]] = None,
) -> Optional["Model"]:
    """
    Model method to convert raw sql row from database into ormar.Model instance.
    Traverses nested models if they were specified in select_related for query.

    Called recurrently and returns model instance if it's present in the row.
    Note that it's processing one row at a time, so if there are duplicates of
    parent row that needs to be joined/combined
    (like parent row in sql join with 2+ child rows)
    instances populated in this method are later combined in the QuerySet.
    Other method working directly on raw database results is in prefetch_query,
    where rows are populated in a different way as they do not have
    nested models in result.

    :param used_prefixes: list of already extracted prefixes
    :type used_prefixes: List[str]
    :param proxy_source_model: source model from which querysetproxy is constructed
    :type proxy_source_model: Optional[Type["ModelRow"]]
    :param excludable: structure of fields to include and exclude
    :type excludable: ExcludableItems
    :param current_relation_str: name of the relation field
    :type current_relation_str: str
    :param source_model: model on which relation was defined
    :type source_model: Type[Model]
    :param row: raw result row from the database
    :type row: ResultProxy
    :param select_related: list of names of related models fetched from database
    :type select_related: List
    :param related_models: list or dict of related models
    :type related_models: Union[List, Dict]
    :param related_field: field with relation declaration
    :type related_field: ForeignKeyField
    :return: returns model if model is populated from database
    :rtype: Optional[Model]
    """
    item: Dict[str, Any] = {}
    select_related = select_related or []
    related_models = related_models or []
    table_prefix = ""
    used_prefixes = used_prefixes if used_prefixes is not None else []
    excludable = excludable or ExcludableItems()

    if select_related:
        related_models = group_related_list(select_related)

    if related_field:
        table_prefix = cls._process_table_prefix(
            source_model=source_model,
            current_relation_str=current_relation_str,
            related_field=related_field,
            used_prefixes=used_prefixes,
        )

    item = cls._populate_nested_models_from_row(
        item=item,
        row=row,
        related_models=related_models,
        excludable=excludable,
        current_relation_str=current_relation_str,
        source_model=source_model,  # type: ignore
        proxy_source_model=proxy_source_model,  # type: ignore
        table_prefix=table_prefix,
        used_prefixes=used_prefixes,
    )
    item = cls.extract_prefixed_table_columns(
        item=item, row=row, table_prefix=table_prefix, excludable=excludable
    )

    instance: Optional["Model"] = None
    if item.get(cls.ormar_config.pkname, None) is not None:
        item["__excluded__"] = cls.get_names_to_exclude(
            excludable=excludable, alias=table_prefix
        )
        instance = cast("Model", cls(**item))
        instance.set_save_status(True)
    return instance

get_child(items, key=None) staticmethod

Used to get nested dictionaries keys if they exists otherwise returns passed items.

Parameters:

Name	Type	Description	Default
items	Union[Set, Dict, None]	bag of items to include or exclude	required
key	Optional[str]	name of the child to extract	None

Returns:

Type	Description
Union[Set, Dict, None]	child extracted from items if exists

Source code in ormar\models\mixins\excludable_mixin.py

@staticmethod
def get_child(
    items: Union[Set, Dict, None], key: Optional[str] = None
) -> Union[Set, Dict, None]:
    """
    Used to get nested dictionaries keys if they exists otherwise returns
    passed items.
    :param items: bag of items to include or exclude
    :type items:  Union[Set, Dict, None]
    :param key: name of the child to extract
    :type key: str
    :return: child extracted from items if exists
    :rtype: Union[Set, Dict, None]
    """
    if isinstance(items, dict):
        return items.get(key, {})
    return items

get_names_to_exclude(excludable, alias) classmethod

Returns a set of models field names that should be explicitly excluded during model initialization.

Those fields will be set to None to avoid ormar/pydantic setting default values on them. They should be returned as None in any case.

Used in parsing data from database rows that construct Models by initializing them with dicts constructed from those db rows.

Parameters:

Name	Type	Description	Default
alias	str	alias of current relation	required
excludable	ExcludableItems	structure of fields to include and exclude	required

Returns:

Type	Description
Set	set of field names that should be excluded

Source code in ormar\models\mixins\excludable_mixin.py

@classmethod
def get_names_to_exclude(cls, excludable: ExcludableItems, alias: str) -> Set:
    """
    Returns a set of models field names that should be explicitly excluded
    during model initialization.

    Those fields will be set to None to avoid ormar/pydantic setting default
    values on them. They should be returned as None in any case.

    Used in parsing data from database rows that construct Models by initializing
    them with dicts constructed from those db rows.

    :param alias: alias of current relation
    :type alias: str
    :param excludable: structure of fields to include and exclude
    :type excludable: ExcludableItems
    :return: set of field names that should be excluded
    :rtype: Set
    """
    model = cast(Type["Model"], cls)
    model_excludable = excludable.get(model_cls=model, alias=alias)
    fields_names = cls.extract_db_own_fields()
    if model_excludable.include:
        fields_to_keep = model_excludable.include.intersection(fields_names)
    else:
        fields_to_keep = fields_names

    fields_to_exclude = fields_names - fields_to_keep

    if model_excludable.exclude:
        fields_to_exclude = fields_to_exclude.union(
            model_excludable.exclude.intersection(fields_names)
        )
    fields_to_exclude = fields_to_exclude - {cls.ormar_config.pkname}

    return fields_to_exclude

own_table_columns(model, excludable, alias='', use_alias=False, add_pk_columns=True) classmethod

Returns list of aliases or field names for given model. Aliases/names switch is use_alias flag.

If provided only fields included in fields will be returned. If provided fields in exclude_fields will be excluded in return.

Primary key field is always added and cannot be excluded (will be added anyway).

Parameters:

Name	Type	Description	Default
add_pk_columns	bool	flag if add primary key - always yes if ormar parses data	True
alias	str	relation prefix	''
excludable	ExcludableItems	structure of fields to include and exclude	required
model	Union[Type[Model], Type[ModelRow]]	model on columns are selected	required
use_alias	bool	flag if aliases or field names should be used	False

Returns:

Type	Description
List[str]	list of column field names or aliases

Source code in ormar\models\mixins\excludable_mixin.py

@classmethod
def own_table_columns(
    cls,
    model: Union[Type["Model"], Type["ModelRow"]],
    excludable: ExcludableItems,
    alias: str = "",
    use_alias: bool = False,
    add_pk_columns: bool = True,
) -> List[str]:
    """
    Returns list of aliases or field names for given model.
    Aliases/names switch is use_alias flag.

    If provided only fields included in fields will be returned.
    If provided fields in exclude_fields will be excluded in return.

    Primary key field is always added and cannot be excluded (will be added anyway).

    :param add_pk_columns: flag if add primary key - always yes if ormar parses data
    :type add_pk_columns: bool
    :param alias: relation prefix
    :type alias: str
    :param excludable: structure of fields to include and exclude
    :type excludable: ExcludableItems
    :param model: model on columns are selected
    :type model: Type["Model"]
    :param use_alias: flag if aliases or field names should be used
    :type use_alias: bool
    :return: list of column field names or aliases
    :rtype: List[str]
    """
    model_excludable = excludable.get(model_cls=model, alias=alias)  # type: ignore
    columns = [
        model.get_column_name_from_alias(col.name) if not use_alias else col.name
        for col in model.ormar_config.table.columns
    ]
    field_names = [
        model.get_column_name_from_alias(col.name)
        for col in model.ormar_config.table.columns
    ]
    if model_excludable.include:
        columns = [
            col
            for col, name in zip(columns, field_names)
            if model_excludable.is_included(name)
        ]
    if model_excludable.exclude:
        columns = [
            col
            for col, name in zip(columns, field_names)
            if not model_excludable.is_excluded(name)
        ]

    # always has to return pk column for ormar to work
    if add_pk_columns:
        columns = cls._populate_pk_column(
            model=model, columns=columns, use_alias=use_alias
        )

    return columns

MergeModelMixin

Used to merge models instances returned by database, but already initialized to ormar Models.keys

Models can duplicate during joins when parent model has multiple child rows, in the end all parent (main) models should be unique.

Source code in ormar\models\mixins\merge_mixin.py

class MergeModelMixin:
    """
    Used to merge models instances returned by database,
    but already initialized to ormar Models.keys

    Models can duplicate during joins when parent model has multiple child rows,
    in the end all parent (main) models should be unique.
    """

    @classmethod
    def _recursive_add(cls, model_group: List["Model"]) -> List["Model"]:
        """
        Instead of accumulating the model additions one by one, this recursively adds
        the models. E.G.
        [1, 2, 3, 4].accumulate_add() would give [3, 3, 4], then [6, 4], then [10]
        where this method looks like
        [1, 2, 3, 4].recursive_add() gives [[3], [7]], [10]
        It's the same number of adds, but it gives better O(N) performance on sublists
        """
        if len(model_group) <= 1:
            return model_group

        added_values = []
        iterable_group = iter(model_group)
        for model in iterable_group:
            next_model = next(iterable_group, None)
            if next_model is not None:
                combined = cls.merge_two_instances(next_model, model)
            else:
                combined = model
            added_values.append(combined)

        return cls._recursive_add(added_values)

    @classmethod
    def merge_instances_list(cls, result_rows: List["Model"]) -> List["Model"]:
        """
        Merges a list of models into list of unique models.

        Models can duplicate during joins when parent model has multiple child rows,
        in the end all parent (main) models should be unique.

        :param result_rows: list of already initialized Models with child models
        populated, each instance is one row in db and some models can duplicate
        :type result_rows: List["Model"]
        :return: list of merged models where each main model is unique
        :rtype: List["Model"]
        """
        merged_rows: List["Model"] = []
        grouped_instances: Dict = {}

        for model in result_rows:
            grouped_instances.setdefault(model.pk, []).append(model)

        for group in grouped_instances.values():
            model = cls._recursive_add(group)[0]
            merged_rows.append(model)

        return merged_rows

    @classmethod
    def merge_two_instances(
        cls, one: "Model", other: "Model", relation_map: Optional[Dict] = None
    ) -> "Model":
        """
        Merges current (other) Model and previous one (one) and returns the current
        Model instance with data merged from previous one.

        If needed it's calling itself recurrently and merges also children models.

        :param relation_map: map of models relations to follow
        :type relation_map: Dict
        :param one: previous model instance
        :type one: Model
        :param other: current model instance
        :type other: Model
        :return: current Model instance with data merged from previous one.
        :rtype: Model
        """
        relation_map = (
            relation_map
            if relation_map is not None
            else translate_list_to_dict(one._iterate_related_models())
        )
        for field_name in relation_map:
            current_field = getattr(one, field_name)
            other_value = getattr(other, field_name, [])
            if isinstance(current_field, list):
                value_to_set = cls._merge_items_lists(
                    field_name=field_name,
                    current_field=current_field,
                    other_value=other_value,
                    relation_map=relation_map,
                )
                setattr(other, field_name, value_to_set)
            elif (
                isinstance(current_field, ormar.Model)
                and isinstance(other_value, ormar.Model)
                and current_field.pk == other_value.pk
            ):
                setattr(
                    other,
                    field_name,
                    cls.merge_two_instances(
                        current_field,
                        other_value,
                        relation_map=one._skip_ellipsis(  # type: ignore
                            relation_map, field_name, default_return=dict()
                        ),
                    ),
                )
        other.set_save_status(True)
        return other

    @classmethod
    def _merge_items_lists(
        cls,
        field_name: str,
        current_field: List,
        other_value: List,
        relation_map: Optional[Dict],
    ) -> List:
        """
        Takes two list of nested models and process them going deeper
        according with the map.

        If model from one's list is in other -> they are merged with relations
        to follow passed from map.

        If one's model is not in other it's simply appended to the list.

        :param field_name: name of the current relation field
        :type field_name: str
        :param current_field: list of nested models from one model
        :type current_field: List[Model]
        :param other_value: list of nested models from other model
        :type other_value: List[Model]
        :param relation_map: map of relations to follow
        :type relation_map: Dict
        :return: merged list of models
        :rtype: List[Model]
        """
        value_to_set = [x for x in other_value]
        for cur_field in current_field:
            if cur_field in other_value:
                old_value = next((x for x in other_value if x == cur_field), None)
                new_val = cls.merge_two_instances(
                    cur_field,
                    cast("Model", old_value),
                    relation_map=cur_field._skip_ellipsis(  # type: ignore
                        relation_map, field_name, default_return=dict()
                    ),
                )
                value_to_set = [x for x in value_to_set if x != cur_field] + [new_val]
            else:
                value_to_set.append(cur_field)
        return value_to_set

merge_instances_list(result_rows) classmethod

Merges a list of models into list of unique models.

Models can duplicate during joins when parent model has multiple child rows, in the end all parent (main) models should be unique.

Parameters:

Name	Type	Description	Default
result_rows	List[Model]	list of already initialized Models with child models populated, each instance is one row in db and some models can duplicate	required

Returns:

Type	Description
List["Model"]	list of merged models where each main model is unique

Source code in ormar\models\mixins\merge_mixin.py

@classmethod
def merge_instances_list(cls, result_rows: List["Model"]) -> List["Model"]:
    """
    Merges a list of models into list of unique models.

    Models can duplicate during joins when parent model has multiple child rows,
    in the end all parent (main) models should be unique.

    :param result_rows: list of already initialized Models with child models
    populated, each instance is one row in db and some models can duplicate
    :type result_rows: List["Model"]
    :return: list of merged models where each main model is unique
    :rtype: List["Model"]
    """
    merged_rows: List["Model"] = []
    grouped_instances: Dict = {}

    for model in result_rows:
        grouped_instances.setdefault(model.pk, []).append(model)

    for group in grouped_instances.values():
        model = cls._recursive_add(group)[0]
        merged_rows.append(model)

    return merged_rows

merge_two_instances(one, other, relation_map=None) classmethod

Merges current (other) Model and previous one (one) and returns the current Model instance with data merged from previous one.

If needed it's calling itself recurrently and merges also children models.

Parameters:

Name	Type	Description	Default
relation_map	Optional[Dict]	map of models relations to follow	None
one	Model	previous model instance	required
other	Model	current model instance	required

Returns:

Type	Description
Model	current Model instance with data merged from previous one.

Source code in ormar\models\mixins\merge_mixin.py

@classmethod
def merge_two_instances(
    cls, one: "Model", other: "Model", relation_map: Optional[Dict] = None
) -> "Model":
    """
    Merges current (other) Model and previous one (one) and returns the current
    Model instance with data merged from previous one.

    If needed it's calling itself recurrently and merges also children models.

    :param relation_map: map of models relations to follow
    :type relation_map: Dict
    :param one: previous model instance
    :type one: Model
    :param other: current model instance
    :type other: Model
    :return: current Model instance with data merged from previous one.
    :rtype: Model
    """
    relation_map = (
        relation_map
        if relation_map is not None
        else translate_list_to_dict(one._iterate_related_models())
    )
    for field_name in relation_map:
        current_field = getattr(one, field_name)
        other_value = getattr(other, field_name, [])
        if isinstance(current_field, list):
            value_to_set = cls._merge_items_lists(
                field_name=field_name,
                current_field=current_field,
                other_value=other_value,
                relation_map=relation_map,
            )
            setattr(other, field_name, value_to_set)
        elif (
            isinstance(current_field, ormar.Model)
            and isinstance(other_value, ormar.Model)
            and current_field.pk == other_value.pk
        ):
            setattr(
                other,
                field_name,
                cls.merge_two_instances(
                    current_field,
                    other_value,
                    relation_map=one._skip_ellipsis(  # type: ignore
                        relation_map, field_name, default_return=dict()
                    ),
                ),
            )
    other.set_save_status(True)
    return other

PydanticMixin

Bases: RelationMixin

Source code in ormar\models\mixins\pydantic_mixin.py

class PydanticMixin(RelationMixin):
    __cache__: Dict[str, Type[pydantic.BaseModel]] = {}

    if TYPE_CHECKING:  # pragma: no cover
        __pydantic_decorators__: DecoratorInfos
        model_fields: Dict[str, FieldInfo]
        _skip_ellipsis: Callable
        _get_not_excluded_fields: Callable

    @classmethod
    def get_pydantic(
        cls,
        *,
        include: Union[Set, Dict, None] = None,
        exclude: Union[Set, Dict, None] = None,
    ) -> Type[pydantic.BaseModel]:
        """
        Returns a pydantic model out of ormar model.

        Converts also nested ormar models into pydantic models.

        Can be used to fully exclude certain fields in fastapi response and requests.

        :param include: fields of own and nested models to include
        :type include: Union[Set, Dict, None]
        :param exclude: fields of own and nested models to exclude
        :type exclude: Union[Set, Dict, None]
        """
        relation_map = translate_list_to_dict(cls._iterate_related_models())

        return cls._convert_ormar_to_pydantic(
            include=include, exclude=exclude, relation_map=relation_map
        )

    @classmethod
    def _convert_ormar_to_pydantic(
        cls,
        relation_map: Dict[str, Any],
        include: Union[Set, Dict, None] = None,
        exclude: Union[Set, Dict, None] = None,
    ) -> Type[pydantic.BaseModel]:
        if include and isinstance(include, Set):
            include = translate_list_to_dict(include)
        if exclude and isinstance(exclude, Set):
            exclude = translate_list_to_dict(exclude)
        fields_dict: Dict[str, Any] = dict()
        defaults: Dict[str, Any] = dict()
        fields_to_process = cls._get_not_excluded_fields(
            fields={*cls.ormar_config.model_fields.keys()},
            include=include,
            exclude=exclude,
        )
        fields_to_process.sort(
            key=lambda x: list(cls.ormar_config.model_fields.keys()).index(x)
        )

        cache_key = f"{cls.__name__}_{str(include)}_{str(exclude)}"
        if cache_key in cls.__cache__:
            return cls.__cache__[cache_key]

        for name in fields_to_process:
            field = cls._determine_pydantic_field_type(
                name=name,
                defaults=defaults,
                include=include,
                exclude=exclude,
                relation_map=relation_map,
            )
            if field is not None:
                fields_dict[name] = field
        model = type(
            f"{cls.__name__}_{''.join(choices(string.ascii_uppercase, k=3))}",
            (pydantic.BaseModel,),
            {"__annotations__": fields_dict, **defaults},
        )
        model = cast(Type[pydantic.BaseModel], model)
        cls._copy_field_validators(model=model)
        cls.__cache__[cache_key] = model
        return model

    @classmethod
    def _determine_pydantic_field_type(
        cls,
        name: str,
        defaults: Dict,
        include: Union[Set, Dict, None],
        exclude: Union[Set, Dict, None],
        relation_map: Dict[str, Any],
    ) -> Any:
        field = cls.ormar_config.model_fields[name]
        target: Any = None
        if field.is_relation and name in relation_map:
            target, default = cls._determined_included_relation_field_type(
                name=name,
                field=field,
                include=include,
                exclude=exclude,
                defaults=defaults,
                relation_map=relation_map,
            )
        elif not field.is_relation:
            defaults[name] = cls.model_fields[name].default
            target = field.__type__
        if target is not None and field.nullable:
            target = Optional[target]
        return target

    @classmethod
    def _determined_included_relation_field_type(
        cls,
        name: str,
        field: Union[BaseField, ForeignKeyField, ManyToManyField],
        include: Union[Set, Dict, None],
        exclude: Union[Set, Dict, None],
        defaults: Dict,
        relation_map: Dict[str, Any],
    ) -> Tuple[Type[BaseModel], Dict]:
        target = field.to._convert_ormar_to_pydantic(
            include=cls._skip_ellipsis(include, name),
            exclude=cls._skip_ellipsis(exclude, name),
            relation_map=cls._skip_ellipsis(relation_map, name, default_return=dict()),
        )
        if field.is_multi or field.virtual:
            target = List[target]  # type: ignore
        if field.nullable:
            defaults[name] = None
        return target, defaults

    @classmethod
    def _copy_field_validators(cls, model: Type[pydantic.BaseModel]) -> None:
        """
        Copy field validators from ormar model to generated pydantic model.
        """
        filed_names = list(model.model_fields.keys())
        cls.copy_selected_validators_type(
            model=model, fields=filed_names, validator_type="field_validators"
        )
        cls.copy_selected_validators_type(
            model=model, fields=filed_names, validator_type="validators"
        )

        class_validators = cls.__pydantic_decorators__.root_validators
        model.__pydantic_decorators__.root_validators.update(
            copy.deepcopy(class_validators)
        )
        model_validators = cls.__pydantic_decorators__.model_validators
        model.__pydantic_decorators__.model_validators.update(
            copy.deepcopy(model_validators)
        )
        model.model_rebuild(force=True)

    @classmethod
    def copy_selected_validators_type(
        cls, model: Type[pydantic.BaseModel], fields: List[str], validator_type: str
    ) -> None:
        """
        Copy field validators from ormar model to generated pydantic model.
        """
        validators = getattr(cls.__pydantic_decorators__, validator_type)
        for name, decorator in validators.items():
            if any(field_name in decorator.info.fields for field_name in fields):
                copied_decorator = copy.deepcopy(decorator)
                copied_decorator.info.fields = [
                    field_name
                    for field_name in decorator.info.fields
                    if field_name in fields
                ]
                getattr(model.__pydantic_decorators__, validator_type)[
                    name
                ] = copied_decorator

copy_selected_validators_type(model, fields, validator_type) classmethod

Copy field validators from ormar model to generated pydantic model.

Source code in ormar\models\mixins\pydantic_mixin.py

@classmethod
def copy_selected_validators_type(
    cls, model: Type[pydantic.BaseModel], fields: List[str], validator_type: str
) -> None:
    """
    Copy field validators from ormar model to generated pydantic model.
    """
    validators = getattr(cls.__pydantic_decorators__, validator_type)
    for name, decorator in validators.items():
        if any(field_name in decorator.info.fields for field_name in fields):
            copied_decorator = copy.deepcopy(decorator)
            copied_decorator.info.fields = [
                field_name
                for field_name in decorator.info.fields
                if field_name in fields
            ]
            getattr(model.__pydantic_decorators__, validator_type)[
                name
            ] = copied_decorator

get_pydantic(*, include=None, exclude=None) classmethod

Returns a pydantic model out of ormar model.

Converts also nested ormar models into pydantic models.

Can be used to fully exclude certain fields in fastapi response and requests.

Parameters:

Name	Type	Description	Default
include	Union[Set, Dict, None]	fields of own and nested models to include	None
exclude	Union[Set, Dict, None]	fields of own and nested models to exclude	None

Source code in ormar\models\mixins\pydantic_mixin.py

@classmethod
def get_pydantic(
    cls,
    *,
    include: Union[Set, Dict, None] = None,
    exclude: Union[Set, Dict, None] = None,
) -> Type[pydantic.BaseModel]:
    """
    Returns a pydantic model out of ormar model.

    Converts also nested ormar models into pydantic models.

    Can be used to fully exclude certain fields in fastapi response and requests.

    :param include: fields of own and nested models to include
    :type include: Union[Set, Dict, None]
    :param exclude: fields of own and nested models to exclude
    :type exclude: Union[Set, Dict, None]
    """
    relation_map = translate_list_to_dict(cls._iterate_related_models())

    return cls._convert_ormar_to_pydantic(
        include=include, exclude=exclude, relation_map=relation_map
    )

SavePrepareMixin

Bases: RelationMixin, AliasMixin

Used to prepare models to be saved in database

Source code in ormar\models\mixins\save_mixin.py

class SavePrepareMixin(RelationMixin, AliasMixin):
    """
    Used to prepare models to be saved in database
    """

    if TYPE_CHECKING:  # pragma: nocover
        _skip_ellipsis: Callable
        _json_fields: Set[str]
        _bytes_fields: Set[str]
        __pydantic_core_schema__: CoreSchema
        __ormar_fields_validators__: Optional[Dict[str, SchemaValidator]]

    @classmethod
    def prepare_model_to_save(cls, new_kwargs: dict) -> dict:
        """
        Combines all preparation methods before saving.
        Removes primary key for if it's nullable or autoincrement pk field,
        and it's set to None.
        Substitute related models with their primary key values as fk column.
        Populates the default values for field with default set and no value.
        Translate columns into aliases (db names).

        :param new_kwargs: dictionary of model that is about to be saved
        :type new_kwargs: Dict[str, str]
        :return: dictionary of model that is about to be saved
        :rtype: Dict[str, str]
        """
        new_kwargs = cls._remove_pk_from_kwargs(new_kwargs)
        new_kwargs = cls._remove_not_ormar_fields(new_kwargs)
        new_kwargs = cls.substitute_models_with_pks(new_kwargs)
        new_kwargs = cls.populate_default_values(new_kwargs)
        new_kwargs = cls.reconvert_str_to_bytes(new_kwargs)
        new_kwargs = cls.translate_columns_to_aliases(new_kwargs)
        return new_kwargs

    @classmethod
    def prepare_model_to_update(cls, new_kwargs: dict) -> dict:
        """
        Combines all preparation methods before updating.
        :param new_kwargs: dictionary of model that is about to be saved
        :type new_kwargs: Dict[str, str]
        :return: dictionary of model that is about to be updated
        :rtype: Dict[str, str]
        """
        new_kwargs = cls.parse_non_db_fields(new_kwargs)
        new_kwargs = cls.substitute_models_with_pks(new_kwargs)
        new_kwargs = cls.reconvert_str_to_bytes(new_kwargs)
        new_kwargs = cls.dump_all_json_fields_to_str(new_kwargs)
        new_kwargs = cls.translate_columns_to_aliases(new_kwargs)
        new_kwargs = cls.translate_enum_columns(new_kwargs)
        return new_kwargs

    @classmethod
    def translate_enum_columns(cls, new_kwargs: dict) -> dict:
        for key, value in new_kwargs.items():
            if isinstance(value, Enum):
                new_kwargs[key] = value.name
        return new_kwargs

    @classmethod
    def _remove_not_ormar_fields(cls, new_kwargs: dict) -> dict:
        """
        Removes primary key for if it's nullable or autoincrement pk field,
        and it's set to None.

        :param new_kwargs: dictionary of model that is about to be saved
        :type new_kwargs: Dict[str, str]
        :return: dictionary of model that is about to be saved
        :rtype: Dict[str, str]
        """
        ormar_fields = {k for k, v in cls.ormar_config.model_fields.items()}
        new_kwargs = {k: v for k, v in new_kwargs.items() if k in ormar_fields}
        return new_kwargs

    @classmethod
    def _remove_pk_from_kwargs(cls, new_kwargs: dict) -> dict:
        """
        Removes primary key for if it's nullable or autoincrement pk field,
        and it's set to None.

        :param new_kwargs: dictionary of model that is about to be saved
        :type new_kwargs: Dict[str, str]
        :return: dictionary of model that is about to be saved
        :rtype: Dict[str, str]
        """
        pkname = cls.ormar_config.pkname
        pk = cls.ormar_config.model_fields[pkname]
        if new_kwargs.get(pkname, ormar.Undefined) is None and (
            pk.nullable or pk.autoincrement
        ):
            del new_kwargs[pkname]
        return new_kwargs

    @classmethod
    def parse_non_db_fields(cls, model_dict: Dict) -> Dict:
        """
        Receives dictionary of model that is about to be saved and changes uuid fields
        to strings in bulk_update.

        :param model_dict: dictionary of model that is about to be saved
        :type model_dict: Dict
        :return: dictionary of model that is about to be saved
        :rtype: Dict
        """
        for name, field in cls.ormar_config.model_fields.items():
            if field.__type__ == uuid.UUID and name in model_dict:
                parsers = {"string": lambda x: str(x), "hex": lambda x: "%.32x" % x.int}
                uuid_format = field.column_type.uuid_format
                parser: Callable[..., Any] = parsers.get(uuid_format, lambda x: x)
                model_dict[name] = parser(model_dict[name])
        return model_dict

    @classmethod
    def substitute_models_with_pks(cls, model_dict: Dict) -> Dict:  # noqa  CCR001
        """
        Receives dictionary of model that is about to be saved and changes all related
        models that are stored as foreign keys to their fk value.

        :param model_dict: dictionary of model that is about to be saved
        :type model_dict: Dict
        :return: dictionary of model that is about to be saved
        :rtype: Dict
        """
        for field in cls.extract_related_names():
            field_value = model_dict.get(field, None)
            if field_value is not None:
                target_field = cls.ormar_config.model_fields[field]
                target_pkname = target_field.to.ormar_config.pkname
                if isinstance(field_value, ormar.Model):  # pragma: no cover
                    pk_value = getattr(field_value, target_pkname)
                    if not pk_value:
                        raise ModelPersistenceError(
                            f"You cannot save {field_value.get_name()} "
                            f"model without pk set!"
                        )
                    model_dict[field] = pk_value
                elif isinstance(field_value, (list, dict)) and field_value:
                    if isinstance(field_value, list):
                        model_dict[field] = [
                            target.get(target_pkname) for target in field_value
                        ]
                    else:
                        model_dict[field] = field_value.get(target_pkname)
                else:
                    model_dict.pop(field, None)
        return model_dict

    @classmethod
    def reconvert_str_to_bytes(cls, model_dict: Dict) -> Dict:
        """
        Receives dictionary of model that is about to be saved and changes
        all bytes fields that are represented as strings back into bytes.

        :param model_dict: dictionary of model that is about to be saved
        :type model_dict: Dict
        :return: dictionary of model that is about to be saved
        :rtype: Dict
        """
        bytes_base64_fields = {
            name
            for name, field in cls.ormar_config.model_fields.items()
            if field.represent_as_base64_str
        }
        for key, value in model_dict.items():
            if key in cls._bytes_fields and isinstance(value, str):
                model_dict[key] = (
                    value.encode("utf-8")
                    if key not in bytes_base64_fields
                    else base64.b64decode(value)
                )
        return model_dict

    @classmethod
    def dump_all_json_fields_to_str(cls, model_dict: Dict) -> Dict:
        """
        Receives dictionary of model that is about to be saved and changes
        all json fields into strings

        :param model_dict: dictionary of model that is about to be saved
        :type model_dict: Dict
        :return: dictionary of model that is about to be saved
        :rtype: Dict
        """
        for key, value in model_dict.items():
            if key in cls._json_fields:
                model_dict[key] = encode_json(value)
        return model_dict

    @classmethod
    def populate_default_values(cls, new_kwargs: Dict) -> Dict:
        """
        Receives dictionary of model that is about to be saved and populates the default
        value on the fields that have the default value set, but no actual value was
        passed by the user.

        :param new_kwargs: dictionary of model that is about to be saved
        :type new_kwargs: Dict
        :return: dictionary of model that is about to be saved
        :rtype: Dict
        """
        for field_name, field in cls.ormar_config.model_fields.items():
            if field_name not in new_kwargs and field.has_default(use_server=False):
                new_kwargs[field_name] = field.get_default()
            # clear fields with server_default set as None
            if (
                field.server_default is not None
                and new_kwargs.get(field_name, None) is None
            ):
                new_kwargs.pop(field_name, None)
        return new_kwargs

    @classmethod
    def validate_enums(cls, new_kwargs: Dict) -> Dict:
        """
        Receives dictionary of model that is about to be saved and validates the
        fields with choices set to see if the value is allowed.

        :param new_kwargs: dictionary of model that is about to be saved
        :type new_kwargs: Dict
        :return: dictionary of model that is about to be saved
        :rtype: Dict
        """
        validators = cls._build_individual_schema_validator()
        for key, value in new_kwargs.items():
            if key in validators:
                validators[key].validate_python(value)
        return new_kwargs

    @classmethod
    def _build_individual_schema_validator(cls) -> Any:
        if cls.__ormar_fields_validators__ is not None:
            return cls.__ormar_fields_validators__
        field_validators = {}
        for key, field in cls._extract_pydantic_fields().items():
            if cls.__pydantic_core_schema__["type"] == "definitions":
                schema = {
                    "type": "definitions",
                    "schema": field["schema"],
                    "definitions": cls.__pydantic_core_schema__["definitions"],
                }
            else:
                schema = field["schema"]
            field_validators[key] = create_schema_validator(
                schema, cls, cls.__module__, cls.__qualname__, "BaseModel"
            )
        cls.__ormar_fields_validators__ = field_validators
        return cls.__ormar_fields_validators__

    @classmethod
    def _extract_pydantic_fields(cls) -> Any:
        if cls.__pydantic_core_schema__["type"] == "model":
            return cls.__pydantic_core_schema__["schema"]["fields"]
        elif cls.__pydantic_core_schema__["type"] == "definitions":
            main_schema = cls.__pydantic_core_schema__["schema"]
            if "schema_ref" in main_schema:
                reference_id = main_schema["schema_ref"]
                return next(
                    ref
                    for ref in cls.__pydantic_core_schema__["definitions"]
                    if ref["ref"] == reference_id
                )["schema"]["fields"]
            return main_schema["schema"]["fields"]

    @staticmethod
    async def _upsert_model(
        instance: "Model",
        save_all: bool,
        previous_model: Optional["Model"],
        relation_field: Optional["ForeignKeyField"],
        update_count: int,
    ) -> int:
        """
        Method updates given instance if:

        * instance is not saved or
        * instance have no pk or
        * save_all=True flag is set

        and instance is not __pk_only__.

        If relation leading to instance is a ManyToMany also the through model is saved

        :param instance: current model to upsert
        :type instance: Model
        :param save_all: flag if all models should be saved or only not saved ones
        :type save_all: bool
        :param relation_field: field with relation
        :type relation_field: Optional[ForeignKeyField]
        :param previous_model: previous model from which method came
        :type previous_model: Model
        :param update_count: no of updated models
        :type update_count: int
        :return: no of updated models
        :rtype: int
        """
        if (
            save_all or not instance.pk or not instance.saved
        ) and not instance.__pk_only__:
            await instance.upsert(__force_save__=True)
            if relation_field and relation_field.is_multi:
                await instance._upsert_through_model(
                    instance=instance,
                    relation_field=relation_field,
                    previous_model=cast("Model", previous_model),
                )
            update_count += 1
        return update_count

    @staticmethod
    async def _upsert_through_model(
        instance: "Model", previous_model: "Model", relation_field: "ForeignKeyField"
    ) -> None:
        """
        Upsert through model for m2m relation.

        :param instance: current model to upsert
        :type instance: Model
        :param relation_field: field with relation
        :type relation_field: Optional[ForeignKeyField]
        :param previous_model: previous model from which method came
        :type previous_model: Model
        """
        through_name = previous_model.ormar_config.model_fields[
            relation_field.name
        ].through.get_name()
        through = getattr(instance, through_name)
        if through:
            through_dict = through.model_dump(exclude=through.extract_related_names())
        else:
            through_dict = {}
        await getattr(
            previous_model, relation_field.name
        ).queryset_proxy.upsert_through_instance(instance, **through_dict)

    async def _update_relation_list(
        self,
        fields_list: Collection["ForeignKeyField"],
        follow: bool,
        save_all: bool,
        relation_map: Dict,
        update_count: int,
    ) -> int:
        """
        Internal method used in save_related to follow deeper from
        related models and update numbers of updated related instances.

        :type save_all: flag if all models should be saved
        :type save_all: bool
        :param fields_list: list of ormar fields to follow and save
        :type fields_list: Collection["ForeignKeyField"]
        :param relation_map: map of relations to follow
        :type relation_map: Dict
        :param follow: flag to trigger deep save -
        by default only directly related models are saved
        with follow=True also related models of related models are saved
        :type follow: bool
        :param update_count: internal parameter for recursive calls -
        number of updated instances
        :type update_count: int
        :return: tuple of update count and visited
        :rtype: int
        """
        for field in fields_list:
            values = self._get_field_values(name=field.name)
            for value in values:
                if follow:
                    update_count = await value.save_related(
                        follow=follow,
                        save_all=save_all,
                        relation_map=self._skip_ellipsis(  # type: ignore
                            relation_map, field.name, default_return={}
                        ),
                        update_count=update_count,
                        previous_model=self,
                        relation_field=field,
                    )
                else:
                    update_count = await value._upsert_model(
                        instance=value,
                        save_all=save_all,
                        previous_model=self,
                        relation_field=field,
                        update_count=update_count,
                    )
        return update_count

    def _get_field_values(self, name: str) -> List:
        """
        Extract field values and ensures it is a list.

        :param name: name of the field
        :type name: str
        :return: list of values
        :rtype: List
        """
        values = getattr(self, name) or []
        if not isinstance(values, list):
            values = [values]
        return values

dump_all_json_fields_to_str(model_dict) classmethod

Receives dictionary of model that is about to be saved and changes all json fields into strings

Parameters:

Name	Type	Description	Default
model_dict	Dict	dictionary of model that is about to be saved	required

Returns:

Type	Description
Dict	dictionary of model that is about to be saved

Source code in ormar\models\mixins\save_mixin.py

@classmethod
def dump_all_json_fields_to_str(cls, model_dict: Dict) -> Dict:
    """
    Receives dictionary of model that is about to be saved and changes
    all json fields into strings

    :param model_dict: dictionary of model that is about to be saved
    :type model_dict: Dict
    :return: dictionary of model that is about to be saved
    :rtype: Dict
    """
    for key, value in model_dict.items():
        if key in cls._json_fields:
            model_dict[key] = encode_json(value)
    return model_dict

parse_non_db_fields(model_dict) classmethod

Receives dictionary of model that is about to be saved and changes uuid fields to strings in bulk_update.

Parameters:

Name	Type	Description	Default
model_dict	Dict	dictionary of model that is about to be saved	required

Returns:

Type	Description
Dict	dictionary of model that is about to be saved

Source code in ormar\models\mixins\save_mixin.py

@classmethod
def parse_non_db_fields(cls, model_dict: Dict) -> Dict:
    """
    Receives dictionary of model that is about to be saved and changes uuid fields
    to strings in bulk_update.

    :param model_dict: dictionary of model that is about to be saved
    :type model_dict: Dict
    :return: dictionary of model that is about to be saved
    :rtype: Dict
    """
    for name, field in cls.ormar_config.model_fields.items():
        if field.__type__ == uuid.UUID and name in model_dict:
            parsers = {"string": lambda x: str(x), "hex": lambda x: "%.32x" % x.int}
            uuid_format = field.column_type.uuid_format
            parser: Callable[..., Any] = parsers.get(uuid_format, lambda x: x)
            model_dict[name] = parser(model_dict[name])
    return model_dict

populate_default_values(new_kwargs) classmethod

Receives dictionary of model that is about to be saved and populates the default value on the fields that have the default value set, but no actual value was passed by the user.

Parameters:

Name	Type	Description	Default
new_kwargs	Dict	dictionary of model that is about to be saved	required

Returns:

Type	Description
Dict	dictionary of model that is about to be saved

Source code in ormar\models\mixins\save_mixin.py

@classmethod
def populate_default_values(cls, new_kwargs: Dict) -> Dict:
    """
    Receives dictionary of model that is about to be saved and populates the default
    value on the fields that have the default value set, but no actual value was
    passed by the user.

    :param new_kwargs: dictionary of model that is about to be saved
    :type new_kwargs: Dict
    :return: dictionary of model that is about to be saved
    :rtype: Dict
    """
    for field_name, field in cls.ormar_config.model_fields.items():
        if field_name not in new_kwargs and field.has_default(use_server=False):
            new_kwargs[field_name] = field.get_default()
        # clear fields with server_default set as None
        if (
            field.server_default is not None
            and new_kwargs.get(field_name, None) is None
        ):
            new_kwargs.pop(field_name, None)
    return new_kwargs

prepare_model_to_save(new_kwargs) classmethod

Combines all preparation methods before saving. Removes primary key for if it's nullable or autoincrement pk field, and it's set to None. Substitute related models with their primary key values as fk column. Populates the default values for field with default set and no value. Translate columns into aliases (db names).

Parameters:

Name	Type	Description	Default
new_kwargs	dict	dictionary of model that is about to be saved	required

Returns:

Type	Description
Dict[str, str]	dictionary of model that is about to be saved

Source code in ormar\models\mixins\save_mixin.py

@classmethod
def prepare_model_to_save(cls, new_kwargs: dict) -> dict:
    """
    Combines all preparation methods before saving.
    Removes primary key for if it's nullable or autoincrement pk field,
    and it's set to None.
    Substitute related models with their primary key values as fk column.
    Populates the default values for field with default set and no value.
    Translate columns into aliases (db names).

    :param new_kwargs: dictionary of model that is about to be saved
    :type new_kwargs: Dict[str, str]
    :return: dictionary of model that is about to be saved
    :rtype: Dict[str, str]
    """
    new_kwargs = cls._remove_pk_from_kwargs(new_kwargs)
    new_kwargs = cls._remove_not_ormar_fields(new_kwargs)
    new_kwargs = cls.substitute_models_with_pks(new_kwargs)
    new_kwargs = cls.populate_default_values(new_kwargs)
    new_kwargs = cls.reconvert_str_to_bytes(new_kwargs)
    new_kwargs = cls.translate_columns_to_aliases(new_kwargs)
    return new_kwargs

prepare_model_to_update(new_kwargs) classmethod

Combines all preparation methods before updating.

Parameters:

Name	Type	Description	Default
new_kwargs	dict	dictionary of model that is about to be saved	required

Returns:

Type	Description
Dict[str, str]	dictionary of model that is about to be updated

Source code in ormar\models\mixins\save_mixin.py

@classmethod
def prepare_model_to_update(cls, new_kwargs: dict) -> dict:
    """
    Combines all preparation methods before updating.
    :param new_kwargs: dictionary of model that is about to be saved
    :type new_kwargs: Dict[str, str]
    :return: dictionary of model that is about to be updated
    :rtype: Dict[str, str]
    """
    new_kwargs = cls.parse_non_db_fields(new_kwargs)
    new_kwargs = cls.substitute_models_with_pks(new_kwargs)
    new_kwargs = cls.reconvert_str_to_bytes(new_kwargs)
    new_kwargs = cls.dump_all_json_fields_to_str(new_kwargs)
    new_kwargs = cls.translate_columns_to_aliases(new_kwargs)
    new_kwargs = cls.translate_enum_columns(new_kwargs)
    return new_kwargs

reconvert_str_to_bytes(model_dict) classmethod

Receives dictionary of model that is about to be saved and changes all bytes fields that are represented as strings back into bytes.

Parameters:

Name	Type	Description	Default
model_dict	Dict	dictionary of model that is about to be saved	required

Returns:

Type	Description
Dict	dictionary of model that is about to be saved

Source code in ormar\models\mixins\save_mixin.py

@classmethod
def reconvert_str_to_bytes(cls, model_dict: Dict) -> Dict:
    """
    Receives dictionary of model that is about to be saved and changes
    all bytes fields that are represented as strings back into bytes.

    :param model_dict: dictionary of model that is about to be saved
    :type model_dict: Dict
    :return: dictionary of model that is about to be saved
    :rtype: Dict
    """
    bytes_base64_fields = {
        name
        for name, field in cls.ormar_config.model_fields.items()
        if field.represent_as_base64_str
    }
    for key, value in model_dict.items():
        if key in cls._bytes_fields and isinstance(value, str):
            model_dict[key] = (
                value.encode("utf-8")
                if key not in bytes_base64_fields
                else base64.b64decode(value)
            )
    return model_dict

substitute_models_with_pks(model_dict) classmethod

Receives dictionary of model that is about to be saved and changes all related models that are stored as foreign keys to their fk value.

Parameters:

Name	Type	Description	Default
model_dict	Dict	dictionary of model that is about to be saved	required

Returns:

Type	Description
Dict	dictionary of model that is about to be saved

Source code in ormar\models\mixins\save_mixin.py

@classmethod
def substitute_models_with_pks(cls, model_dict: Dict) -> Dict:  # noqa  CCR001
    """
    Receives dictionary of model that is about to be saved and changes all related
    models that are stored as foreign keys to their fk value.

    :param model_dict: dictionary of model that is about to be saved
    :type model_dict: Dict
    :return: dictionary of model that is about to be saved
    :rtype: Dict
    """
    for field in cls.extract_related_names():
        field_value = model_dict.get(field, None)
        if field_value is not None:
            target_field = cls.ormar_config.model_fields[field]
            target_pkname = target_field.to.ormar_config.pkname
            if isinstance(field_value, ormar.Model):  # pragma: no cover
                pk_value = getattr(field_value, target_pkname)
                if not pk_value:
                    raise ModelPersistenceError(
                        f"You cannot save {field_value.get_name()} "
                        f"model without pk set!"
                    )
                model_dict[field] = pk_value
            elif isinstance(field_value, (list, dict)) and field_value:
                if isinstance(field_value, list):
                    model_dict[field] = [
                        target.get(target_pkname) for target in field_value
                    ]
                else:
                    model_dict[field] = field_value.get(target_pkname)
            else:
                model_dict.pop(field, None)
    return model_dict

validate_enums(new_kwargs) classmethod

Receives dictionary of model that is about to be saved and validates the fields with choices set to see if the value is allowed.

Parameters:

Name	Type	Description	Default
new_kwargs	Dict	dictionary of model that is about to be saved	required

Returns:

Type	Description
Dict	dictionary of model that is about to be saved

Source code in ormar\models\mixins\save_mixin.py

@classmethod
def validate_enums(cls, new_kwargs: Dict) -> Dict:
    """
    Receives dictionary of model that is about to be saved and validates the
    fields with choices set to see if the value is allowed.

    :param new_kwargs: dictionary of model that is about to be saved
    :type new_kwargs: Dict
    :return: dictionary of model that is about to be saved
    :rtype: Dict
    """
    validators = cls._build_individual_schema_validator()
    for key, value in new_kwargs.items():
        if key in validators:
            validators[key].validate_python(value)
    return new_kwargs