model

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)