ormar

The ormar package is an async mini ORM for Python, with support for Postgres, MySQL, and SQLite.

The main benefit of using ormar are:

• getting an async ORM that can be used with async frameworks (fastapi, starlette etc.)
• getting just one model to maintain - you don't have to maintain pydantic and other orm model (sqlalchemy, peewee, gino etc.)

The goal was to create a simple ORM that can be used directly (as request and response models) with fastapi that bases it's data validation on pydantic.

Ormar - apart form obvious ORM in name - get it's name from ormar in swedish which means snakes, and ormar(e) in italian which means cabinet.

And what's a better name for python ORM than snakes cabinet :)

BaseField

Bases: FieldInfo

BaseField serves as a parent class for all basic Fields in ormar. It keeps all common parameters available for all fields as well as set of useful functions.

All values are kept as class variables, ormar Fields are never instantiated. Subclasses pydantic.FieldInfo to keep the fields related to pydantic field types like ConstrainedStr

Source code in ormar/fields/base.py

class BaseField(FieldInfo):
    """
    BaseField serves as a parent class for all basic Fields in ormar.
    It keeps all common parameters available for all fields as well as
    set of useful functions.

    All values are kept as class variables, ormar Fields are never instantiated.
    Subclasses pydantic.FieldInfo to keep the fields related
    to pydantic field types like ConstrainedStr
    """

    def __init__(self, **kwargs: Any) -> None:
        self.__type__: type = kwargs.pop("__type__", None)
        self.__pydantic_type__: type = kwargs.pop("__pydantic_type__", None)
        self.__sample__: type = kwargs.pop("__sample__", None)
        self.related_name = kwargs.pop("related_name", None)

        self.column_type: sqlalchemy.Column = kwargs.pop("column_type", None)
        self.constraints: List = kwargs.pop("constraints", list())
        self.name: str = kwargs.pop("name", None)
        self.db_alias: str = kwargs.pop("alias", None)

        self.primary_key: bool = kwargs.pop("primary_key", False)
        self.autoincrement: bool = kwargs.pop("autoincrement", False)
        self.nullable: bool = kwargs.pop("nullable", False)
        self.sql_nullable: bool = kwargs.pop("sql_nullable", False)
        self.index: bool = kwargs.pop("index", False)
        self.unique: bool = kwargs.pop("unique", False)

        self.virtual: bool = kwargs.pop(
            "virtual", None
        )  # ManyToManyFields and reverse ForeignKeyFields
        self.is_multi: bool = kwargs.pop("is_multi", None)  # ManyToManyField
        self.is_relation: bool = kwargs.pop(
            "is_relation", None
        )  # ForeignKeyField + subclasses
        self.is_through: bool = kwargs.pop("is_through", False)  # ThroughFields

        self.through_relation_name = kwargs.pop("through_relation_name", None)
        self.through_reverse_relation_name = kwargs.pop(
            "through_reverse_relation_name", None
        )

        self.skip_reverse: bool = kwargs.pop("skip_reverse", False)
        self.skip_field: bool = kwargs.pop("skip_field", False)

        self.owner: Type["Model"] = kwargs.pop("owner", None)
        self.to: Type["Model"] = kwargs.pop("to", None)
        self.to_pk_only: Type["Model"] = kwargs.pop("to_pk_only", None)
        self.through: Type["Model"] = kwargs.pop("through", None)
        self.self_reference: bool = kwargs.pop("self_reference", False)
        self.self_reference_primary: Optional[str] = kwargs.pop(
            "self_reference_primary", None
        )
        self.orders_by: Optional[List[str]] = kwargs.pop("orders_by", None)
        self.related_orders_by: Optional[List[str]] = kwargs.pop(
            "related_orders_by", None
        )

        self.encrypt_secret: str = kwargs.pop("encrypt_secret", None)
        self.encrypt_backend: EncryptBackends = kwargs.pop(
            "encrypt_backend", EncryptBackends.NONE
        )
        self.encrypt_custom_backend: Optional[Type[EncryptBackend]] = kwargs.pop(
            "encrypt_custom_backend", None
        )

        self.ormar_default: Any = kwargs.pop("default", None)
        self.server_default: Any = kwargs.pop("server_default", None)

        self.comment: str = kwargs.pop("comment", None)

        self.represent_as_base64_str: bool = kwargs.pop(
            "represent_as_base64_str", False
        )

        for name, value in kwargs.items():
            setattr(self, name, value)

        kwargs.update(self.get_pydantic_default())
        super().__init__(**kwargs)

    def is_valid_uni_relation(self) -> bool:
        """
        Checks if field is a relation definition but only for ForeignKey relation,
        so excludes ManyToMany fields, as well as virtual ForeignKey
        (second side of FK relation).

        Is used to define if a field is a db ForeignKey column that
        should be saved/populated when dealing with internal/own
        Model columns only.

        :return: result of the check
        :rtype: bool
        """
        return not self.is_multi and not self.virtual

    def get_alias(self) -> str:
        """
        Used to translate Model column names to database column names during db queries.

        :return: returns custom database column name if defined by user,
        otherwise field name in ormar/pydantic
        :rtype: str
        """
        return self.db_alias if self.db_alias else self.name

    def get_pydantic_default(self) -> Dict:
        """
        Generates base pydantic.FieldInfo with only default and optionally
        required to fix pydantic Json field being set to required=False.
        Used in an ormar Model Metaclass.

        :return: instance of base pydantic.FieldInfo
        :rtype: pydantic.FieldInfo
        """
        base = self.default_value()
        if base is None:
            base = dict(default=None) if self.nullable else dict(default=_Unset)
        return base

    def default_value(self, use_server: bool = False) -> Optional[Dict]:
        """
        Returns a FieldInfo instance with populated default
        (static) or default_factory (function).
        If the field is a autoincrement primary key the default is None.
        Otherwise field have to has either default, or default_factory populated.

        If all default conditions fail None is returned.

        Used in converting to pydantic FieldInfo.

        :param use_server: flag marking if server_default should be
        treated as default value, default False
        :type use_server: bool
        :return: returns a call to pydantic.Field
        which is returning a FieldInfo instance
        :rtype: Optional[pydantic.FieldInfo]
        """
        if self.is_auto_primary_key():
            return dict(default=None)
        if self.has_default(use_server=use_server):
            default = (
                self.ormar_default
                if self.ormar_default is not None
                else self.server_default
            )
            if callable(default):
                return dict(default_factory=default)
            return dict(default=default)
        return None

    def get_default(
        self, use_server: bool = False, call_default_factory: bool = True
    ) -> Any:  # noqa CCR001
        """
        Return default value for a field.
        If the field is Callable the function is called and actual result is returned.
        Used to populate default_values for pydantic Model in ormar Model Metaclass.

        :param use_server: flag marking if server_default should be
        treated as default value, default False
        :type use_server: bool
        :return: default value for the field if set, otherwise implicit None
        :rtype: Any
        """
        if self.has_default():
            default = (
                self.ormar_default
                if self.ormar_default is not None
                else self._get_default_server_value(use_server=use_server)
            )
            return self._get_default_callable_value(
                default=default,
                call_default_factory=call_default_factory,
            )

    def _get_default_server_value(self, use_server: bool) -> Any:
        """
        Return default value for a server side if use_server is True
        """
        return self.server_default if use_server else None

    @staticmethod
    def _get_default_callable_value(default: Any, call_default_factory: bool) -> Any:
        """
        Return default factory value if call_default_factory is True
        and default is a callable.
        """
        return default() if (callable(default) and call_default_factory) else default

    def has_default(self, use_server: bool = True) -> bool:
        """
        Checks if the field has default value set.

        :param use_server: flag marking if server_default should be
        treated as default value, default False
        :type use_server: bool
        :return: result of the check if default value is set
        :rtype: bool
        """
        return self.ormar_default is not None or (
            self.server_default is not None and use_server
        )

    def is_auto_primary_key(self) -> bool:
        """
        Checks if field is first a primary key and if it,
        it's than check if it's set to autoincrement.
        Autoincrement primary_key is nullable/optional.

        :return: result of the check for primary key and autoincrement
        :rtype: bool
        """
        if self.primary_key:
            return self.autoincrement
        return False

    def construct_constraints(self) -> List:
        """
        Converts list of ormar constraints into sqlalchemy ForeignKeys.
        Has to be done dynamically as sqlalchemy binds ForeignKey to the table.
        And we need a new ForeignKey for subclasses of current model

        :return: List of sqlalchemy foreign keys - by default one.
        :rtype: List[sqlalchemy.schema.ForeignKey]
        """
        constraints = [
            sqlalchemy.ForeignKey(
                con.reference,
                ondelete=con.ondelete,
                onupdate=con.onupdate,
                name=f"fk_{self.owner.ormar_config.tablename}_{self.to.ormar_config.tablename}"
                f"_{self.to.get_column_alias(self.to.ormar_config.pkname)}_{self.name}",
            )
            for con in self.constraints
        ]
        return constraints

    def get_column(self, name: str) -> sqlalchemy.Column:
        """
        Returns definition of sqlalchemy.Column used in creation of sqlalchemy.Table.
        Populates name, column type constraints, as well as a number of parameters like
        primary_key, index, unique, nullable, default and server_default.

        :param name: name of the db column - used if alias is not set
        :type name: str
        :return: actual definition of the database column as sqlalchemy requires.
        :rtype: sqlalchemy.Column
        """
        if self.encrypt_backend == EncryptBackends.NONE:
            column = sqlalchemy.Column(
                self.db_alias or name,
                self.column_type,
                *self.construct_constraints(),
                primary_key=self.primary_key,
                nullable=self.sql_nullable,
                index=self.index,
                unique=self.unique,
                default=self.ormar_default,
                server_default=self.server_default,
                comment=self.comment,
            )
        else:
            column = self._get_encrypted_column(name=name)
        return column

    def _get_encrypted_column(self, name: str) -> sqlalchemy.Column:
        """
        Returns EncryptedString column type instead of actual column.

        :param name: column name
        :type name: str
        :return: newly defined column
        :rtype:  sqlalchemy.Column
        """
        if self.primary_key or self.is_relation:
            raise ModelDefinitionError(
                "Primary key field and relations fields" "cannot be encrypted!"
            )
        column = sqlalchemy.Column(
            self.db_alias or name,
            EncryptedString(
                _field_type=self,
                encrypt_secret=self.encrypt_secret,
                encrypt_backend=self.encrypt_backend,
                encrypt_custom_backend=self.encrypt_custom_backend,
            ),
            nullable=self.nullable,
            index=self.index,
            unique=self.unique,
            default=self.ormar_default,
            server_default=self.server_default,
        )
        return column

    def expand_relationship(
        self,
        value: Any,
        child: Union["Model", "NewBaseModel"],
        to_register: bool = True,
    ) -> Any:
        """
        Function overwritten for relations, in basic field the value is returned as is.
        For relations the child model is first constructed (if needed),
        registered in relation and returned.
        For relation fields the value can be a pk value (Any type of field),
        dict (from Model) or actual instance/list of a "Model".

        :param value: a Model field value, returned untouched for non relation fields.
        :type value: Any
        :param child: a child Model to register
        :type child: Union["Model", "NewBaseModel"]
        :param to_register: flag if the relation should be set in RelationshipManager
        :type to_register: bool
        :return: returns untouched value for normal fields, expands only for relations
        :rtype: Any
        """
        return value

    def set_self_reference_flag(self) -> None:
        """
        Sets `self_reference` to True if field to and owner are same model.
        :return: None
        :rtype: None
        """
        if self.owner is not None and (
            self.owner == self.to or self.owner.ormar_config == self.to.ormar_config
        ):
            self.self_reference = True
            self.self_reference_primary = self.name

    def has_unresolved_forward_refs(self) -> bool:
        """
        Verifies if the filed has any ForwardRefs that require updating before the
        model can be used.

        :return: result of the check
        :rtype: bool
        """
        return False

    def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None:
        """
        Evaluates the ForwardRef to actual Field based on global and local namespaces

        :param globalns: global namespace
        :type globalns: Any
        :param localns: local namespace
        :type localns: Any
        :return: None
        :rtype: None
        """

    def get_related_name(self) -> str:
        """
        Returns name to use for reverse relation.
        It's either set as `related_name` or by default it's owner model. get_name + 's'
        :return: name of the related_name or default related name.
        :rtype: str
        """
        return ""  # pragma: no cover

construct_constraints()

Converts list of ormar constraints into sqlalchemy ForeignKeys. Has to be done dynamically as sqlalchemy binds ForeignKey to the table. And we need a new ForeignKey for subclasses of current model

:return: List of sqlalchemy foreign keys - by default one. :rtype: List[sqlalchemy.schema.ForeignKey]

Source code in ormar/fields/base.py

def construct_constraints(self) -> List:
    """
    Converts list of ormar constraints into sqlalchemy ForeignKeys.
    Has to be done dynamically as sqlalchemy binds ForeignKey to the table.
    And we need a new ForeignKey for subclasses of current model

    :return: List of sqlalchemy foreign keys - by default one.
    :rtype: List[sqlalchemy.schema.ForeignKey]
    """
    constraints = [
        sqlalchemy.ForeignKey(
            con.reference,
            ondelete=con.ondelete,
            onupdate=con.onupdate,
            name=f"fk_{self.owner.ormar_config.tablename}_{self.to.ormar_config.tablename}"
            f"_{self.to.get_column_alias(self.to.ormar_config.pkname)}_{self.name}",
        )
        for con in self.constraints
    ]
    return constraints

default_value(use_server=False)

Returns a FieldInfo instance with populated default (static) or default_factory (function). If the field is a autoincrement primary key the default is None. Otherwise field have to has either default, or default_factory populated.

If all default conditions fail None is returned.

Used in converting to pydantic FieldInfo.

:param use_server: flag marking if server_default should be treated as default value, default False :type use_server: bool :return: returns a call to pydantic.Field which is returning a FieldInfo instance :rtype: Optional[pydantic.FieldInfo]

Source code in ormar/fields/base.py

def default_value(self, use_server: bool = False) -> Optional[Dict]:
    """
    Returns a FieldInfo instance with populated default
    (static) or default_factory (function).
    If the field is a autoincrement primary key the default is None.
    Otherwise field have to has either default, or default_factory populated.

    If all default conditions fail None is returned.

    Used in converting to pydantic FieldInfo.

    :param use_server: flag marking if server_default should be
    treated as default value, default False
    :type use_server: bool
    :return: returns a call to pydantic.Field
    which is returning a FieldInfo instance
    :rtype: Optional[pydantic.FieldInfo]
    """
    if self.is_auto_primary_key():
        return dict(default=None)
    if self.has_default(use_server=use_server):
        default = (
            self.ormar_default
            if self.ormar_default is not None
            else self.server_default
        )
        if callable(default):
            return dict(default_factory=default)
        return dict(default=default)
    return None

evaluate_forward_ref(globalns, localns)

Evaluates the ForwardRef to actual Field based on global and local namespaces

:param globalns: global namespace :type globalns: Any :param localns: local namespace :type localns: Any :return: None :rtype: None

Source code in ormar/fields/base.py

def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None:
    """
    Evaluates the ForwardRef to actual Field based on global and local namespaces

    :param globalns: global namespace
    :type globalns: Any
    :param localns: local namespace
    :type localns: Any
    :return: None
    :rtype: None
    """

expand_relationship(value, child, to_register=True)

Function overwritten for relations, in basic field the value is returned as is. For relations the child model is first constructed (if needed), registered in relation and returned. For relation fields the value can be a pk value (Any type of field), dict (from Model) or actual instance/list of a "Model".

:param value: a Model field value, returned untouched for non relation fields. :type value: Any :param child: a child Model to register :type child: Union["Model", "NewBaseModel"] :param to_register: flag if the relation should be set in RelationshipManager :type to_register: bool :return: returns untouched value for normal fields, expands only for relations :rtype: Any

Source code in ormar/fields/base.py

def expand_relationship(
    self,
    value: Any,
    child: Union["Model", "NewBaseModel"],
    to_register: bool = True,
) -> Any:
    """
    Function overwritten for relations, in basic field the value is returned as is.
    For relations the child model is first constructed (if needed),
    registered in relation and returned.
    For relation fields the value can be a pk value (Any type of field),
    dict (from Model) or actual instance/list of a "Model".

    :param value: a Model field value, returned untouched for non relation fields.
    :type value: Any
    :param child: a child Model to register
    :type child: Union["Model", "NewBaseModel"]
    :param to_register: flag if the relation should be set in RelationshipManager
    :type to_register: bool
    :return: returns untouched value for normal fields, expands only for relations
    :rtype: Any
    """
    return value

get_alias()

Used to translate Model column names to database column names during db queries.

:return: returns custom database column name if defined by user, otherwise field name in ormar/pydantic :rtype: str

Source code in ormar/fields/base.py

def get_alias(self) -> str:
    """
    Used to translate Model column names to database column names during db queries.

    :return: returns custom database column name if defined by user,
    otherwise field name in ormar/pydantic
    :rtype: str
    """
    return self.db_alias if self.db_alias else self.name

get_column(name)

Returns definition of sqlalchemy.Column used in creation of sqlalchemy.Table. Populates name, column type constraints, as well as a number of parameters like primary_key, index, unique, nullable, default and server_default.

:param name: name of the db column - used if alias is not set :type name: str :return: actual definition of the database column as sqlalchemy requires. :rtype: sqlalchemy.Column

Source code in ormar/fields/base.py

def get_column(self, name: str) -> sqlalchemy.Column:
    """
    Returns definition of sqlalchemy.Column used in creation of sqlalchemy.Table.
    Populates name, column type constraints, as well as a number of parameters like
    primary_key, index, unique, nullable, default and server_default.

    :param name: name of the db column - used if alias is not set
    :type name: str
    :return: actual definition of the database column as sqlalchemy requires.
    :rtype: sqlalchemy.Column
    """
    if self.encrypt_backend == EncryptBackends.NONE:
        column = sqlalchemy.Column(
            self.db_alias or name,
            self.column_type,
            *self.construct_constraints(),
            primary_key=self.primary_key,
            nullable=self.sql_nullable,
            index=self.index,
            unique=self.unique,
            default=self.ormar_default,
            server_default=self.server_default,
            comment=self.comment,
        )
    else:
        column = self._get_encrypted_column(name=name)
    return column

get_default(use_server=False, call_default_factory=True)

Return default value for a field. If the field is Callable the function is called and actual result is returned. Used to populate default_values for pydantic Model in ormar Model Metaclass.

:param use_server: flag marking if server_default should be treated as default value, default False :type use_server: bool :return: default value for the field if set, otherwise implicit None :rtype: Any

Source code in ormar/fields/base.py

def get_default(
    self, use_server: bool = False, call_default_factory: bool = True
) -> Any:  # noqa CCR001
    """
    Return default value for a field.
    If the field is Callable the function is called and actual result is returned.
    Used to populate default_values for pydantic Model in ormar Model Metaclass.

    :param use_server: flag marking if server_default should be
    treated as default value, default False
    :type use_server: bool
    :return: default value for the field if set, otherwise implicit None
    :rtype: Any
    """
    if self.has_default():
        default = (
            self.ormar_default
            if self.ormar_default is not None
            else self._get_default_server_value(use_server=use_server)
        )
        return self._get_default_callable_value(
            default=default,
            call_default_factory=call_default_factory,
        )

get_pydantic_default()

Generates base pydantic.FieldInfo with only default and optionally required to fix pydantic Json field being set to required=False. Used in an ormar Model Metaclass.

:return: instance of base pydantic.FieldInfo :rtype: pydantic.FieldInfo

Source code in ormar/fields/base.py

def get_pydantic_default(self) -> Dict:
    """
    Generates base pydantic.FieldInfo with only default and optionally
    required to fix pydantic Json field being set to required=False.
    Used in an ormar Model Metaclass.

    :return: instance of base pydantic.FieldInfo
    :rtype: pydantic.FieldInfo
    """
    base = self.default_value()
    if base is None:
        base = dict(default=None) if self.nullable else dict(default=_Unset)
    return base

get_related_name()

Returns name to use for reverse relation. It's either set as related_name or by default it's owner model. get_name + 's' :return: name of the related_name or default related name. :rtype: str

Source code in ormar/fields/base.py

def get_related_name(self) -> str:
    """
    Returns name to use for reverse relation.
    It's either set as `related_name` or by default it's owner model. get_name + 's'
    :return: name of the related_name or default related name.
    :rtype: str
    """
    return ""  # pragma: no cover

has_default(use_server=True)

Checks if the field has default value set.

:param use_server: flag marking if server_default should be treated as default value, default False :type use_server: bool :return: result of the check if default value is set :rtype: bool

Source code in ormar/fields/base.py

def has_default(self, use_server: bool = True) -> bool:
    """
    Checks if the field has default value set.

    :param use_server: flag marking if server_default should be
    treated as default value, default False
    :type use_server: bool
    :return: result of the check if default value is set
    :rtype: bool
    """
    return self.ormar_default is not None or (
        self.server_default is not None and use_server
    )

has_unresolved_forward_refs()

Verifies if the filed has any ForwardRefs that require updating before the model can be used.

:return: result of the check :rtype: bool

Source code in ormar/fields/base.py

def has_unresolved_forward_refs(self) -> bool:
    """
    Verifies if the filed has any ForwardRefs that require updating before the
    model can be used.

    :return: result of the check
    :rtype: bool
    """
    return False

is_auto_primary_key()

Checks if field is first a primary key and if it, it's than check if it's set to autoincrement. Autoincrement primary_key is nullable/optional.

:return: result of the check for primary key and autoincrement :rtype: bool

Source code in ormar/fields/base.py

def is_auto_primary_key(self) -> bool:
    """
    Checks if field is first a primary key and if it,
    it's than check if it's set to autoincrement.
    Autoincrement primary_key is nullable/optional.

    :return: result of the check for primary key and autoincrement
    :rtype: bool
    """
    if self.primary_key:
        return self.autoincrement
    return False

is_valid_uni_relation()

Checks if field is a relation definition but only for ForeignKey relation, so excludes ManyToMany fields, as well as virtual ForeignKey (second side of FK relation).

Is used to define if a field is a db ForeignKey column that should be saved/populated when dealing with internal/own Model columns only.

:return: result of the check :rtype: bool

Source code in ormar/fields/base.py

def is_valid_uni_relation(self) -> bool:
    """
    Checks if field is a relation definition but only for ForeignKey relation,
    so excludes ManyToMany fields, as well as virtual ForeignKey
    (second side of FK relation).

    Is used to define if a field is a db ForeignKey column that
    should be saved/populated when dealing with internal/own
    Model columns only.

    :return: result of the check
    :rtype: bool
    """
    return not self.is_multi and not self.virtual

set_self_reference_flag()

Sets self_reference to True if field to and owner are same model. :return: None :rtype: None

Source code in ormar/fields/base.py

def set_self_reference_flag(self) -> None:
    """
    Sets `self_reference` to True if field to and owner are same model.
    :return: None
    :rtype: None
    """
    if self.owner is not None and (
        self.owner == self.to or self.owner.ormar_config == self.to.ormar_config
    ):
        self.self_reference = True
        self.self_reference_primary = self.name

BigInteger

Bases: Integer, int

BigInteger field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class BigInteger(Integer, int):
    """
    BigInteger field factory that construct Field classes and populated their values.
    """

    _type = int
    _sample = 0

    def __new__(  # type: ignore
        cls,
        *,
        minimum: Optional[int] = None,
        maximum: Optional[int] = None,
        multiple_of: Optional[int] = None,
        **kwargs: Any
    ) -> BaseField:
        autoincrement = kwargs.pop("autoincrement", None)
        autoincrement = (
            autoincrement
            if autoincrement is not None
            else kwargs.get("primary_key", False)
        )
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        kwargs["ge"] = kwargs["minimum"]
        kwargs["le"] = kwargs["maximum"]
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.BigInteger()

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.BigInteger()

Boolean

Bases: ModelFieldFactory, int

Boolean field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class Boolean(ModelFieldFactory, int):
    """
    Boolean field factory that construct Field classes and populated their values.
    """

    _type = bool
    _sample = True

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.Boolean()

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.Boolean()

CheckColumns

Bases: CheckConstraint

Subclass of sqlalchemy.CheckConstraint. Used to avoid importing anything from sqlalchemy by user.

Note that some databases do not actively support check constraints such as MySQL.

Source code in ormar/fields/constraints.py

class CheckColumns(CheckConstraint):
    """
    Subclass of sqlalchemy.CheckConstraint.
    Used to avoid importing anything from sqlalchemy by user.

    Note that some databases do not actively support check constraints such as MySQL.
    """

Date

Bases: ModelFieldFactory, date

Date field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class Date(ModelFieldFactory, datetime.date):
    """
    Date field factory that construct Field classes and populated their values.
    """

    _type = datetime.date
    _sample = "date"

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.Date()

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.Date()

DateTime

Bases: ModelFieldFactory, datetime

DateTime field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class DateTime(ModelFieldFactory, datetime.datetime):
    """
    DateTime field factory that construct Field classes and populated their values.
    """

    _type = datetime.datetime
    _sample = "datetime"

    def __new__(  # type: ignore # noqa CFQ002
        cls, *, timezone: bool = False, **kwargs: Any
    ) -> BaseField:  # type: ignore
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.DateTime(timezone=kwargs.get("timezone", False))

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.DateTime(timezone=kwargs.get("timezone", False))

Decimal

Bases: ModelFieldFactory, Decimal

Decimal field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class Decimal(ModelFieldFactory, decimal.Decimal):
    """
    Decimal field factory that construct Field classes and populated their values.
    """

    _type = decimal.Decimal
    _sample = 0.0

    def __new__(  # type: ignore # noqa CFQ002
        cls,
        *,
        minimum: Optional[float] = None,
        maximum: Optional[float] = None,
        multiple_of: Optional[int] = None,
        precision: Optional[int] = None,
        scale: Optional[int] = None,
        max_digits: Optional[int] = None,
        decimal_places: Optional[int] = None,
        **kwargs: Any
    ) -> BaseField:
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        kwargs["ge"] = kwargs["minimum"]
        kwargs["le"] = kwargs["maximum"]

        if kwargs.get("max_digits"):
            kwargs["precision"] = kwargs["max_digits"]
        elif kwargs.get("precision"):
            kwargs["max_digits"] = kwargs["precision"]

        if kwargs.get("decimal_places"):
            kwargs["scale"] = kwargs["decimal_places"]
        elif kwargs.get("scale"):
            kwargs["decimal_places"] = kwargs["scale"]

        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        precision = kwargs.get("precision")
        scale = kwargs.get("scale")
        return sqlalchemy.DECIMAL(precision=precision, scale=scale)

    @classmethod
    def validate(cls, **kwargs: Any) -> None:
        """
        Used to validate if all required parameters on a given field type are set.
        :param kwargs: all params passed during construction
        :type kwargs: Any
        """
        precision = kwargs.get("precision")
        scale = kwargs.get("scale")
        if precision is None or precision < 0 or scale is None or scale < 0:
            raise ModelDefinitionError(
                "Parameters scale and precision are required for field Decimal"
            )

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    precision = kwargs.get("precision")
    scale = kwargs.get("scale")
    return sqlalchemy.DECIMAL(precision=precision, scale=scale)

validate(**kwargs) classmethod

Used to validate if all required parameters on a given field type are set. :param kwargs: all params passed during construction :type kwargs: Any

Source code in ormar/fields/model_fields.py

@classmethod
def validate(cls, **kwargs: Any) -> None:
    """
    Used to validate if all required parameters on a given field type are set.
    :param kwargs: all params passed during construction
    :type kwargs: Any
    """
    precision = kwargs.get("precision")
    scale = kwargs.get("scale")
    if precision is None or precision < 0 or scale is None or scale < 0:
        raise ModelDefinitionError(
            "Parameters scale and precision are required for field Decimal"
        )

Enum

Bases: ModelFieldFactory

Enum field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class Enum(ModelFieldFactory):
    """
    Enum field factory that construct Field classes and populated their values.
    """

    _type = E
    _sample = None

    def __new__(  # type: ignore # noqa CFQ002
        cls, *, enum_class: Type[E], **kwargs: Any
    ) -> BaseField:
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        return super().__new__(cls, **kwargs)

    @classmethod
    def validate(cls, **kwargs: Any) -> None:
        enum_class = kwargs.get("enum_class")
        if enum_class is None or not isinstance(enum_class, EnumMeta):
            raise ModelDefinitionError("Enum Field choices must be EnumType")

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        enum_cls = kwargs.get("enum_class")
        return sqlalchemy.Enum(enum_cls)

ExcludableItems

Keeps a dictionary of Excludables by alias + model_name keys to allow quick lookup by nested models without need to travers deeply nested dictionaries and passing include/exclude around

Source code in ormar/models/excludable.py

class ExcludableItems:
    """
    Keeps a dictionary of Excludables by alias + model_name keys
    to allow quick lookup by nested models without need to travers
    deeply nested dictionaries and passing include/exclude around
    """

    def __init__(self) -> None:
        self.items: Dict[str, Excludable] = dict()

    @classmethod
    def from_excludable(cls, other: "ExcludableItems") -> "ExcludableItems":
        """
        Copy passed ExcludableItems to avoid inplace modifications.

        :param other: other excludable items to be copied
        :type other: ormar.models.excludable.ExcludableItems
        :return: copy of other
        :rtype: ormar.models.excludable.ExcludableItems
        """
        new_excludable = cls()
        for key, value in other.items.items():
            new_excludable.items[key] = value.get_copy()
        return new_excludable

    def include_entry_count(self) -> int:
        """
        Returns count of include items inside
        """
        count = 0
        for key in self.items.keys():
            count += len(self.items[key].include)
        return count

    def get(self, model_cls: Type["Model"], alias: str = "") -> Excludable:
        """
        Return Excludable for given model and alias.

        :param model_cls: target model to check
        :type model_cls: ormar.models.metaclass.ModelMetaclass
        :param alias: table alias from relation manager
        :type alias: str
        :return: Excludable for given model and alias
        :rtype: ormar.models.excludable.Excludable
        """
        key = f"{alias + '_' if alias else ''}{model_cls.get_name(lower=True)}"
        excludable = self.items.get(key)
        if not excludable:
            excludable = Excludable()
            self.items[key] = excludable
        return excludable

    def build(
        self,
        items: Union[List[str], str, Tuple[str], Set[str], Dict],
        model_cls: Type["Model"],
        is_exclude: bool = False,
    ) -> None:
        """
        Receives the one of the types of items and parses them as to achieve
        a end situation with one excludable per alias/model in relation.

        Each excludable has two sets of values - one to include, one to exclude.

        :param items: values to be included or excluded
        :type items: Union[List[str], str, Tuple[str], Set[str], Dict]
        :param model_cls: source model from which relations are constructed
        :type model_cls: ormar.models.metaclass.ModelMetaclass
        :param is_exclude: flag if items should be included or excluded
        :type is_exclude: bool
        """
        if isinstance(items, str):
            items = {items}

        if isinstance(items, Dict):
            self._traverse_dict(
                values=items,
                source_model=model_cls,
                model_cls=model_cls,
                is_exclude=is_exclude,
            )

        else:
            items = set(items)
            nested_items = set(x for x in items if "__" in x)
            items.difference_update(nested_items)
            self._set_excludes(
                items=items,
                model_name=model_cls.get_name(lower=True),
                is_exclude=is_exclude,
            )
            if nested_items:
                self._traverse_list(
                    values=nested_items, model_cls=model_cls, is_exclude=is_exclude
                )

    def _set_excludes(
        self, items: Set, model_name: str, is_exclude: bool, alias: str = ""
    ) -> None:
        """
        Sets set of values to be included or excluded for given key and model.

        :param items: items to include/exclude
        :type items: set
        :param model_name: name of model to construct key
        :type model_name: str
        :param is_exclude: flag if values should be included or excluded
        :type is_exclude: bool
        :param alias:
        :type alias: str
        """
        key = f"{alias + '_' if alias else ''}{model_name}"
        excludable = self.items.get(key)
        if not excludable:
            excludable = Excludable()
        excludable.set_values(value=items, is_exclude=is_exclude)
        self.items[key] = excludable

    def _traverse_dict(  # noqa: CFQ002
        self,
        values: Dict,
        source_model: Type["Model"],
        model_cls: Type["Model"],
        is_exclude: bool,
        related_items: Optional[List] = None,
        alias: str = "",
    ) -> None:
        """
        Goes through dict of nested values and construct/update Excludables.

        :param values: items to include/exclude
        :type values: Dict
        :param source_model: source model from which relations are constructed
        :type source_model: ormar.models.metaclass.ModelMetaclass
        :param model_cls: model from which current relation is constructed
        :type model_cls: ormar.models.metaclass.ModelMetaclass
        :param is_exclude: flag if values should be included or excluded
        :type is_exclude: bool
        :param related_items: list of names of related fields chain
        :type related_items: List
        :param alias: alias of relation
        :type alias: str
        """
        self_fields = set()
        related_items = related_items[:] if related_items else []
        for key, value in values.items():
            if value is ...:
                self_fields.add(key)
            elif isinstance(value, set):
                (
                    table_prefix,
                    target_model,
                    _,
                    _,
                ) = get_relationship_alias_model_and_str(
                    source_model=source_model, related_parts=related_items + [key]
                )
                self._set_excludes(
                    items=value,
                    model_name=target_model.get_name(),
                    is_exclude=is_exclude,
                    alias=table_prefix,
                )
            else:
                # dict
                related_items.append(key)
                (
                    table_prefix,
                    target_model,
                    _,
                    _,
                ) = get_relationship_alias_model_and_str(
                    source_model=source_model, related_parts=related_items
                )
                self._traverse_dict(
                    values=value,
                    source_model=source_model,
                    model_cls=target_model,
                    is_exclude=is_exclude,
                    related_items=related_items,
                    alias=table_prefix,
                )
        if self_fields:
            self._set_excludes(
                items=self_fields,
                model_name=model_cls.get_name(),
                is_exclude=is_exclude,
                alias=alias,
            )

    def _traverse_list(
        self, values: Set[str], model_cls: Type["Model"], is_exclude: bool
    ) -> None:
        """
        Goes through list of values and construct/update Excludables.

        :param values: items to include/exclude
        :type values: set
        :param model_cls: model from which current relation is constructed
        :type model_cls: ormar.models.metaclass.ModelMetaclass
        :param is_exclude: flag if values should be included or excluded
        :type is_exclude: bool
        """
        # here we have only nested related keys
        for key in values:
            key_split = key.split("__")
            related_items, field_name = key_split[:-1], key_split[-1]
            (table_prefix, target_model, _, _) = get_relationship_alias_model_and_str(
                source_model=model_cls, related_parts=related_items
            )
            self._set_excludes(
                items={field_name},
                model_name=target_model.get_name(),
                is_exclude=is_exclude,
                alias=table_prefix,
            )

build(items, model_cls, is_exclude=False)

Receives the one of the types of items and parses them as to achieve a end situation with one excludable per alias/model in relation.

Each excludable has two sets of values - one to include, one to exclude.

:param items: values to be included or excluded :type items: Union[List[str], str, Tuple[str], Set[str], Dict] :param model_cls: source model from which relations are constructed :type model_cls: ormar.models.metaclass.ModelMetaclass :param is_exclude: flag if items should be included or excluded :type is_exclude: bool

Source code in ormar/models/excludable.py

def build(
    self,
    items: Union[List[str], str, Tuple[str], Set[str], Dict],
    model_cls: Type["Model"],
    is_exclude: bool = False,
) -> None:
    """
    Receives the one of the types of items and parses them as to achieve
    a end situation with one excludable per alias/model in relation.

    Each excludable has two sets of values - one to include, one to exclude.

    :param items: values to be included or excluded
    :type items: Union[List[str], str, Tuple[str], Set[str], Dict]
    :param model_cls: source model from which relations are constructed
    :type model_cls: ormar.models.metaclass.ModelMetaclass
    :param is_exclude: flag if items should be included or excluded
    :type is_exclude: bool
    """
    if isinstance(items, str):
        items = {items}

    if isinstance(items, Dict):
        self._traverse_dict(
            values=items,
            source_model=model_cls,
            model_cls=model_cls,
            is_exclude=is_exclude,
        )

    else:
        items = set(items)
        nested_items = set(x for x in items if "__" in x)
        items.difference_update(nested_items)
        self._set_excludes(
            items=items,
            model_name=model_cls.get_name(lower=True),
            is_exclude=is_exclude,
        )
        if nested_items:
            self._traverse_list(
                values=nested_items, model_cls=model_cls, is_exclude=is_exclude
            )

from_excludable(other) classmethod

Copy passed ExcludableItems to avoid inplace modifications.

:param other: other excludable items to be copied :type other: ormar.models.excludable.ExcludableItems :return: copy of other :rtype: ormar.models.excludable.ExcludableItems

Source code in ormar/models/excludable.py

@classmethod
def from_excludable(cls, other: "ExcludableItems") -> "ExcludableItems":
    """
    Copy passed ExcludableItems to avoid inplace modifications.

    :param other: other excludable items to be copied
    :type other: ormar.models.excludable.ExcludableItems
    :return: copy of other
    :rtype: ormar.models.excludable.ExcludableItems
    """
    new_excludable = cls()
    for key, value in other.items.items():
        new_excludable.items[key] = value.get_copy()
    return new_excludable

get(model_cls, alias='')

Return Excludable for given model and alias.

:param model_cls: target model to check :type model_cls: ormar.models.metaclass.ModelMetaclass :param alias: table alias from relation manager :type alias: str :return: Excludable for given model and alias :rtype: ormar.models.excludable.Excludable

Source code in ormar/models/excludable.py

def get(self, model_cls: Type["Model"], alias: str = "") -> Excludable:
    """
    Return Excludable for given model and alias.

    :param model_cls: target model to check
    :type model_cls: ormar.models.metaclass.ModelMetaclass
    :param alias: table alias from relation manager
    :type alias: str
    :return: Excludable for given model and alias
    :rtype: ormar.models.excludable.Excludable
    """
    key = f"{alias + '_' if alias else ''}{model_cls.get_name(lower=True)}"
    excludable = self.items.get(key)
    if not excludable:
        excludable = Excludable()
        self.items[key] = excludable
    return excludable

include_entry_count()

Returns count of include items inside

Source code in ormar/models/excludable.py

def include_entry_count(self) -> int:
    """
    Returns count of include items inside
    """
    count = 0
    for key in self.items.keys():
        count += len(self.items[key].include)
    return count

Float

Bases: ModelFieldFactory, float

Float field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class Float(ModelFieldFactory, float):
    """
    Float field factory that construct Field classes and populated their values.
    """

    _type = float
    _sample = 0.0

    def __new__(  # type: ignore
        cls,
        *,
        minimum: Optional[float] = None,
        maximum: Optional[float] = None,
        multiple_of: Optional[int] = None,
        **kwargs: Any
    ) -> BaseField:
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        kwargs["ge"] = kwargs["minimum"]
        kwargs["le"] = kwargs["maximum"]
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.Float()

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.Float()

ForeignKeyField

Bases: BaseField

Actual class returned from ForeignKey function call and stored in model_fields.

Source code in ormar/fields/foreign_key.py

class ForeignKeyField(BaseField):
    """
    Actual class returned from ForeignKey function call and stored in model_fields.
    """

    def __init__(self, **kwargs: Any) -> None:
        if TYPE_CHECKING:  # pragma: no cover
            self.__type__: type
            self.to: Type["Model"]
        self.ondelete: str = kwargs.pop("ondelete", None)
        self.onupdate: str = kwargs.pop("onupdate", None)
        super().__init__(**kwargs)

    def get_source_related_name(self) -> str:
        """
        Returns name to use for source relation name.
        For FK it's the same, differs for m2m fields.
        It's either set as `related_name` or by default it's owner model. get_name + 's'
        :return: name of the related_name or default related name.
        :rtype: str
        """
        return self.get_related_name()

    def get_related_name(self) -> str:
        """
        Returns name to use for reverse relation.
        It's either set as `related_name` or by default it's owner model. get_name + 's'
        :return: name of the related_name or default related name.
        :rtype: str
        """
        return self.related_name or self.owner.get_name() + "s"

    def default_target_field_name(self) -> str:
        """
        Returns default target model name on through model.
        :return: name of the field
        :rtype: str
        """
        prefix = "from_" if self.self_reference else ""
        return self.through_reverse_relation_name or f"{prefix}{self.to.get_name()}"

    def default_source_field_name(self) -> str:
        """
        Returns default target model name on through model.
        :return: name of the field
        :rtype: str
        """
        prefix = "to_" if self.self_reference else ""
        return self.through_relation_name or f"{prefix}{self.owner.get_name()}"

    def get_filter_clause_target(self) -> Type["Model"]:
        return self.to

    def get_model_relation_fields(self, use_alias: bool = False) -> str:
        """
        Extract names of the database columns or model fields that are connected
        with given relation based on use_alias switch and which side of the relation
        the current field is - reverse or normal.

        :param use_alias: use db names aliases or model fields
        :type use_alias: bool
        :return: name or names of the related columns/ fields
        :rtype: Union[str, List[str]]
        """
        if use_alias:
            return self._get_model_relation_fields_alias()
        return self._get_model_relation_fields_name()

    def _get_model_relation_fields_name(self) -> str:
        if self.virtual:
            return self.owner.ormar_config.pkname
        return self.name

    def _get_model_relation_fields_alias(self) -> str:
        if self.virtual:
            return self.owner.ormar_config.model_fields[
                self.owner.ormar_config.pkname
            ].get_alias()
        return self.get_alias()

    def get_related_field_alias(self) -> str:
        """
        Extract names of the related database columns or that are connected
        with given relation based to use as a target in filter clause.

        :return: name or names of the related columns/ fields
        :rtype: Union[str, Dict[str, str]]
        """
        if self.virtual:
            field_name = self.get_related_name()
            field = self.to.ormar_config.model_fields[field_name]
            return field.get_alias()
        target_field = self.to.get_column_alias(self.to.ormar_config.pkname)
        return target_field

    def get_related_field_name(self) -> Union[str, List[str]]:
        """
        Returns name of the relation field that should be used in prefetch query.
        This field is later used to register relation in prefetch query,
        populate relations dict, and populate nested model in prefetch query.

        :return: name(s) of the field
        :rtype: Union[str, List[str]]
        """
        if self.virtual:
            return self.get_related_name()
        return self.to.ormar_config.pkname

    def _evaluate_forward_ref(
        self, globalns: Any, localns: Any, is_through: bool = False
    ) -> None:
        target = "through" if is_through else "to"
        target_obj = getattr(self, target)
        if sys.version_info.minor <= 8:  # pragma: no cover
            evaluated = target_obj._evaluate(globalns, localns)
        else:  # pragma: no cover
            evaluated = target_obj._evaluate(globalns, localns, recursive_guard=set())
        setattr(self, target, evaluated)

    def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None:
        """
        Evaluates the ForwardRef to actual Field based on global and local namespaces

        :param globalns: global namespace
        :type globalns: Any
        :param localns: local namespace
        :type localns: Any
        :return: None
        :rtype: None
        """
        if self.to.__class__ == ForwardRef:
            self._evaluate_forward_ref(globalns, localns)
            (
                self.__type__,
                self.constraints,
                self.column_type,
                self.to_pk_only,
            ) = populate_fk_params_based_on_to_model(
                to=self.to,
                nullable=self.nullable,
                ondelete=self.ondelete,
                onupdate=self.onupdate,
            )

    def _extract_model_from_sequence(
        self, value: List, child: "Model", to_register: bool
    ) -> List["Model"]:
        """
        Takes a list of Models and registers them on parent.
        Registration is mutual, so children have also reference to parent.

        Used in reverse FK relations.

        :param value: list of Model
        :type value: List
        :param child: child/ related Model
        :type child: Model
        :param to_register: flag if the relation should be set in RelationshipManager
        :type to_register: bool
        :return: list (if needed) registered Models
        :rtype: List["Model"]
        """
        return [
            self.expand_relationship(  # type: ignore
                value=val, child=child, to_register=to_register
            )
            for val in value
        ]

    def _register_existing_model(
        self, value: "Model", child: "Model", to_register: bool
    ) -> "Model":
        """
        Takes already created instance and registers it for parent.
        Registration is mutual, so children have also reference to parent.

        Used in reverse FK relations and normal FK for single models.

        :param value: already instantiated Model
        :type value: Model
        :param child: child/ related Model
        :type child: Model
        :param to_register: flag if the relation should be set in RelationshipManager
        :type to_register: bool
        :return: (if needed) registered Model
        :rtype: Model
        """
        if to_register:
            self.register_relation(model=value, child=child)
        return value

    def _construct_model_from_dict(
        self, value: dict, child: "Model", to_register: bool
    ) -> "Model":
        """
        Takes a dictionary, creates a instance and registers it for parent.
        If dictionary contains only one field and it's a pk it is a __pk_only__ model.
        Registration is mutual, so children have also reference to parent.

        Used in normal FK for dictionaries.

        :param value: dictionary of a Model
        :type value: dict
        :param child: child/ related Model
        :type child: Model
        :param to_register: flag if the relation should be set in RelationshipManager
        :type to_register: bool
        :return: (if needed) registered Model
        :rtype: Model
        """
        pk_only_model = None
        keys = set(value.keys())
        own_keys = keys - self.to.extract_related_names()
        if (
            len(own_keys) == 1
            and list(own_keys)[0] == self.to.ormar_config.pkname
            and value.get(self.to.ormar_config.pkname) is not None
            and not self.is_through
        ):
            value["__pk_only__"] = True
            pk_only_model = self.to_pk_only(**value)
        model = self.to(**value)
        if to_register:
            self.register_relation(model=model, child=child)
        return pk_only_model if pk_only_model is not None else model

    def _construct_model_from_pk(
        self, value: Any, child: "Model", to_register: bool
    ) -> "Model":
        """
        Takes a pk value, creates a dummy instance and registers it for parent.
        Registration is mutual, so children have also reference to parent.

        Used in normal FK for dictionaries.

        :param value: value of a related pk / fk column
        :type value: Any
        :param child: child/ related Model
        :type child: Model
        :param to_register: flag if the relation should be set in RelationshipManager
        :type to_register: bool
        :return: (if needed) registered Model
        :rtype: Model
        """
        if self.to.pk_type() == uuid.UUID and isinstance(value, str):  # pragma: nocover
            value = uuid.UUID(value)
        if not isinstance(value, self.to.pk_type()):
            if isinstance(value, self.to_pk_only):
                value = getattr(value, self.to.ormar_config.pkname)
            else:
                raise RelationshipInstanceError(
                    f"Relationship error - ForeignKey {self.to.__name__} "
                    f"is of type {self.to.pk_type()} "
                    f"while {type(value)} passed as a parameter."
                )
        model = create_dummy_instance(fk=self.to, pk=value)
        if to_register:
            self.register_relation(model=model, child=child)
        return model

    def register_relation(self, model: "Model", child: "Model") -> None:
        """
        Registers relation between parent and child in relation manager.
        Relation manager is kep on each model (different instance).

        Used in Metaclass and sometimes some relations are missing
        (i.e. cloned Models in fastapi might miss one).

        :param model: parent model (with relation definition)
        :type model: Model class
        :param child: child model
        :type child: Model class
        """
        model._orm.add(parent=model, child=child, field=self)

    def has_unresolved_forward_refs(self) -> bool:
        """
        Verifies if the filed has any ForwardRefs that require updating before the
        model can be used.

        :return: result of the check
        :rtype: bool
        """
        return self.to.__class__ == ForwardRef

    def expand_relationship(
        self,
        value: Any,
        child: Union["Model", "NewBaseModel"],
        to_register: bool = True,
    ) -> Optional[Union["Model", List["Model"]]]:
        """
        For relations the child model is first constructed (if needed),
        registered in relation and returned.
        For relation fields the value can be a pk value (Any type of field),
        dict (from Model) or actual instance/list of a "Model".

        Selects the appropriate constructor based on a passed value.

        :param value: a Model field value, returned untouched for non relation fields.
        :type value: Any
        :param child: a child Model to register
        :type child: Union["Model", "NewBaseModel"]
        :param to_register: flag if the relation should be set in RelationshipManager
        :type to_register: bool
        :return: returns a Model or a list of Models
        :rtype: Optional[Union["Model", List["Model"]]]
        """
        if value is None:
            return None if not self.virtual else []
        constructors = {
            f"{self.to.__name__}": self._register_existing_model,
            "dict": self._construct_model_from_dict,
            "list": self._extract_model_from_sequence,
        }

        model = constructors.get(  # type: ignore
            value.__class__.__name__, self._construct_model_from_pk
        )(value, child, to_register)
        return model

    def get_relation_name(self) -> str:  # pragma: no cover
        """
        Returns name of the relation, which can be a own name or through model
        names for m2m models

        :return: result of the check
        :rtype: bool
        """
        return self.name

    def get_source_model(self) -> Type["Model"]:  # pragma: no cover
        """
        Returns model from which the relation comes -> either owner or through model

        :return: source model
        :rtype: Type["Model"]
        """
        return self.owner

default_source_field_name()

Returns default target model name on through model. :return: name of the field :rtype: str

Source code in ormar/fields/foreign_key.py

def default_source_field_name(self) -> str:
    """
    Returns default target model name on through model.
    :return: name of the field
    :rtype: str
    """
    prefix = "to_" if self.self_reference else ""
    return self.through_relation_name or f"{prefix}{self.owner.get_name()}"

default_target_field_name()

Returns default target model name on through model. :return: name of the field :rtype: str

Source code in ormar/fields/foreign_key.py

def default_target_field_name(self) -> str:
    """
    Returns default target model name on through model.
    :return: name of the field
    :rtype: str
    """
    prefix = "from_" if self.self_reference else ""
    return self.through_reverse_relation_name or f"{prefix}{self.to.get_name()}"

evaluate_forward_ref(globalns, localns)

Evaluates the ForwardRef to actual Field based on global and local namespaces

:param globalns: global namespace :type globalns: Any :param localns: local namespace :type localns: Any :return: None :rtype: None

Source code in ormar/fields/foreign_key.py

def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None:
    """
    Evaluates the ForwardRef to actual Field based on global and local namespaces

    :param globalns: global namespace
    :type globalns: Any
    :param localns: local namespace
    :type localns: Any
    :return: None
    :rtype: None
    """
    if self.to.__class__ == ForwardRef:
        self._evaluate_forward_ref(globalns, localns)
        (
            self.__type__,
            self.constraints,
            self.column_type,
            self.to_pk_only,
        ) = populate_fk_params_based_on_to_model(
            to=self.to,
            nullable=self.nullable,
            ondelete=self.ondelete,
            onupdate=self.onupdate,
        )

expand_relationship(value, child, to_register=True)

For relations the child model is first constructed (if needed), registered in relation and returned. For relation fields the value can be a pk value (Any type of field), dict (from Model) or actual instance/list of a "Model".

Selects the appropriate constructor based on a passed value.

:param value: a Model field value, returned untouched for non relation fields. :type value: Any :param child: a child Model to register :type child: Union["Model", "NewBaseModel"] :param to_register: flag if the relation should be set in RelationshipManager :type to_register: bool :return: returns a Model or a list of Models :rtype: Optional[Union["Model", List["Model"]]]

Source code in ormar/fields/foreign_key.py

def expand_relationship(
    self,
    value: Any,
    child: Union["Model", "NewBaseModel"],
    to_register: bool = True,
) -> Optional[Union["Model", List["Model"]]]:
    """
    For relations the child model is first constructed (if needed),
    registered in relation and returned.
    For relation fields the value can be a pk value (Any type of field),
    dict (from Model) or actual instance/list of a "Model".

    Selects the appropriate constructor based on a passed value.

    :param value: a Model field value, returned untouched for non relation fields.
    :type value: Any
    :param child: a child Model to register
    :type child: Union["Model", "NewBaseModel"]
    :param to_register: flag if the relation should be set in RelationshipManager
    :type to_register: bool
    :return: returns a Model or a list of Models
    :rtype: Optional[Union["Model", List["Model"]]]
    """
    if value is None:
        return None if not self.virtual else []
    constructors = {
        f"{self.to.__name__}": self._register_existing_model,
        "dict": self._construct_model_from_dict,
        "list": self._extract_model_from_sequence,
    }

    model = constructors.get(  # type: ignore
        value.__class__.__name__, self._construct_model_from_pk
    )(value, child, to_register)
    return model

get_model_relation_fields(use_alias=False)

Extract names of the database columns or model fields that are connected with given relation based on use_alias switch and which side of the relation the current field is - reverse or normal.

:param use_alias: use db names aliases or model fields :type use_alias: bool :return: name or names of the related columns/ fields :rtype: Union[str, List[str]]

Source code in ormar/fields/foreign_key.py

def get_model_relation_fields(self, use_alias: bool = False) -> str:
    """
    Extract names of the database columns or model fields that are connected
    with given relation based on use_alias switch and which side of the relation
    the current field is - reverse or normal.

    :param use_alias: use db names aliases or model fields
    :type use_alias: bool
    :return: name or names of the related columns/ fields
    :rtype: Union[str, List[str]]
    """
    if use_alias:
        return self._get_model_relation_fields_alias()
    return self._get_model_relation_fields_name()

get_related_field_alias()

Extract names of the related database columns or that are connected with given relation based to use as a target in filter clause.

:return: name or names of the related columns/ fields :rtype: Union[str, Dict[str, str]]

Source code in ormar/fields/foreign_key.py

def get_related_field_alias(self) -> str:
    """
    Extract names of the related database columns or that are connected
    with given relation based to use as a target in filter clause.

    :return: name or names of the related columns/ fields
    :rtype: Union[str, Dict[str, str]]
    """
    if self.virtual:
        field_name = self.get_related_name()
        field = self.to.ormar_config.model_fields[field_name]
        return field.get_alias()
    target_field = self.to.get_column_alias(self.to.ormar_config.pkname)
    return target_field

get_related_field_name()

Returns name of the relation field that should be used in prefetch query. This field is later used to register relation in prefetch query, populate relations dict, and populate nested model in prefetch query.

:return: name(s) of the field :rtype: Union[str, List[str]]

Source code in ormar/fields/foreign_key.py

def get_related_field_name(self) -> Union[str, List[str]]:
    """
    Returns name of the relation field that should be used in prefetch query.
    This field is later used to register relation in prefetch query,
    populate relations dict, and populate nested model in prefetch query.

    :return: name(s) of the field
    :rtype: Union[str, List[str]]
    """
    if self.virtual:
        return self.get_related_name()
    return self.to.ormar_config.pkname

get_related_name()

Returns name to use for reverse relation. It's either set as related_name or by default it's owner model. get_name + 's' :return: name of the related_name or default related name. :rtype: str

Source code in ormar/fields/foreign_key.py

def get_related_name(self) -> str:
    """
    Returns name to use for reverse relation.
    It's either set as `related_name` or by default it's owner model. get_name + 's'
    :return: name of the related_name or default related name.
    :rtype: str
    """
    return self.related_name or self.owner.get_name() + "s"

get_relation_name()

Returns name of the relation, which can be a own name or through model names for m2m models

:return: result of the check :rtype: bool

Source code in ormar/fields/foreign_key.py

def get_relation_name(self) -> str:  # pragma: no cover
    """
    Returns name of the relation, which can be a own name or through model
    names for m2m models

    :return: result of the check
    :rtype: bool
    """
    return self.name

get_source_model()

Returns model from which the relation comes -> either owner or through model

:return: source model :rtype: Type["Model"]

Source code in ormar/fields/foreign_key.py

def get_source_model(self) -> Type["Model"]:  # pragma: no cover
    """
    Returns model from which the relation comes -> either owner or through model

    :return: source model
    :rtype: Type["Model"]
    """
    return self.owner

get_source_related_name()

Returns name to use for source relation name. For FK it's the same, differs for m2m fields. It's either set as related_name or by default it's owner model. get_name + 's' :return: name of the related_name or default related name. :rtype: str

Source code in ormar/fields/foreign_key.py

def get_source_related_name(self) -> str:
    """
    Returns name to use for source relation name.
    For FK it's the same, differs for m2m fields.
    It's either set as `related_name` or by default it's owner model. get_name + 's'
    :return: name of the related_name or default related name.
    :rtype: str
    """
    return self.get_related_name()

has_unresolved_forward_refs()

Verifies if the filed has any ForwardRefs that require updating before the model can be used.

:return: result of the check :rtype: bool

Source code in ormar/fields/foreign_key.py

def has_unresolved_forward_refs(self) -> bool:
    """
    Verifies if the filed has any ForwardRefs that require updating before the
    model can be used.

    :return: result of the check
    :rtype: bool
    """
    return self.to.__class__ == ForwardRef

register_relation(model, child)

Registers relation between parent and child in relation manager. Relation manager is kep on each model (different instance).

Used in Metaclass and sometimes some relations are missing (i.e. cloned Models in fastapi might miss one).

:param model: parent model (with relation definition) :type model: Model class :param child: child model :type child: Model class

Source code in ormar/fields/foreign_key.py

def register_relation(self, model: "Model", child: "Model") -> None:
    """
    Registers relation between parent and child in relation manager.
    Relation manager is kep on each model (different instance).

    Used in Metaclass and sometimes some relations are missing
    (i.e. cloned Models in fastapi might miss one).

    :param model: parent model (with relation definition)
    :type model: Model class
    :param child: child model
    :type child: Model class
    """
    model._orm.add(parent=model, child=child, field=self)

Integer

Bases: ModelFieldFactory, int

Integer field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class Integer(ModelFieldFactory, int):
    """
    Integer field factory that construct Field classes and populated their values.
    """

    _type = int
    _sample = 0

    def __new__(  # type: ignore
        cls,
        *,
        minimum: Optional[int] = None,
        maximum: Optional[int] = None,
        multiple_of: Optional[int] = None,
        **kwargs: Any
    ) -> BaseField:
        autoincrement = kwargs.pop("autoincrement", None)
        autoincrement = (
            autoincrement
            if autoincrement is not None
            else kwargs.get("primary_key", False)
        )
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        kwargs["ge"] = kwargs["minimum"]
        kwargs["le"] = kwargs["maximum"]
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.Integer()

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.Integer()

JSON

Bases: ModelFieldFactory, Json

JSON field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class JSON(ModelFieldFactory, pydantic.Json):
    """
    JSON field factory that construct Field classes and populated their values.
    """

    _type = pydantic.Json
    _sample = '{"json": "json"}'

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.JSON(none_as_null=kwargs.get("sql_nullable", False))

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.JSON(none_as_null=kwargs.get("sql_nullable", False))

LargeBinary

Bases: ModelFieldFactory, bytes

LargeBinary field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class LargeBinary(ModelFieldFactory, bytes):
    """
    LargeBinary field factory that construct Field classes
    and populated their values.
    """

    _type = bytes
    _sample = "bytes"

    def __new__(  # type: ignore # noqa CFQ002
        cls,
        *,
        max_length: int,
        represent_as_base64_str: bool = False,
        **kwargs: Any
    ) -> BaseField:  # type: ignore
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.LargeBinary(length=kwargs.get("max_length"))

    @classmethod
    def validate(cls, **kwargs: Any) -> None:
        """
        Used to validate if all required parameters on a given field type are set.
        :param kwargs: all params passed during construction
        :type kwargs: Any
        """
        max_length = kwargs.get("max_length", None)
        if max_length <= 0:
            raise ModelDefinitionError(
                "Parameter max_length is required for field LargeBinary"
            )

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.LargeBinary(length=kwargs.get("max_length"))

validate(**kwargs) classmethod

Used to validate if all required parameters on a given field type are set. :param kwargs: all params passed during construction :type kwargs: Any

Source code in ormar/fields/model_fields.py

@classmethod
def validate(cls, **kwargs: Any) -> None:
    """
    Used to validate if all required parameters on a given field type are set.
    :param kwargs: all params passed during construction
    :type kwargs: Any
    """
    max_length = kwargs.get("max_length", None)
    if max_length <= 0:
        raise ModelDefinitionError(
            "Parameter max_length is required for field LargeBinary"
        )

ManyToManyField

Bases: ForeignKeyField, QuerySetProtocol, RelationProtocol

Actual class returned from ManyToMany function call and stored in model_fields.

Source code in ormar/fields/many_to_many.py

class ManyToManyField(  # type: ignore
    ForeignKeyField,
    ormar.QuerySetProtocol,
    ormar.RelationProtocol,
):
    """
    Actual class returned from ManyToMany function call and stored in model_fields.
    """

    def __init__(self, **kwargs: Any) -> None:
        if TYPE_CHECKING:  # pragma: no cover
            self.__type__: type
            self.to: Type["Model"]
            self.through: Type["Model"]
        super().__init__(**kwargs)

    def get_source_related_name(self) -> str:
        """
        Returns name to use for source relation name.
        For FK it's the same, differs for m2m fields.
        It's either set as `related_name` or by default it's field name.
        :return: name of the related_name or default related name.
        :rtype: str
        """
        return (
            self.through.ormar_config.model_fields[
                self.default_source_field_name()
            ].related_name
            or self.name
        )

    def has_unresolved_forward_refs(self) -> bool:
        """
        Verifies if the filed has any ForwardRefs that require updating before the
        model can be used.

        :return: result of the check
        :rtype: bool
        """
        return self.to.__class__ == ForwardRef or self.through.__class__ == ForwardRef

    def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None:
        """
        Evaluates the ForwardRef to actual Field based on global and local namespaces

        :param globalns: global namespace
        :type globalns: Any
        :param localns: local namespace
        :type localns: Any
        :return: None
        :rtype: None
        """
        if self.to.__class__ == ForwardRef:
            self._evaluate_forward_ref(globalns, localns)
            (
                self.__type__,
                self.column_type,
                pk_only_model,
            ) = populate_m2m_params_based_on_to_model(
                to=self.to, nullable=self.nullable
            )
            self.to_pk_only = pk_only_model

        if self.through.__class__ == ForwardRef:
            self._evaluate_forward_ref(globalns, localns, is_through=True)

            forbid_through_relations(self.through)

    def get_relation_name(self) -> str:
        """
        Returns name of the relation, which can be a own name or through model
        names for m2m models

        :return: result of the check
        :rtype: bool
        """
        if self.self_reference and self.name == self.self_reference_primary:
            return self.default_source_field_name()
        return self.default_target_field_name()

    def get_source_model(self) -> Type["Model"]:
        """
        Returns model from which the relation comes -> either owner or through model

        :return: source model
        :rtype: Type["Model"]
        """
        return self.through

    def get_filter_clause_target(self) -> Type["Model"]:
        return self.through

    def get_model_relation_fields(self, use_alias: bool = False) -> str:
        """
        Extract names of the database columns or model fields that are connected
        with given relation based on use_alias switch.

        :param use_alias: use db names aliases or model fields
        :type use_alias: bool
        :return: name or names of the related columns/ fields
        :rtype: Union[str, List[str]]
        """
        pk_field = self.owner.ormar_config.model_fields[self.owner.ormar_config.pkname]
        result = pk_field.get_alias() if use_alias else pk_field.name
        return result

    def get_related_field_alias(self) -> str:
        """
        Extract names of the related database columns or that are connected
        with given relation based to use as a target in filter clause.

        :return: name or names of the related columns/ fields
        :rtype: Union[str, Dict[str, str]]
        """
        if self.self_reference and self.self_reference_primary == self.name:
            field_name = self.default_target_field_name()
        else:
            field_name = self.default_source_field_name()
        sub_field = self.through.ormar_config.model_fields[field_name]
        return sub_field.get_alias()

    def get_related_field_name(self) -> Union[str, List[str]]:
        """
        Returns name of the relation field that should be used in prefetch query.
        This field is later used to register relation in prefetch query,
        populate relations dict, and populate nested model in prefetch query.

        :return: name(s) of the field
        :rtype: Union[str, List[str]]
        """
        if self.self_reference and self.self_reference_primary == self.name:
            return self.default_target_field_name()
        return self.default_source_field_name()

    def create_default_through_model(self) -> None:
        """
        Creates default empty through model if no additional fields are required.
        """
        owner_name = self.owner.get_name(lower=False)
        to_name = self.to.get_name(lower=False)
        class_name = f"{owner_name}{to_name}"
        table_name = f"{owner_name.lower()}s_{to_name.lower()}s"
        base_namespace = {
            "__module__": self.owner.__module__,
            "__qualname__": f"{self.owner.__qualname__}.{class_name}",
        }
        new_config = ormar.models.ormar_config.OrmarConfig(
            tablename=table_name,
            database=self.owner.ormar_config.database,
            metadata=self.owner.ormar_config.metadata,
        )
        through_model = type(
            class_name,
            (ormar.Model,),
            {
                **base_namespace,
                "ormar_config": new_config,
                "id": ormar.Integer(name="id", primary_key=True),
            },
        )
        self.through = cast(Type["Model"], through_model)

create_default_through_model()

Creates default empty through model if no additional fields are required.

Source code in ormar/fields/many_to_many.py

def create_default_through_model(self) -> None:
    """
    Creates default empty through model if no additional fields are required.
    """
    owner_name = self.owner.get_name(lower=False)
    to_name = self.to.get_name(lower=False)
    class_name = f"{owner_name}{to_name}"
    table_name = f"{owner_name.lower()}s_{to_name.lower()}s"
    base_namespace = {
        "__module__": self.owner.__module__,
        "__qualname__": f"{self.owner.__qualname__}.{class_name}",
    }
    new_config = ormar.models.ormar_config.OrmarConfig(
        tablename=table_name,
        database=self.owner.ormar_config.database,
        metadata=self.owner.ormar_config.metadata,
    )
    through_model = type(
        class_name,
        (ormar.Model,),
        {
            **base_namespace,
            "ormar_config": new_config,
            "id": ormar.Integer(name="id", primary_key=True),
        },
    )
    self.through = cast(Type["Model"], through_model)

evaluate_forward_ref(globalns, localns)

Evaluates the ForwardRef to actual Field based on global and local namespaces

:param globalns: global namespace :type globalns: Any :param localns: local namespace :type localns: Any :return: None :rtype: None

Source code in ormar/fields/many_to_many.py

def evaluate_forward_ref(self, globalns: Any, localns: Any) -> None:
    """
    Evaluates the ForwardRef to actual Field based on global and local namespaces

    :param globalns: global namespace
    :type globalns: Any
    :param localns: local namespace
    :type localns: Any
    :return: None
    :rtype: None
    """
    if self.to.__class__ == ForwardRef:
        self._evaluate_forward_ref(globalns, localns)
        (
            self.__type__,
            self.column_type,
            pk_only_model,
        ) = populate_m2m_params_based_on_to_model(
            to=self.to, nullable=self.nullable
        )
        self.to_pk_only = pk_only_model

    if self.through.__class__ == ForwardRef:
        self._evaluate_forward_ref(globalns, localns, is_through=True)

        forbid_through_relations(self.through)

get_model_relation_fields(use_alias=False)

Extract names of the database columns or model fields that are connected with given relation based on use_alias switch.

:param use_alias: use db names aliases or model fields :type use_alias: bool :return: name or names of the related columns/ fields :rtype: Union[str, List[str]]

Source code in ormar/fields/many_to_many.py

def get_model_relation_fields(self, use_alias: bool = False) -> str:
    """
    Extract names of the database columns or model fields that are connected
    with given relation based on use_alias switch.

    :param use_alias: use db names aliases or model fields
    :type use_alias: bool
    :return: name or names of the related columns/ fields
    :rtype: Union[str, List[str]]
    """
    pk_field = self.owner.ormar_config.model_fields[self.owner.ormar_config.pkname]
    result = pk_field.get_alias() if use_alias else pk_field.name
    return result

get_related_field_alias()

Extract names of the related database columns or that are connected with given relation based to use as a target in filter clause.

:return: name or names of the related columns/ fields :rtype: Union[str, Dict[str, str]]

Source code in ormar/fields/many_to_many.py

def get_related_field_alias(self) -> str:
    """
    Extract names of the related database columns or that are connected
    with given relation based to use as a target in filter clause.

    :return: name or names of the related columns/ fields
    :rtype: Union[str, Dict[str, str]]
    """
    if self.self_reference and self.self_reference_primary == self.name:
        field_name = self.default_target_field_name()
    else:
        field_name = self.default_source_field_name()
    sub_field = self.through.ormar_config.model_fields[field_name]
    return sub_field.get_alias()

get_related_field_name()

Returns name of the relation field that should be used in prefetch query. This field is later used to register relation in prefetch query, populate relations dict, and populate nested model in prefetch query.

:return: name(s) of the field :rtype: Union[str, List[str]]

Source code in ormar/fields/many_to_many.py

def get_related_field_name(self) -> Union[str, List[str]]:
    """
    Returns name of the relation field that should be used in prefetch query.
    This field is later used to register relation in prefetch query,
    populate relations dict, and populate nested model in prefetch query.

    :return: name(s) of the field
    :rtype: Union[str, List[str]]
    """
    if self.self_reference and self.self_reference_primary == self.name:
        return self.default_target_field_name()
    return self.default_source_field_name()

get_relation_name()

Returns name of the relation, which can be a own name or through model names for m2m models

:return: result of the check :rtype: bool

Source code in ormar/fields/many_to_many.py

def get_relation_name(self) -> str:
    """
    Returns name of the relation, which can be a own name or through model
    names for m2m models

    :return: result of the check
    :rtype: bool
    """
    if self.self_reference and self.name == self.self_reference_primary:
        return self.default_source_field_name()
    return self.default_target_field_name()

get_source_model()

Returns model from which the relation comes -> either owner or through model

:return: source model :rtype: Type["Model"]

Source code in ormar/fields/many_to_many.py

def get_source_model(self) -> Type["Model"]:
    """
    Returns model from which the relation comes -> either owner or through model

    :return: source model
    :rtype: Type["Model"]
    """
    return self.through

get_source_related_name()

Returns name to use for source relation name. For FK it's the same, differs for m2m fields. It's either set as related_name or by default it's field name. :return: name of the related_name or default related name. :rtype: str

Source code in ormar/fields/many_to_many.py

def get_source_related_name(self) -> str:
    """
    Returns name to use for source relation name.
    For FK it's the same, differs for m2m fields.
    It's either set as `related_name` or by default it's field name.
    :return: name of the related_name or default related name.
    :rtype: str
    """
    return (
        self.through.ormar_config.model_fields[
            self.default_source_field_name()
        ].related_name
        or self.name
    )

has_unresolved_forward_refs()

Verifies if the filed has any ForwardRefs that require updating before the model can be used.

:return: result of the check :rtype: bool

Source code in ormar/fields/many_to_many.py

def has_unresolved_forward_refs(self) -> bool:
    """
    Verifies if the filed has any ForwardRefs that require updating before the
    model can be used.

    :return: result of the check
    :rtype: bool
    """
    return self.to.__class__ == ForwardRef or self.through.__class__ == ForwardRef

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.

:return: number of deleted rows (for some backends) :rtype: int

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.

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

:return: reloaded Model :rtype: Model

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.

: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

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.

:return: saved Model :rtype: 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.

: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

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.

: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

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.

:param kwargs: list of fields to update :type kwargs: Any :return: saved Model :rtype: 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)

ModelDefinitionError

Bases: AsyncOrmException

Raised for errors related to the model definition itself:

• defining a Field without required parameters
• defining a model with more than one primary_key
• defining a model without primary_key

Source code in ormar/exceptions.py

class ModelDefinitionError(AsyncOrmException):
    """
    Raised for errors related to the model definition itself:

    * defining a Field without required parameters
    * defining a model with more than one primary_key
    * defining a model without primary_key
    """

    pass

MultipleMatches

Bases: AsyncOrmException

Raised for database queries that should return one row (i.e. get, first etc.) but has multiple matching results in response.

Source code in ormar/exceptions.py

class MultipleMatches(AsyncOrmException):
    """
    Raised for database queries that should return one row (i.e. get, first etc.)
    but has multiple matching results in response.
    """

    pass

NoMatch

Bases: AsyncOrmException

Raised for database queries that has no matching result (empty result).

Source code in ormar/exceptions.py

class NoMatch(AsyncOrmException):
    """
    Raised for database queries that has no matching result (empty result).
    """

    pass

OrderAction

Bases: QueryAction

Order Actions is populated by queryset when order_by() is called.

All required params are extracted but kept raw until actual filter clause value is required -> then the action is converted into text() clause.

Extracted in order to easily change table prefixes on complex relations.

Source code in ormar/queryset/actions/order_action.py

class OrderAction(QueryAction):
    """
    Order Actions is populated by queryset when order_by() is called.

    All required params are extracted but kept raw until actual filter clause value
    is required -> then the action is converted into text() clause.

    Extracted in order to easily change table prefixes on complex relations.
    """

    def __init__(
        self, order_str: str, model_cls: Type["Model"], alias: Optional[str] = None
    ) -> None:
        self.direction: str = ""
        super().__init__(query_str=order_str, model_cls=model_cls)
        self.is_source_model_order = False
        if alias:
            self.table_prefix = alias
        if self.source_model == self.target_model and "__" not in self.related_str:
            self.is_source_model_order = True

    @property
    def field_alias(self) -> str:
        return self.target_model.get_column_alias(self.field_name)

    @property
    def is_postgres_bool(self) -> bool:
        dialect = self.target_model.ormar_config.database._backend._dialect.name
        field_type = self.target_model.ormar_config.model_fields[
            self.field_name
        ].__type__
        return dialect == "postgresql" and field_type is bool

    def get_field_name_text(self) -> str:
        """
        Escapes characters if it's required.
        Substitutes values of the models if value is a ormar Model with its pk value.
        Compiles the clause.

        :return: complied and escaped clause
        :rtype: sqlalchemy.sql.elements.TextClause
        """
        prefix = f"{self.table_prefix}_" if self.table_prefix else ""
        return f"{prefix}{self.table}" f".{self.field_alias}"

    def get_min_or_max(self) -> sqlalchemy.sql.expression.TextClause:
        """
        Used in limit sub queries where you need to use aggregated functions
        in order to order by columns not included in group by. For postgres bool
        field it's using bool_or function as aggregates does not work with this type
        of columns.

        :return: min or max function to order
        :rtype: sqlalchemy.sql.elements.TextClause
        """
        prefix = f"{self.table_prefix}_" if self.table_prefix else ""
        if self.direction == "":
            function = "min" if not self.is_postgres_bool else "bool_or"
            return text(f"{function}({prefix}{self.table}" f".{self.field_alias})")
        function = "max" if not self.is_postgres_bool else "bool_or"
        return text(f"{function}({prefix}{self.table}" f".{self.field_alias}) desc")

    def get_text_clause(self) -> sqlalchemy.sql.expression.TextClause:
        """
        Escapes characters if it's required.
        Substitutes values of the models if value is a ormar Model with its pk value.
        Compiles the clause.

        :return: complied and escaped clause
        :rtype: sqlalchemy.sql.elements.TextClause
        """
        dialect = self.target_model.ormar_config.database._backend._dialect
        quoter = dialect.identifier_preparer.quote
        prefix = f"{self.table_prefix}_" if self.table_prefix else ""
        table_name = self.table.name
        field_name = self.field_alias
        if not prefix:
            table_name = quoter(table_name)
        else:
            table_name = quoter(f"{prefix}{table_name}")
        field_name = quoter(field_name)
        return text(f"{table_name}.{field_name} {self.direction}")

    def _split_value_into_parts(self, order_str: str) -> None:
        if order_str.startswith("-"):
            self.direction = "desc"
            order_str = order_str[1:]
        parts = order_str.split("__")
        self.field_name = parts[-1]
        self.related_parts = parts[:-1]

    def check_if_filter_apply(self, target_model: Type["Model"], alias: str) -> bool:
        """
        Checks filter conditions to find if they apply to current join.

        :param target_model: model which is now processed
        :type target_model: Type["Model"]
        :param alias: prefix of the relation
        :type alias: str
        :return: result of the check
        :rtype: bool
        """
        return target_model == self.target_model and alias == self.table_prefix

check_if_filter_apply(target_model, alias)

Checks filter conditions to find if they apply to current join.

:param target_model: model which is now processed :type target_model: Type["Model"] :param alias: prefix of the relation :type alias: str :return: result of the check :rtype: bool

Source code in ormar/queryset/actions/order_action.py

def check_if_filter_apply(self, target_model: Type["Model"], alias: str) -> bool:
    """
    Checks filter conditions to find if they apply to current join.

    :param target_model: model which is now processed
    :type target_model: Type["Model"]
    :param alias: prefix of the relation
    :type alias: str
    :return: result of the check
    :rtype: bool
    """
    return target_model == self.target_model and alias == self.table_prefix

get_field_name_text()

Escapes characters if it's required. Substitutes values of the models if value is a ormar Model with its pk value. Compiles the clause.

:return: complied and escaped clause :rtype: sqlalchemy.sql.elements.TextClause

Source code in ormar/queryset/actions/order_action.py

def get_field_name_text(self) -> str:
    """
    Escapes characters if it's required.
    Substitutes values of the models if value is a ormar Model with its pk value.
    Compiles the clause.

    :return: complied and escaped clause
    :rtype: sqlalchemy.sql.elements.TextClause
    """
    prefix = f"{self.table_prefix}_" if self.table_prefix else ""
    return f"{prefix}{self.table}" f".{self.field_alias}"

get_min_or_max()

Used in limit sub queries where you need to use aggregated functions in order to order by columns not included in group by. For postgres bool field it's using bool_or function as aggregates does not work with this type of columns.

:return: min or max function to order :rtype: sqlalchemy.sql.elements.TextClause

Source code in ormar/queryset/actions/order_action.py

def get_min_or_max(self) -> sqlalchemy.sql.expression.TextClause:
    """
    Used in limit sub queries where you need to use aggregated functions
    in order to order by columns not included in group by. For postgres bool
    field it's using bool_or function as aggregates does not work with this type
    of columns.

    :return: min or max function to order
    :rtype: sqlalchemy.sql.elements.TextClause
    """
    prefix = f"{self.table_prefix}_" if self.table_prefix else ""
    if self.direction == "":
        function = "min" if not self.is_postgres_bool else "bool_or"
        return text(f"{function}({prefix}{self.table}" f".{self.field_alias})")
    function = "max" if not self.is_postgres_bool else "bool_or"
    return text(f"{function}({prefix}{self.table}" f".{self.field_alias}) desc")

get_text_clause()

Escapes characters if it's required. Substitutes values of the models if value is a ormar Model with its pk value. Compiles the clause.

:return: complied and escaped clause :rtype: sqlalchemy.sql.elements.TextClause

Source code in ormar/queryset/actions/order_action.py

def get_text_clause(self) -> sqlalchemy.sql.expression.TextClause:
    """
    Escapes characters if it's required.
    Substitutes values of the models if value is a ormar Model with its pk value.
    Compiles the clause.

    :return: complied and escaped clause
    :rtype: sqlalchemy.sql.elements.TextClause
    """
    dialect = self.target_model.ormar_config.database._backend._dialect
    quoter = dialect.identifier_preparer.quote
    prefix = f"{self.table_prefix}_" if self.table_prefix else ""
    table_name = self.table.name
    field_name = self.field_alias
    if not prefix:
        table_name = quoter(table_name)
    else:
        table_name = quoter(f"{prefix}{table_name}")
    field_name = quoter(field_name)
    return text(f"{table_name}.{field_name} {self.direction}")

QuerySet

Bases: Generic[T]

Main class to perform database queries, exposed on each model as objects attribute.

Source code in ormar/queryset/queryset.py

class QuerySet(Generic[T]):
    """
    Main class to perform database queries, exposed on each model as objects attribute.
    """

    def __init__(  # noqa CFQ002
        self,
        model_cls: Optional[Type["T"]] = None,
        filter_clauses: Optional[List] = None,
        exclude_clauses: Optional[List] = None,
        select_related: Optional[List] = None,
        limit_count: Optional[int] = None,
        offset: Optional[int] = None,
        excludable: Optional["ExcludableItems"] = None,
        order_bys: Optional[List] = None,
        prefetch_related: Optional[List] = None,
        limit_raw_sql: bool = False,
        proxy_source_model: Optional[Type["Model"]] = None,
    ) -> None:
        self.proxy_source_model = proxy_source_model
        self.model_cls = model_cls
        self.filter_clauses = [] if filter_clauses is None else filter_clauses
        self.exclude_clauses = [] if exclude_clauses is None else exclude_clauses
        self._select_related = [] if select_related is None else select_related
        self._prefetch_related = [] if prefetch_related is None else prefetch_related
        self.limit_count = limit_count
        self.query_offset = offset
        self._excludable = excludable or ormar.ExcludableItems()
        self.order_bys = order_bys or []
        self.limit_sql_raw = limit_raw_sql

    @property
    def model_config(self) -> "OrmarConfig":
        """
        Shortcut to model class OrmarConfig set on QuerySet model.

        :return: OrmarConfig of the model
        :rtype: model's OrmarConfig
        """
        if not self.model_cls:  # pragma nocover
            raise ValueError("Model class of QuerySet is not initialized")
        return self.model_cls.ormar_config

    @property
    def model(self) -> Type["T"]:
        """
        Shortcut to model class set on QuerySet.

        :return: model class
        :rtype: Type[Model]
        """
        if not self.model_cls:  # pragma nocover
            raise ValueError("Model class of QuerySet is not initialized")
        return self.model_cls

    def rebuild_self(  # noqa: CFQ002
        self,
        filter_clauses: Optional[List] = None,
        exclude_clauses: Optional[List] = None,
        select_related: Optional[List] = None,
        limit_count: Optional[int] = None,
        offset: Optional[int] = None,
        excludable: Optional["ExcludableItems"] = None,
        order_bys: Optional[List] = None,
        prefetch_related: Optional[List] = None,
        limit_raw_sql: Optional[bool] = None,
        proxy_source_model: Optional[Type["Model"]] = None,
    ) -> "QuerySet":
        """
        Method that returns new instance of queryset based on passed params,
        all not passed params are taken from current values.
        """
        overwrites = {
            "select_related": "_select_related",
            "offset": "query_offset",
            "excludable": "_excludable",
            "prefetch_related": "_prefetch_related",
            "limit_raw_sql": "limit_sql_raw",
        }
        passed_args = locals()

        def replace_if_none(arg_name: str) -> Any:
            if passed_args.get(arg_name) is None:
                return getattr(self, overwrites.get(arg_name, arg_name))
            return passed_args.get(arg_name)

        return self.__class__(
            model_cls=self.model_cls,
            filter_clauses=replace_if_none("filter_clauses"),
            exclude_clauses=replace_if_none("exclude_clauses"),
            select_related=replace_if_none("select_related"),
            limit_count=replace_if_none("limit_count"),
            offset=replace_if_none("offset"),
            excludable=replace_if_none("excludable"),
            order_bys=replace_if_none("order_bys"),
            prefetch_related=replace_if_none("prefetch_related"),
            limit_raw_sql=replace_if_none("limit_raw_sql"),
            proxy_source_model=replace_if_none("proxy_source_model"),
        )

    async def _prefetch_related_models(
        self, models: List["T"], rows: List
    ) -> List["T"]:
        """
        Performs prefetch query for selected models names.

        :param models: list of already parsed main Models from main query
        :type models: List[Model]
        :param rows: database rows from main query
        :type rows: List[sqlalchemy.engine.result.RowProxy]
        :return: list of models with prefetch models populated
        :rtype: List[Model]
        """
        query = PrefetchQuery(
            model_cls=self.model,
            excludable=self._excludable,
            prefetch_related=self._prefetch_related,
            select_related=self._select_related,
            orders_by=self.order_bys,
        )
        return await query.prefetch_related(models=models)  # type: ignore

    async def _process_query_result_rows(self, rows: List) -> List["T"]:
        """
        Process database rows and initialize ormar Model from each of the rows.

        :param rows: list of database rows from query result
        :type rows: List[sqlalchemy.engine.result.RowProxy]
        :return: list of models
        :rtype: List[Model]
        """
        result_rows = []
        for row in rows:
            result_rows.append(
                self.model.from_row(
                    row=row,
                    select_related=self._select_related,
                    excludable=self._excludable,
                    source_model=self.model,
                    proxy_source_model=self.proxy_source_model,
                )
            )
            await asyncio.sleep(0)

        if result_rows:
            return self.model.merge_instances_list(result_rows)  # type: ignore
        return cast(List["T"], result_rows)

    def _resolve_filter_groups(
        self, groups: Any
    ) -> Tuple[List[FilterGroup], List[str]]:
        """
        Resolves filter groups to populate FilterAction params in group tree.

        :param groups: tuple of FilterGroups
        :type groups: Any
        :return: list of resolver groups
        :rtype: Tuple[List[FilterGroup], List[str]]
        """
        filter_groups = []
        select_related = self._select_related
        if groups:
            for group in groups:
                if not isinstance(group, FilterGroup):
                    raise QueryDefinitionError(
                        "Only ormar.and_ and ormar.or_ "
                        "can be passed as filter positional"
                        " arguments,"
                        "other values need to be passed by"
                        "keyword arguments"
                    )
                _, select_related = group.resolve(
                    model_cls=self.model,
                    select_related=self._select_related,
                    filter_clauses=self.filter_clauses,
                )
                filter_groups.append(group)
        return filter_groups, select_related

    @staticmethod
    def check_single_result_rows_count(rows: Sequence[Optional["T"]]) -> None:
        """
        Verifies if the result has one and only one row.

        :param rows: one element list of Models
        :type rows: List[Model]
        """
        if not rows or rows[0] is None:
            raise NoMatch()
        if len(rows) > 1:
            raise MultipleMatches()

    @property
    def database(self) -> databases.Database:
        """
        Shortcut to models database from OrmarConfig class.

        :return: database
        :rtype: databases.Database
        """
        return self.model_config.database

    @property
    def table(self) -> sqlalchemy.Table:
        """
        Shortcut to models table from OrmarConfig.

        :return: database table
        :rtype: sqlalchemy.Table
        """
        return self.model_config.table

    def build_select_expression(
        self,
        limit: Optional[int] = None,
        offset: Optional[int] = None,
        order_bys: Optional[List] = None,
    ) -> sqlalchemy.sql.select:
        """
        Constructs the actual database query used in the QuerySet.
        If any of the params is not passed the QuerySet own value is used.

        :param limit: number to limit the query
        :type limit: int
        :param offset: number to offset by
        :type offset: int
        :param order_bys: list of order-by fields names
        :type order_bys: List
        :return: built sqlalchemy select expression
        :rtype: sqlalchemy.sql.selectable.Select
        """
        qry = Query(
            model_cls=self.model,
            select_related=self._select_related,
            filter_clauses=self.filter_clauses,
            exclude_clauses=self.exclude_clauses,
            offset=offset or self.query_offset,
            excludable=self._excludable,
            order_bys=order_bys or self.order_bys,
            limit_raw_sql=self.limit_sql_raw,
            limit_count=limit if limit is not None else self.limit_count,
        )
        exp = qry.build_select_expression()
        # print("\n", exp.compile(compile_kwargs={"literal_binds": True}))
        return exp

    def filter(  # noqa: A003
        self, *args: Any, _exclude: bool = False, **kwargs: Any
    ) -> "QuerySet[T]":
        """
        Allows you to filter by any `Model` attribute/field
        as well as to fetch instances, with a filter across an FK relationship.

        You can use special filter suffix to change the filter operands:

        *  exact - like `album__name__exact='Malibu'` (exact match)
        *  iexact - like `album__name__iexact='malibu'` (exact match case insensitive)
        *  contains - like `album__name__contains='Mal'` (sql like)
        *  icontains - like `album__name__icontains='mal'` (sql like case insensitive)
        *  in - like `album__name__in=['Malibu', 'Barclay']` (sql in)
        *  isnull - like `album__name__isnull=True` (sql is null)
           (isnotnull `album__name__isnull=False` (sql is not null))
        *  gt - like `position__gt=3` (sql >)
        *  gte - like `position__gte=3` (sql >=)
        *  lt - like `position__lt=3` (sql <)
        *  lte - like `position__lte=3` (sql <=)
        *  startswith - like `album__name__startswith='Mal'` (exact start match)
        *  istartswith - like `album__name__istartswith='mal'` (case insensitive)
        *  endswith - like `album__name__endswith='ibu'` (exact end match)
        *  iendswith - like `album__name__iendswith='IBU'` (case insensitive)

        Note that you can also use python style filters - check the docs!

        :param _exclude: flag if it should be exclude or filter
        :type _exclude: bool
        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: filtered QuerySet
        :rtype: QuerySet
        """
        filter_groups, select_related = self._resolve_filter_groups(groups=args)
        qryclause = QueryClause(
            model_cls=self.model,
            select_related=select_related,
            filter_clauses=self.filter_clauses,
        )
        filter_clauses, select_related = qryclause.prepare_filter(**kwargs)
        filter_clauses = filter_clauses + filter_groups  # type: ignore
        if _exclude:
            exclude_clauses = filter_clauses
            filter_clauses = self.filter_clauses
        else:
            exclude_clauses = self.exclude_clauses
            filter_clauses = filter_clauses

        return self.rebuild_self(
            filter_clauses=filter_clauses,
            exclude_clauses=exclude_clauses,
            select_related=select_related,
        )

    def exclude(self, *args: Any, **kwargs: Any) -> "QuerySet[T]":  # noqa: A003
        """
        Works exactly the same as filter and all modifiers (suffixes) are the same,
        but returns a *not* condition.

        So if you use `filter(name='John')` which is `where name = 'John'` in SQL,
        the `exclude(name='John')` equals to `where name <> 'John'`

        Note that all conditions are joined so if you pass multiple values it
        becomes a union of conditions.

        `exclude(name='John', age>=35)` will become
        `where not (name='John' and age>=35)`

        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: filtered QuerySet
        :rtype: QuerySet
        """
        return self.filter(_exclude=True, *args, **kwargs)

    def select_related(self, related: Union[List, str, FieldAccessor]) -> "QuerySet[T]":
        """
        Allows to prefetch related models during the same query.

        **With `select_related` always only one query is run against the database**,
        meaning that one (sometimes complicated) join is generated and later nested
        models are processed in python.

        To fetch related model use `ForeignKey` names.

        To chain related `Models` relation use double underscores between names.

        :param related: list of relation field names, can be linked by '__' to nest
        :type related: Union[List, str]
        :return: QuerySet
        :rtype: QuerySet
        """
        if not isinstance(related, list):
            related = [related]
        related = [
            rel._access_chain if isinstance(rel, FieldAccessor) else rel
            for rel in related
        ]

        related = sorted(list(set(list(self._select_related) + related)))
        return self.rebuild_self(select_related=related)

    def select_all(self, follow: bool = False) -> "QuerySet[T]":
        """
        By default adds only directly related models.

        If follow=True is set it adds 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 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.model.extract_related_names())
        if follow:
            relations = self.model._iterate_related_models()
        return self.rebuild_self(select_related=relations)

    def prefetch_related(
        self, related: Union[List, str, FieldAccessor]
    ) -> "QuerySet[T]":
        """
        Allows to prefetch related models during query - but opposite to
        `select_related` each subsequent model is fetched in a separate database query.

        **With `prefetch_related` always one query per Model is run against the
        database**, meaning that you will have multiple queries executed one
        after another.

        To fetch related model use `ForeignKey` names.

        To chain related `Models` relation use double underscores between names.

        :param related: list of relation field names, can be linked by '__' to nest
        :type related: Union[List, str]
        :return: QuerySet
        :rtype: QuerySet
        """
        if not isinstance(related, list):
            related = [related]
        related = [
            rel._access_chain if isinstance(rel, FieldAccessor) else rel
            for rel in related
        ]

        related = list(set(list(self._prefetch_related) + related))
        return self.rebuild_self(prefetch_related=related)

    def fields(
        self, columns: Union[List, str, Set, Dict], _is_exclude: bool = False
    ) -> "QuerySet[T]":
        """
        With `fields()` you can select subset of model columns to limit the data load.

        Note that `fields()` and `exclude_fields()` works both for main models
        (on normal queries like `get`, `all` etc.)
        as well as `select_related` and `prefetch_related`
        models (with nested notation).

        You can select specified fields by passing a `str, List[str], Set[str] or
        dict` with nested definition.

        To include related models use notation
        `{related_name}__{column}[__{optional_next} etc.]`.

        `fields()` can be called several times, building up the columns to select.

        If you include related models into `select_related()` call but you won't specify
        columns for those models in fields - implies a list of all fields for
        those nested models.

        Mandatory fields cannot be excluded as it will raise `ValidationError`,
        to exclude a field it has to be nullable.

        Pk column cannot be excluded - it's always auto added even if
        not explicitly included.

        You can also pass fields to include as dictionary or set.

        To mark a field as included in a dictionary use it's name as key
        and ellipsis as value.

        To traverse nested models use nested dictionaries.

        To include fields at last level instead of nested dictionary a set can be used.

        To include whole nested model specify model related field name and ellipsis.

        :param _is_exclude: flag if it's exclude or include operation
        :type _is_exclude: bool
        :param columns: columns to include
        :type columns: Union[List, str, Set, Dict]
        :return: QuerySet
        :rtype: QuerySet
        """
        excludable = ormar.ExcludableItems.from_excludable(self._excludable)
        excludable.build(
            items=columns,
            model_cls=self.model_cls,  # type: ignore
            is_exclude=_is_exclude,
        )

        return self.rebuild_self(excludable=excludable)

    def exclude_fields(self, columns: Union[List, str, Set, Dict]) -> "QuerySet[T]":
        """
        With `exclude_fields()` you can select subset of model columns that will
        be excluded to limit the data load.

        It's the opposite of `fields()` method so check documentation above
        to see what options are available.

        Especially check above how you can pass also nested dictionaries
        and sets as a mask to exclude fields from whole hierarchy.

        Note that `fields()` and `exclude_fields()` works both for main models
        (on normal queries like `get`, `all` etc.)
        as well as `select_related` and `prefetch_related` models
        (with nested notation).

        Mandatory fields cannot be excluded as it will raise `ValidationError`,
        to exclude a field it has to be nullable.

        Pk column cannot be excluded - it's always auto added even
        if explicitly excluded.

        :param columns: columns to exclude
        :type columns: Union[List, str, Set, Dict]
        :return: QuerySet
        :rtype: QuerySet
        """
        return self.fields(columns=columns, _is_exclude=True)

    def order_by(self, columns: Union[List, str, OrderAction]) -> "QuerySet[T]":
        """
        With `order_by()` you can order the results from database based on your
        choice of fields.

        You can provide a string with field name or list of strings with fields names.

        Ordering in sql will be applied in order of names you provide in order_by.

        By default if you do not provide ordering `ormar` explicitly orders by
        all primary keys

        If you are sorting by nested models that causes that the result rows are
        unsorted by the main model `ormar` will combine those children rows into
        one main model.

        The main model will never duplicate in the result

        To order by main model field just provide a field name

        To sort on nested models separate field names with dunder '__'.

        You can sort this way across all relation types -> `ForeignKey`,
        reverse virtual FK and `ManyToMany` fields.

        To sort in descending order provide a hyphen in front of the field name

        :param columns: columns by which models should be sorted
        :type columns: Union[List, str]
        :return: QuerySet
        :rtype: QuerySet
        """
        if not isinstance(columns, list):
            columns = [columns]

        orders_by = [
            (
                OrderAction(order_str=x, model_cls=self.model_cls)  # type: ignore
                if not isinstance(x, OrderAction)
                else x
            )
            for x in columns
        ]

        order_bys = self.order_bys + [x for x in orders_by if x not in self.order_bys]
        return self.rebuild_self(order_bys=order_bys)

    async def values(
        self,
        fields: Union[List, str, Set, Dict, None] = None,
        exclude_through: bool = False,
        _as_dict: bool = True,
        _flatten: bool = False,
    ) -> List:
        """
        Return a list of dictionaries with column values in order of the fields
        passed or all fields from queried models.

        To filter for given row use filter/exclude methods before values,
        to limit number of rows use limit/offset or paginate before values.

        Note that it always return a list even for one row from database.

        :param exclude_through: flag if through models should be excluded
        :type exclude_through: bool
        :param _flatten: internal parameter to flatten one element tuples
        :type _flatten: bool
        :param _as_dict: internal parameter if return dict or tuples
        :type _as_dict: bool
        :param fields: field name or list of field names to extract from db
        :type fields:  Union[List, str, Set, Dict]
        """
        if fields:
            return await self.fields(columns=fields).values(
                _as_dict=_as_dict, _flatten=_flatten, exclude_through=exclude_through
            )
        expr = self.build_select_expression()
        rows = await self.database.fetch_all(expr)
        if not rows:
            return []
        alias_resolver = ReverseAliasResolver(
            select_related=self._select_related,
            excludable=self._excludable,
            model_cls=self.model_cls,  # type: ignore
            exclude_through=exclude_through,
        )
        column_map = alias_resolver.resolve_columns(
            columns_names=list(cast(LegacyRow, rows[0]).keys())
        )
        result = [
            {column_map.get(k): v for k, v in dict(x).items() if k in column_map}
            for x in rows
        ]
        if _as_dict:
            return result
        if _flatten and self._excludable.include_entry_count() != 1:
            raise QueryDefinitionError(
                "You cannot flatten values_list if more than one field is selected!"
            )
        tuple_result = [tuple(x.values()) for x in result]
        return tuple_result if not _flatten else [x[0] for x in tuple_result]

    async def values_list(
        self,
        fields: Union[List, str, Set, Dict, None] = None,
        flatten: bool = False,
        exclude_through: bool = False,
    ) -> List:
        """
        Return a list of tuples with column values in order of the fields passed or
        all fields from queried models.

        When one field is passed you can flatten the list of tuples into list of values
        of that single field.

        To filter for given row use filter/exclude methods before values,
        to limit number of rows use limit/offset or paginate before values.

        Note that it always return a list even for one row from database.

        :param exclude_through: flag if through models should be excluded
        :type exclude_through: bool
        :param fields: field name or list of field names to extract from db
        :type fields: Union[str, List[str]]
        :param flatten: when one field is passed you can flatten the list of tuples
        :type flatten: bool
        """
        return await self.values(
            fields=fields,
            exclude_through=exclude_through,
            _as_dict=False,
            _flatten=flatten,
        )

    async def exists(self) -> bool:
        """
        Returns a bool value to confirm if there are rows matching the given criteria
        (applied with `filter` and `exclude` if set).

        :return: result of the check
        :rtype: bool
        """
        expr = self.build_select_expression()
        expr = sqlalchemy.exists(expr).select()
        return await self.database.fetch_val(expr)

    async def count(self, distinct: bool = True) -> int:
        """
        Returns number of rows matching the given criteria
        (applied with `filter` and `exclude` if set before).
        If `distinct` is `True` (the default), this will return
        the number of primary rows selected. If `False`,
        the count will be the total number of rows returned
        (including extra rows for `one-to-many` or `many-to-many`
        left `select_related` table joins).
        `False` is the legacy (buggy) behavior for workflows that depend on it.

        :param distinct: flag if the primary table rows should be distinct or not

        :return: number of rows
        :rtype: int
        """
        expr = self.build_select_expression().alias("subquery_for_count")
        expr = sqlalchemy.func.count().select().select_from(expr)
        if distinct:
            pk_column_name = self.model.get_column_alias(self.model_config.pkname)
            expr_distinct = expr.group_by(pk_column_name).alias("subquery_for_group")
            expr = sqlalchemy.func.count().select().select_from(expr_distinct)
        return await self.database.fetch_val(expr)

    async def _query_aggr_function(self, func_name: str, columns: List) -> Any:
        func = getattr(sqlalchemy.func, func_name)
        select_actions = [
            SelectAction(select_str=column, model_cls=self.model) for column in columns
        ]
        if func_name in ["sum", "avg"]:
            if any(not x.is_numeric for x in select_actions):
                raise QueryDefinitionError(
                    "You can use sum and svg only with" "numeric types of columns"
                )
        select_columns = [x.apply_func(func, use_label=True) for x in select_actions]
        expr = self.build_select_expression().alias(f"subquery_for_{func_name}")
        expr = sqlalchemy.select(select_columns).select_from(expr)
        # print("\n", expr.compile(compile_kwargs={"literal_binds": True}))
        result = await self.database.fetch_one(expr)
        return dict(result) if len(result) > 1 else result[0]  # type: ignore

    async def max(self, columns: Union[str, List[str]]) -> Any:  # noqa: A003
        """
        Returns max value of columns for rows matching the given criteria
        (applied with `filter` and `exclude` if set before).

        :return: max value of column(s)
        :rtype: Any
        """
        if not isinstance(columns, list):
            columns = [columns]
        return await self._query_aggr_function(func_name="max", columns=columns)

    async def min(self, columns: Union[str, List[str]]) -> Any:  # noqa: A003
        """
        Returns min value of columns for rows matching the given criteria
        (applied with `filter` and `exclude` if set before).

        :return: min value of column(s)
        :rtype: Any
        """
        if not isinstance(columns, list):
            columns = [columns]
        return await self._query_aggr_function(func_name="min", columns=columns)

    async def sum(self, columns: Union[str, List[str]]) -> Any:  # noqa: A003
        """
        Returns sum value of columns for rows matching the given criteria
        (applied with `filter` and `exclude` if set before).

        :return: sum value of columns
        :rtype: int
        """
        if not isinstance(columns, list):
            columns = [columns]
        return await self._query_aggr_function(func_name="sum", columns=columns)

    async def avg(self, columns: Union[str, List[str]]) -> Any:
        """
        Returns avg value of columns for rows matching the given criteria
        (applied with `filter` and `exclude` if set before).

        :return: avg value of columns
        :rtype: Union[int, float, List]
        """
        if not isinstance(columns, list):
            columns = [columns]
        return await self._query_aggr_function(func_name="avg", columns=columns)

    async def update(self, each: bool = False, **kwargs: Any) -> int:
        """
        Updates the model table after applying the filters from kwargs.

        You have to either pass a filter to narrow down a query or explicitly pass
        each=True flag to affect whole table.

        :param each: flag if whole table should be affected if no filter is passed
        :type each: bool
        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: number of updated rows
        :rtype: int
        """
        if not each and not (self.filter_clauses or self.exclude_clauses):
            raise QueryDefinitionError(
                "You cannot update without filtering the queryset first. "
                "If you want to update all rows use update(each=True, **kwargs)"
            )

        self_fields = self.model.extract_db_own_fields().union(
            self.model.extract_related_names()
        )
        updates = {k: v for k, v in kwargs.items() if k in self_fields}
        updates = self.model.validate_enums(updates)
        updates = self.model.translate_columns_to_aliases(updates)

        expr = FilterQuery(filter_clauses=self.filter_clauses).apply(
            self.table.update().values(**updates)
        )
        expr = FilterQuery(filter_clauses=self.exclude_clauses, exclude=True).apply(
            expr
        )
        return await self.database.execute(expr)

    async def delete(self, *args: Any, each: bool = False, **kwargs: Any) -> int:
        """
        Deletes from the model table after applying the filters from kwargs.

        You have to either pass a filter to narrow down a query or explicitly pass
        each=True flag to affect whole table.

        :param each: flag if whole table should be affected if no filter is passed
        :type each: bool
        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: number of deleted rows
        :rtype:int
        """
        if kwargs or args:
            return await self.filter(*args, **kwargs).delete()
        if not each and not (self.filter_clauses or self.exclude_clauses):
            raise QueryDefinitionError(
                "You cannot delete without filtering the queryset first. "
                "If you want to delete all rows use delete(each=True)"
            )
        expr = FilterQuery(filter_clauses=self.filter_clauses).apply(
            self.table.delete()
        )
        expr = FilterQuery(filter_clauses=self.exclude_clauses, exclude=True).apply(
            expr
        )
        return await self.database.execute(expr)

    def paginate(self, page: int, page_size: int = 20) -> "QuerySet[T]":
        """
        You can paginate the result which is a combination of offset and limit clauses.
        Limit is set to page size and offset is set to (page-1) * page_size.

        :param page_size: numbers of items per page
        :type page_size: int
        :param page: page number
        :type page: int
        :return: QuerySet
        :rtype: QuerySet
        """
        if page < 1 or page_size < 1:
            raise QueryDefinitionError("Page size and page have to be greater than 0.")

        limit_count = page_size
        query_offset = (page - 1) * page_size
        return self.rebuild_self(limit_count=limit_count, offset=query_offset)

    def limit(
        self, limit_count: int, limit_raw_sql: Optional[bool] = None
    ) -> "QuerySet[T]":
        """
        You can limit the results to desired number of parent models.

        To limit the actual number of database query rows instead of number of main
        models use the `limit_raw_sql` parameter flag, and set it to `True`.

        :param limit_raw_sql: flag if raw sql should be limited
        :type limit_raw_sql: bool
        :param limit_count: number of models to limit
        :type limit_count: int
        :return: QuerySet
        :rtype: QuerySet
        """
        limit_raw_sql = self.limit_sql_raw if limit_raw_sql is None else limit_raw_sql
        return self.rebuild_self(limit_count=limit_count, limit_raw_sql=limit_raw_sql)

    def offset(
        self, offset: int, limit_raw_sql: Optional[bool] = None
    ) -> "QuerySet[T]":
        """
        You can also offset the results by desired number of main models.

        To offset the actual number of database query rows instead of number of main
        models use the `limit_raw_sql` parameter flag, and set it to `True`.

        :param limit_raw_sql: flag if raw sql should be offset
        :type limit_raw_sql: bool
        :param offset: numbers of models to offset
        :type offset: int
        :return: QuerySet
        :rtype: QuerySet
        """
        limit_raw_sql = self.limit_sql_raw if limit_raw_sql is None else limit_raw_sql
        return self.rebuild_self(offset=offset, limit_raw_sql=limit_raw_sql)

    async def first(self, *args: Any, **kwargs: Any) -> "T":
        """
        Gets the first row from the db ordered by primary key column ascending.

        :raises NoMatch: if no rows are returned
        :raises MultipleMatches: if more than 1 row is returned.
        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: returned model
        :rtype: Model
        """
        if kwargs or args:
            return await self.filter(*args, **kwargs).first()

        expr = self.build_select_expression(
            limit=1,
            order_bys=(
                [
                    OrderAction(
                        order_str=f"{self.model.ormar_config.pkname}",
                        model_cls=self.model_cls,  # type: ignore
                    )
                ]
                if not any([x.is_source_model_order for x in self.order_bys])
                else []
            )
            + self.order_bys,
        )
        rows = await self.database.fetch_all(expr)
        processed_rows = await self._process_query_result_rows(rows)
        if self._prefetch_related and processed_rows:
            processed_rows = await self._prefetch_related_models(processed_rows, rows)
        self.check_single_result_rows_count(processed_rows)
        return processed_rows[0]  # type: ignore

    async def first_or_none(self, *args: Any, **kwargs: Any) -> Optional["T"]:
        """
        Gets the first row from the db ordered by primary key column ascending.

        If no match is found None will be returned.

        :raises MultipleMatches: if more than 1 row is returned.
        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: returned model
        :rtype: Model
        """
        try:
            return await self.first(*args, **kwargs)
        except ormar.NoMatch:
            return None

    async def get_or_none(self, *args: Any, **kwargs: Any) -> Optional["T"]:
        """
        Gets the first row from the db meeting the criteria set by kwargs.

        If no criteria set it will return the last row in db sorted by pk.

        Passing a criteria is actually calling filter(*args, **kwargs) method described
        below.

        If no match is found None will be returned.

        :raises MultipleMatches: if more than 1 row is returned.
        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: returned model
        :rtype: Model
        """
        try:
            return await self.get(*args, **kwargs)
        except ormar.NoMatch:
            return None

    async def get(self, *args: Any, **kwargs: Any) -> "T":  # noqa: CCR001
        """
        Gets the first row from the db meeting the criteria set by kwargs.

        If no criteria set it will return the last row in db sorted by pk.

        Passing a criteria is actually calling filter(*args, **kwargs) method described
        below.

        :raises NoMatch: if no rows are returned
        :raises MultipleMatches: if more than 1 row is returned.
        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: returned model
        :rtype: Model
        """
        if kwargs or args:
            return await self.filter(*args, **kwargs).get()

        if not self.filter_clauses:
            expr = self.build_select_expression(
                limit=1,
                order_bys=(
                    [
                        OrderAction(
                            order_str=f"-{self.model.ormar_config.pkname}",
                            model_cls=self.model_cls,  # type: ignore
                        )
                    ]
                    if not any([x.is_source_model_order for x in self.order_bys])
                    else []
                )
                + self.order_bys,
            )
        else:
            expr = self.build_select_expression()

        rows = await self.database.fetch_all(expr)
        processed_rows = await self._process_query_result_rows(rows)
        if self._prefetch_related and processed_rows:
            processed_rows = await self._prefetch_related_models(processed_rows, rows)
        self.check_single_result_rows_count(processed_rows)
        return processed_rows[0]  # type: ignore

    async def get_or_create(
        self,
        _defaults: Optional[Dict[str, Any]] = None,
        *args: Any,
        **kwargs: Any,
    ) -> Tuple["T", bool]:
        """
        Combination of create and get methods.

        Tries to get a row meeting the criteria for kwargs
        and if `NoMatch` exception is raised
        it creates a new one with given kwargs and _defaults.

        Passing a criteria is actually calling filter(*args, **kwargs) method described
        below.

        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :param _defaults: default values for creating object
        :type _defaults: Optional[Dict[str, Any]]
        :return: model instance and a boolean
        :rtype: Tuple("T", bool)
        """
        try:
            return await self.get(*args, **kwargs), False
        except NoMatch:
            _defaults = _defaults or {}
            return await self.create(**{**kwargs, **_defaults}), True

    async def update_or_create(self, **kwargs: Any) -> "T":
        """
        Updates the model, or in case there is no match in database creates a new one.

        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: updated or created model
        :rtype: Model
        """
        pk_name = self.model_config.pkname
        if "pk" in kwargs:
            kwargs[pk_name] = kwargs.pop("pk")
        if pk_name not in kwargs or kwargs.get(pk_name) is None:
            return await self.create(**kwargs)
        model = await self.get(pk=kwargs[pk_name])
        return await model.update(**kwargs)

    async def all(self, *args: Any, **kwargs: Any) -> List["T"]:  # noqa: A003
        """
        Returns all rows from a database for given model for set filter options.

        Passing args and/or kwargs is a shortcut and equals to calling
        `filter(*args, **kwargs).all()`.

        If there are no rows meeting the criteria an empty list is returned.

        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: list of returned models
        :rtype: List[Model]
        """
        if kwargs or args:
            return await self.filter(*args, **kwargs).all()

        expr = self.build_select_expression()
        rows = await self.database.fetch_all(expr)
        result_rows = await self._process_query_result_rows(rows)
        if self._prefetch_related and result_rows:
            result_rows = await self._prefetch_related_models(result_rows, rows)

        return result_rows

    async def iterate(  # noqa: A003
        self,
        *args: Any,
        **kwargs: Any,
    ) -> AsyncGenerator["T", None]:
        """
        Return async iterable generator for all rows from a database for given model.

        Passing args and/or kwargs is a shortcut and equals to calling
        `filter(*args, **kwargs).iterate()`.

        If there are no rows meeting the criteria an empty async generator is returned.

        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: asynchronous iterable generator of returned models
        :rtype: AsyncGenerator[Model]
        """

        if self._prefetch_related:
            raise QueryDefinitionError(
                "Prefetch related queries are not supported in iterators"
            )

        if kwargs or args:
            async for result in self.filter(*args, **kwargs).iterate():
                yield result
            return

        expr = self.build_select_expression()

        rows: list = []
        last_primary_key = None
        pk_alias = self.model.get_column_alias(self.model_config.pkname)

        async for row in self.database.iterate(query=expr):
            current_primary_key = row[pk_alias]
            if last_primary_key == current_primary_key or last_primary_key is None:
                last_primary_key = current_primary_key
                rows.append(row)
                continue

            yield (await self._process_query_result_rows(rows))[0]
            last_primary_key = current_primary_key
            rows = [row]

        if rows:
            yield (await self._process_query_result_rows(rows))[0]

    async def create(self, **kwargs: Any) -> "T":
        """
        Creates the model instance, saves it in a database and returns the updates model
        (with pk populated if not passed and autoincrement is set).

        The allowed kwargs are `Model` fields names and proper value types.

        :param kwargs: fields names and proper value types
        :type kwargs: Any
        :return: created model
        :rtype: Model
        """
        instance = self.model(**kwargs)
        instance = await instance.save()
        return instance

    async def bulk_create(self, objects: List["T"]) -> None:
        """
        Performs a bulk create in one database session to speed up the process.

        Allows you to create multiple objects at once.

        A valid list of `Model` objects needs to be passed.

        Bulk operations do not send signals.

        :param objects: list of ormar models already initialized and ready to save.
        :type objects: List[Model]
        """

        if not objects:
            raise ModelListEmptyError("Bulk create objects are empty!")

        ready_objects = []
        for obj in objects:
            ready_objects.append(obj.prepare_model_to_save(obj.model_dump()))
            await asyncio.sleep(0)  # Allow context switching to prevent blocking

        # don't use execute_many, as in databases it's executed in a loop
        # instead of using execute_many from drivers
        expr = self.table.insert().values(ready_objects)
        await self.database.execute(expr)

        for obj in objects:
            obj.set_save_status(True)

    async def bulk_update(  # noqa:  CCR001
        self, objects: List["T"], columns: Optional[List[str]] = None
    ) -> None:
        """
        Performs bulk update in one database session to speed up the process.

        Allows you to update multiple instance at once.

        All `Models` passed need to have primary key column populated.

        You can also select which fields to update by passing `columns` list
        as a list of string names.

        Bulk operations do not send signals.

        :param objects: list of ormar models
        :type objects: List[Model]
        :param columns: list of columns to update
        :type columns: List[str]
        """
        if not objects:
            raise ModelListEmptyError("Bulk update objects are empty!")

        ready_objects = []
        pk_name = self.model_config.pkname
        if not columns:
            columns = list(
                self.model.extract_db_own_fields().union(
                    self.model.extract_related_names()
                )
            )

        if pk_name not in columns:
            columns.append(pk_name)

        columns = [self.model.get_column_alias(k) for k in columns]

        for obj in objects:
            new_kwargs = obj.model_dump()
            if new_kwargs.get(pk_name) is None:
                raise ModelPersistenceError(
                    "You cannot update unsaved objects. "
                    f"{self.model.__name__} has to have {pk_name} filled."
                )
            new_kwargs = obj.prepare_model_to_update(new_kwargs)
            ready_objects.append(
                {"new_" + k: v for k, v in new_kwargs.items() if k in columns}
            )
            await asyncio.sleep(0)

        pk_column = self.model_config.table.c.get(self.model.get_column_alias(pk_name))
        pk_column_name = self.model.get_column_alias(pk_name)
        table_columns = [c.name for c in self.model_config.table.c]
        expr = self.table.update().where(
            pk_column == bindparam("new_" + pk_column_name)
        )
        expr = expr.values(
            **{
                k: bindparam("new_" + k)
                for k in columns
                if k != pk_column_name and k in table_columns
            }
        )
        # databases bind params only where query is passed as string
        # otherwise it just passes all data to values and results in unconsumed columns
        expr = str(expr)
        await self.database.execute_many(expr, ready_objects)

        for obj in objects:
            obj.set_save_status(True)

        await cast(
            Type["Model"], self.model_cls
        ).ormar_config.signals.post_bulk_update.send(
            sender=self.model_cls, instances=objects  # type: ignore
        )

database: databases.Database property

Shortcut to models database from OrmarConfig class.

:return: database :rtype: databases.Database

model: Type[T] property

Shortcut to model class set on QuerySet.

:return: model class :rtype: Type[Model]

model_config: OrmarConfig property

Shortcut to model class OrmarConfig set on QuerySet model.

:return: OrmarConfig of the model :rtype: model's OrmarConfig

table: sqlalchemy.Table property

Shortcut to models table from OrmarConfig.

:return: database table :rtype: sqlalchemy.Table

all(*args, **kwargs) async

Returns all rows from a database for given model for set filter options.

Passing args and/or kwargs is a shortcut and equals to calling filter(*args, **kwargs).all().

If there are no rows meeting the criteria an empty list is returned.

:param kwargs: fields names and proper value types :type kwargs: Any :return: list of returned models :rtype: List[Model]

Source code in ormar/queryset/queryset.py

async def all(self, *args: Any, **kwargs: Any) -> List["T"]:  # noqa: A003
    """
    Returns all rows from a database for given model for set filter options.

    Passing args and/or kwargs is a shortcut and equals to calling
    `filter(*args, **kwargs).all()`.

    If there are no rows meeting the criteria an empty list is returned.

    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: list of returned models
    :rtype: List[Model]
    """
    if kwargs or args:
        return await self.filter(*args, **kwargs).all()

    expr = self.build_select_expression()
    rows = await self.database.fetch_all(expr)
    result_rows = await self._process_query_result_rows(rows)
    if self._prefetch_related and result_rows:
        result_rows = await self._prefetch_related_models(result_rows, rows)

    return result_rows

avg(columns) async

Returns avg value of columns for rows matching the given criteria (applied with filter and exclude if set before).

:return: avg value of columns :rtype: Union[int, float, List]

Source code in ormar/queryset/queryset.py

async def avg(self, columns: Union[str, List[str]]) -> Any:
    """
    Returns avg value of columns for rows matching the given criteria
    (applied with `filter` and `exclude` if set before).

    :return: avg value of columns
    :rtype: Union[int, float, List]
    """
    if not isinstance(columns, list):
        columns = [columns]
    return await self._query_aggr_function(func_name="avg", columns=columns)

build_select_expression(limit=None, offset=None, order_bys=None)

Constructs the actual database query used in the QuerySet. If any of the params is not passed the QuerySet own value is used.

:param limit: number to limit the query :type limit: int :param offset: number to offset by :type offset: int :param order_bys: list of order-by fields names :type order_bys: List :return: built sqlalchemy select expression :rtype: sqlalchemy.sql.selectable.Select

Source code in ormar/queryset/queryset.py

def build_select_expression(
    self,
    limit: Optional[int] = None,
    offset: Optional[int] = None,
    order_bys: Optional[List] = None,
) -> sqlalchemy.sql.select:
    """
    Constructs the actual database query used in the QuerySet.
    If any of the params is not passed the QuerySet own value is used.

    :param limit: number to limit the query
    :type limit: int
    :param offset: number to offset by
    :type offset: int
    :param order_bys: list of order-by fields names
    :type order_bys: List
    :return: built sqlalchemy select expression
    :rtype: sqlalchemy.sql.selectable.Select
    """
    qry = Query(
        model_cls=self.model,
        select_related=self._select_related,
        filter_clauses=self.filter_clauses,
        exclude_clauses=self.exclude_clauses,
        offset=offset or self.query_offset,
        excludable=self._excludable,
        order_bys=order_bys or self.order_bys,
        limit_raw_sql=self.limit_sql_raw,
        limit_count=limit if limit is not None else self.limit_count,
    )
    exp = qry.build_select_expression()
    # print("\n", exp.compile(compile_kwargs={"literal_binds": True}))
    return exp

bulk_create(objects) async

Performs a bulk create in one database session to speed up the process.

Allows you to create multiple objects at once.

A valid list of Model objects needs to be passed.

Bulk operations do not send signals.

:param objects: list of ormar models already initialized and ready to save. :type objects: List[Model]

Source code in ormar/queryset/queryset.py

async def bulk_create(self, objects: List["T"]) -> None:
    """
    Performs a bulk create in one database session to speed up the process.

    Allows you to create multiple objects at once.

    A valid list of `Model` objects needs to be passed.

    Bulk operations do not send signals.

    :param objects: list of ormar models already initialized and ready to save.
    :type objects: List[Model]
    """

    if not objects:
        raise ModelListEmptyError("Bulk create objects are empty!")

    ready_objects = []
    for obj in objects:
        ready_objects.append(obj.prepare_model_to_save(obj.model_dump()))
        await asyncio.sleep(0)  # Allow context switching to prevent blocking

    # don't use execute_many, as in databases it's executed in a loop
    # instead of using execute_many from drivers
    expr = self.table.insert().values(ready_objects)
    await self.database.execute(expr)

    for obj in objects:
        obj.set_save_status(True)

bulk_update(objects, columns=None) async

Performs bulk update in one database session to speed up the process.

Allows you to update multiple instance at once.

All Models passed need to have primary key column populated.

You can also select which fields to update by passing columns list as a list of string names.

Bulk operations do not send signals.

:param objects: list of ormar models :type objects: List[Model] :param columns: list of columns to update :type columns: List[str]

Source code in ormar/queryset/queryset.py

async def bulk_update(  # noqa:  CCR001
    self, objects: List["T"], columns: Optional[List[str]] = None
) -> None:
    """
    Performs bulk update in one database session to speed up the process.

    Allows you to update multiple instance at once.

    All `Models` passed need to have primary key column populated.

    You can also select which fields to update by passing `columns` list
    as a list of string names.

    Bulk operations do not send signals.

    :param objects: list of ormar models
    :type objects: List[Model]
    :param columns: list of columns to update
    :type columns: List[str]
    """
    if not objects:
        raise ModelListEmptyError("Bulk update objects are empty!")

    ready_objects = []
    pk_name = self.model_config.pkname
    if not columns:
        columns = list(
            self.model.extract_db_own_fields().union(
                self.model.extract_related_names()
            )
        )

    if pk_name not in columns:
        columns.append(pk_name)

    columns = [self.model.get_column_alias(k) for k in columns]

    for obj in objects:
        new_kwargs = obj.model_dump()
        if new_kwargs.get(pk_name) is None:
            raise ModelPersistenceError(
                "You cannot update unsaved objects. "
                f"{self.model.__name__} has to have {pk_name} filled."
            )
        new_kwargs = obj.prepare_model_to_update(new_kwargs)
        ready_objects.append(
            {"new_" + k: v for k, v in new_kwargs.items() if k in columns}
        )
        await asyncio.sleep(0)

    pk_column = self.model_config.table.c.get(self.model.get_column_alias(pk_name))
    pk_column_name = self.model.get_column_alias(pk_name)
    table_columns = [c.name for c in self.model_config.table.c]
    expr = self.table.update().where(
        pk_column == bindparam("new_" + pk_column_name)
    )
    expr = expr.values(
        **{
            k: bindparam("new_" + k)
            for k in columns
            if k != pk_column_name and k in table_columns
        }
    )
    # databases bind params only where query is passed as string
    # otherwise it just passes all data to values and results in unconsumed columns
    expr = str(expr)
    await self.database.execute_many(expr, ready_objects)

    for obj in objects:
        obj.set_save_status(True)

    await cast(
        Type["Model"], self.model_cls
    ).ormar_config.signals.post_bulk_update.send(
        sender=self.model_cls, instances=objects  # type: ignore
    )

check_single_result_rows_count(rows) staticmethod

Verifies if the result has one and only one row.

:param rows: one element list of Models :type rows: List[Model]

Source code in ormar/queryset/queryset.py

@staticmethod
def check_single_result_rows_count(rows: Sequence[Optional["T"]]) -> None:
    """
    Verifies if the result has one and only one row.

    :param rows: one element list of Models
    :type rows: List[Model]
    """
    if not rows or rows[0] is None:
        raise NoMatch()
    if len(rows) > 1:
        raise MultipleMatches()

count(distinct=True) async

Returns number of rows matching the given criteria (applied with filter and exclude if set before). If distinct is True (the default), this will return the number of primary rows selected. If False, the count will be the total number of rows returned (including extra rows for one-to-many or many-to-many left select_related table joins). False is the legacy (buggy) behavior for workflows that depend on it.

:param distinct: flag if the primary table rows should be distinct or not

:return: number of rows :rtype: int

Source code in ormar/queryset/queryset.py

async def count(self, distinct: bool = True) -> int:
    """
    Returns number of rows matching the given criteria
    (applied with `filter` and `exclude` if set before).
    If `distinct` is `True` (the default), this will return
    the number of primary rows selected. If `False`,
    the count will be the total number of rows returned
    (including extra rows for `one-to-many` or `many-to-many`
    left `select_related` table joins).
    `False` is the legacy (buggy) behavior for workflows that depend on it.

    :param distinct: flag if the primary table rows should be distinct or not

    :return: number of rows
    :rtype: int
    """
    expr = self.build_select_expression().alias("subquery_for_count")
    expr = sqlalchemy.func.count().select().select_from(expr)
    if distinct:
        pk_column_name = self.model.get_column_alias(self.model_config.pkname)
        expr_distinct = expr.group_by(pk_column_name).alias("subquery_for_group")
        expr = sqlalchemy.func.count().select().select_from(expr_distinct)
    return await self.database.fetch_val(expr)

create(**kwargs) async

Creates the model instance, saves it in a database and returns the updates model (with pk populated if not passed and autoincrement is set).

The allowed kwargs are Model fields names and proper value types.

:param kwargs: fields names and proper value types :type kwargs: Any :return: created model :rtype: Model

Source code in ormar/queryset/queryset.py

async def create(self, **kwargs: Any) -> "T":
    """
    Creates the model instance, saves it in a database and returns the updates model
    (with pk populated if not passed and autoincrement is set).

    The allowed kwargs are `Model` fields names and proper value types.

    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: created model
    :rtype: Model
    """
    instance = self.model(**kwargs)
    instance = await instance.save()
    return instance

delete(*args, each=False, **kwargs) async

Deletes from the model table after applying the filters from kwargs.

You have to either pass a filter to narrow down a query or explicitly pass each=True flag to affect whole table.

:param each: flag if whole table should be affected if no filter is passed :type each: bool :param kwargs: fields names and proper value types :type kwargs: Any :return: number of deleted rows :rtype:int

Source code in ormar/queryset/queryset.py

async def delete(self, *args: Any, each: bool = False, **kwargs: Any) -> int:
    """
    Deletes from the model table after applying the filters from kwargs.

    You have to either pass a filter to narrow down a query or explicitly pass
    each=True flag to affect whole table.

    :param each: flag if whole table should be affected if no filter is passed
    :type each: bool
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: number of deleted rows
    :rtype:int
    """
    if kwargs or args:
        return await self.filter(*args, **kwargs).delete()
    if not each and not (self.filter_clauses or self.exclude_clauses):
        raise QueryDefinitionError(
            "You cannot delete without filtering the queryset first. "
            "If you want to delete all rows use delete(each=True)"
        )
    expr = FilterQuery(filter_clauses=self.filter_clauses).apply(
        self.table.delete()
    )
    expr = FilterQuery(filter_clauses=self.exclude_clauses, exclude=True).apply(
        expr
    )
    return await self.database.execute(expr)

exclude(*args, **kwargs)

Works exactly the same as filter and all modifiers (suffixes) are the same, but returns a not condition.

So if you use filter(name='John') which is where name = 'John' in SQL, the exclude(name='John') equals to where name <> 'John'

Note that all conditions are joined so if you pass multiple values it becomes a union of conditions.

exclude(name='John', age>=35) will become where not (name='John' and age>=35)

:param kwargs: fields names and proper value types :type kwargs: Any :return: filtered QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def exclude(self, *args: Any, **kwargs: Any) -> "QuerySet[T]":  # noqa: A003
    """
    Works exactly the same as filter and all modifiers (suffixes) are the same,
    but returns a *not* condition.

    So if you use `filter(name='John')` which is `where name = 'John'` in SQL,
    the `exclude(name='John')` equals to `where name <> 'John'`

    Note that all conditions are joined so if you pass multiple values it
    becomes a union of conditions.

    `exclude(name='John', age>=35)` will become
    `where not (name='John' and age>=35)`

    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: filtered QuerySet
    :rtype: QuerySet
    """
    return self.filter(_exclude=True, *args, **kwargs)

exclude_fields(columns)

With exclude_fields() you can select subset of model columns that will be excluded to limit the data load.

It's the opposite of fields() method so check documentation above to see what options are available.

Especially check above how you can pass also nested dictionaries and sets as a mask to exclude fields from whole hierarchy.

Note that fields() and exclude_fields() works both for main models (on normal queries like get, all etc.) as well as select_related and prefetch_related models (with nested notation).

Mandatory fields cannot be excluded as it will raise ValidationError, to exclude a field it has to be nullable.

Pk column cannot be excluded - it's always auto added even if explicitly excluded.

:param columns: columns to exclude :type columns: Union[List, str, Set, Dict] :return: QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def exclude_fields(self, columns: Union[List, str, Set, Dict]) -> "QuerySet[T]":
    """
    With `exclude_fields()` you can select subset of model columns that will
    be excluded to limit the data load.

    It's the opposite of `fields()` method so check documentation above
    to see what options are available.

    Especially check above how you can pass also nested dictionaries
    and sets as a mask to exclude fields from whole hierarchy.

    Note that `fields()` and `exclude_fields()` works both for main models
    (on normal queries like `get`, `all` etc.)
    as well as `select_related` and `prefetch_related` models
    (with nested notation).

    Mandatory fields cannot be excluded as it will raise `ValidationError`,
    to exclude a field it has to be nullable.

    Pk column cannot be excluded - it's always auto added even
    if explicitly excluded.

    :param columns: columns to exclude
    :type columns: Union[List, str, Set, Dict]
    :return: QuerySet
    :rtype: QuerySet
    """
    return self.fields(columns=columns, _is_exclude=True)

exists() async

Returns a bool value to confirm if there are rows matching the given criteria (applied with filter and exclude if set).

:return: result of the check :rtype: bool

Source code in ormar/queryset/queryset.py

async def exists(self) -> bool:
    """
    Returns a bool value to confirm if there are rows matching the given criteria
    (applied with `filter` and `exclude` if set).

    :return: result of the check
    :rtype: bool
    """
    expr = self.build_select_expression()
    expr = sqlalchemy.exists(expr).select()
    return await self.database.fetch_val(expr)

fields(columns, _is_exclude=False)

With fields() you can select subset of model columns to limit the data load.

Note that fields() and exclude_fields() works both for main models (on normal queries like get, all etc.) as well as select_related and prefetch_related models (with nested notation).

You can select specified fields by passing a str, List[str], Set[str] or dict with nested definition.

To include related models use notation {related_name}__{column}[__{optional_next} etc.].

fields() can be called several times, building up the columns to select.

If you include related models into select_related() call but you won't specify columns for those models in fields - implies a list of all fields for those nested models.

Mandatory fields cannot be excluded as it will raise ValidationError, to exclude a field it has to be nullable.

Pk column cannot be excluded - it's always auto added even if not explicitly included.

You can also pass fields to include as dictionary or set.

To mark a field as included in a dictionary use it's name as key and ellipsis as value.

To traverse nested models use nested dictionaries.

To include fields at last level instead of nested dictionary a set can be used.

To include whole nested model specify model related field name and ellipsis.

:param _is_exclude: flag if it's exclude or include operation :type _is_exclude: bool :param columns: columns to include :type columns: Union[List, str, Set, Dict] :return: QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def fields(
    self, columns: Union[List, str, Set, Dict], _is_exclude: bool = False
) -> "QuerySet[T]":
    """
    With `fields()` you can select subset of model columns to limit the data load.

    Note that `fields()` and `exclude_fields()` works both for main models
    (on normal queries like `get`, `all` etc.)
    as well as `select_related` and `prefetch_related`
    models (with nested notation).

    You can select specified fields by passing a `str, List[str], Set[str] or
    dict` with nested definition.

    To include related models use notation
    `{related_name}__{column}[__{optional_next} etc.]`.

    `fields()` can be called several times, building up the columns to select.

    If you include related models into `select_related()` call but you won't specify
    columns for those models in fields - implies a list of all fields for
    those nested models.

    Mandatory fields cannot be excluded as it will raise `ValidationError`,
    to exclude a field it has to be nullable.

    Pk column cannot be excluded - it's always auto added even if
    not explicitly included.

    You can also pass fields to include as dictionary or set.

    To mark a field as included in a dictionary use it's name as key
    and ellipsis as value.

    To traverse nested models use nested dictionaries.

    To include fields at last level instead of nested dictionary a set can be used.

    To include whole nested model specify model related field name and ellipsis.

    :param _is_exclude: flag if it's exclude or include operation
    :type _is_exclude: bool
    :param columns: columns to include
    :type columns: Union[List, str, Set, Dict]
    :return: QuerySet
    :rtype: QuerySet
    """
    excludable = ormar.ExcludableItems.from_excludable(self._excludable)
    excludable.build(
        items=columns,
        model_cls=self.model_cls,  # type: ignore
        is_exclude=_is_exclude,
    )

    return self.rebuild_self(excludable=excludable)

filter(*args, _exclude=False, **kwargs)

Allows you to filter by any Model attribute/field as well as to fetch instances, with a filter across an FK relationship.

You can use special filter suffix to change the filter operands:

• exact - like album__name__exact='Malibu' (exact match)
• iexact - like album__name__iexact='malibu' (exact match case insensitive)
• contains - like album__name__contains='Mal' (sql like)
• icontains - like album__name__icontains='mal' (sql like case insensitive)
• in - like album__name__in=['Malibu', 'Barclay'] (sql in)
• isnull - like album__name__isnull=True (sql is null) (isnotnull album__name__isnull=False (sql is not null))
• gt - like position__gt=3 (sql >)
• gte - like position__gte=3 (sql >=)
• lt - like position__lt=3 (sql <)
• lte - like position__lte=3 (sql <=)
• startswith - like album__name__startswith='Mal' (exact start match)
• istartswith - like album__name__istartswith='mal' (case insensitive)
• endswith - like album__name__endswith='ibu' (exact end match)
• iendswith - like album__name__iendswith='IBU' (case insensitive)

Note that you can also use python style filters - check the docs!

:param _exclude: flag if it should be exclude or filter :type _exclude: bool :param kwargs: fields names and proper value types :type kwargs: Any :return: filtered QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def filter(  # noqa: A003
    self, *args: Any, _exclude: bool = False, **kwargs: Any
) -> "QuerySet[T]":
    """
    Allows you to filter by any `Model` attribute/field
    as well as to fetch instances, with a filter across an FK relationship.

    You can use special filter suffix to change the filter operands:

    *  exact - like `album__name__exact='Malibu'` (exact match)
    *  iexact - like `album__name__iexact='malibu'` (exact match case insensitive)
    *  contains - like `album__name__contains='Mal'` (sql like)
    *  icontains - like `album__name__icontains='mal'` (sql like case insensitive)
    *  in - like `album__name__in=['Malibu', 'Barclay']` (sql in)
    *  isnull - like `album__name__isnull=True` (sql is null)
       (isnotnull `album__name__isnull=False` (sql is not null))
    *  gt - like `position__gt=3` (sql >)
    *  gte - like `position__gte=3` (sql >=)
    *  lt - like `position__lt=3` (sql <)
    *  lte - like `position__lte=3` (sql <=)
    *  startswith - like `album__name__startswith='Mal'` (exact start match)
    *  istartswith - like `album__name__istartswith='mal'` (case insensitive)
    *  endswith - like `album__name__endswith='ibu'` (exact end match)
    *  iendswith - like `album__name__iendswith='IBU'` (case insensitive)

    Note that you can also use python style filters - check the docs!

    :param _exclude: flag if it should be exclude or filter
    :type _exclude: bool
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: filtered QuerySet
    :rtype: QuerySet
    """
    filter_groups, select_related = self._resolve_filter_groups(groups=args)
    qryclause = QueryClause(
        model_cls=self.model,
        select_related=select_related,
        filter_clauses=self.filter_clauses,
    )
    filter_clauses, select_related = qryclause.prepare_filter(**kwargs)
    filter_clauses = filter_clauses + filter_groups  # type: ignore
    if _exclude:
        exclude_clauses = filter_clauses
        filter_clauses = self.filter_clauses
    else:
        exclude_clauses = self.exclude_clauses
        filter_clauses = filter_clauses

    return self.rebuild_self(
        filter_clauses=filter_clauses,
        exclude_clauses=exclude_clauses,
        select_related=select_related,
    )

first(*args, **kwargs) async

Gets the first row from the db ordered by primary key column ascending.

:raises NoMatch: if no rows are returned :raises MultipleMatches: if more than 1 row is returned. :param kwargs: fields names and proper value types :type kwargs: Any :return: returned model :rtype: Model

Source code in ormar/queryset/queryset.py

async def first(self, *args: Any, **kwargs: Any) -> "T":
    """
    Gets the first row from the db ordered by primary key column ascending.

    :raises NoMatch: if no rows are returned
    :raises MultipleMatches: if more than 1 row is returned.
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: returned model
    :rtype: Model
    """
    if kwargs or args:
        return await self.filter(*args, **kwargs).first()

    expr = self.build_select_expression(
        limit=1,
        order_bys=(
            [
                OrderAction(
                    order_str=f"{self.model.ormar_config.pkname}",
                    model_cls=self.model_cls,  # type: ignore
                )
            ]
            if not any([x.is_source_model_order for x in self.order_bys])
            else []
        )
        + self.order_bys,
    )
    rows = await self.database.fetch_all(expr)
    processed_rows = await self._process_query_result_rows(rows)
    if self._prefetch_related and processed_rows:
        processed_rows = await self._prefetch_related_models(processed_rows, rows)
    self.check_single_result_rows_count(processed_rows)
    return processed_rows[0]  # type: ignore

first_or_none(*args, **kwargs) async

Gets the first row from the db ordered by primary key column ascending.

If no match is found None will be returned.

:raises MultipleMatches: if more than 1 row is returned. :param kwargs: fields names and proper value types :type kwargs: Any :return: returned model :rtype: Model

Source code in ormar/queryset/queryset.py

async def first_or_none(self, *args: Any, **kwargs: Any) -> Optional["T"]:
    """
    Gets the first row from the db ordered by primary key column ascending.

    If no match is found None will be returned.

    :raises MultipleMatches: if more than 1 row is returned.
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: returned model
    :rtype: Model
    """
    try:
        return await self.first(*args, **kwargs)
    except ormar.NoMatch:
        return None

get(*args, **kwargs) async

Gets the first row from the db meeting the criteria set by kwargs.

If no criteria set it will return the last row in db sorted by pk.

Passing a criteria is actually calling filter(args, *kwargs) method described below.

:raises NoMatch: if no rows are returned :raises MultipleMatches: if more than 1 row is returned. :param kwargs: fields names and proper value types :type kwargs: Any :return: returned model :rtype: Model

Source code in ormar/queryset/queryset.py

async def get(self, *args: Any, **kwargs: Any) -> "T":  # noqa: CCR001
    """
    Gets the first row from the db meeting the criteria set by kwargs.

    If no criteria set it will return the last row in db sorted by pk.

    Passing a criteria is actually calling filter(*args, **kwargs) method described
    below.

    :raises NoMatch: if no rows are returned
    :raises MultipleMatches: if more than 1 row is returned.
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: returned model
    :rtype: Model
    """
    if kwargs or args:
        return await self.filter(*args, **kwargs).get()

    if not self.filter_clauses:
        expr = self.build_select_expression(
            limit=1,
            order_bys=(
                [
                    OrderAction(
                        order_str=f"-{self.model.ormar_config.pkname}",
                        model_cls=self.model_cls,  # type: ignore
                    )
                ]
                if not any([x.is_source_model_order for x in self.order_bys])
                else []
            )
            + self.order_bys,
        )
    else:
        expr = self.build_select_expression()

    rows = await self.database.fetch_all(expr)
    processed_rows = await self._process_query_result_rows(rows)
    if self._prefetch_related and processed_rows:
        processed_rows = await self._prefetch_related_models(processed_rows, rows)
    self.check_single_result_rows_count(processed_rows)
    return processed_rows[0]  # type: ignore

get_or_create(_defaults=None, *args, **kwargs) async

Combination of create and get methods.

Tries to get a row meeting the criteria for kwargs and if NoMatch exception is raised it creates a new one with given kwargs and _defaults.

Passing a criteria is actually calling filter(args, *kwargs) method described below.

:param kwargs: fields names and proper value types :type kwargs: Any :param _defaults: default values for creating object :type _defaults: Optional[Dict[str, Any]] :return: model instance and a boolean :rtype: Tuple("T", bool)

Source code in ormar/queryset/queryset.py

async def get_or_create(
    self,
    _defaults: Optional[Dict[str, Any]] = None,
    *args: Any,
    **kwargs: Any,
) -> Tuple["T", bool]:
    """
    Combination of create and get methods.

    Tries to get a row meeting the criteria for kwargs
    and if `NoMatch` exception is raised
    it creates a new one with given kwargs and _defaults.

    Passing a criteria is actually calling filter(*args, **kwargs) method described
    below.

    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :param _defaults: default values for creating object
    :type _defaults: Optional[Dict[str, Any]]
    :return: model instance and a boolean
    :rtype: Tuple("T", bool)
    """
    try:
        return await self.get(*args, **kwargs), False
    except NoMatch:
        _defaults = _defaults or {}
        return await self.create(**{**kwargs, **_defaults}), True

get_or_none(*args, **kwargs) async

Gets the first row from the db meeting the criteria set by kwargs.

If no criteria set it will return the last row in db sorted by pk.

Passing a criteria is actually calling filter(args, *kwargs) method described below.

If no match is found None will be returned.

:raises MultipleMatches: if more than 1 row is returned. :param kwargs: fields names and proper value types :type kwargs: Any :return: returned model :rtype: Model

Source code in ormar/queryset/queryset.py

async def get_or_none(self, *args: Any, **kwargs: Any) -> Optional["T"]:
    """
    Gets the first row from the db meeting the criteria set by kwargs.

    If no criteria set it will return the last row in db sorted by pk.

    Passing a criteria is actually calling filter(*args, **kwargs) method described
    below.

    If no match is found None will be returned.

    :raises MultipleMatches: if more than 1 row is returned.
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: returned model
    :rtype: Model
    """
    try:
        return await self.get(*args, **kwargs)
    except ormar.NoMatch:
        return None

iterate(*args, **kwargs) async

Return async iterable generator for all rows from a database for given model.

Passing args and/or kwargs is a shortcut and equals to calling filter(*args, **kwargs).iterate().

If there are no rows meeting the criteria an empty async generator is returned.

:param kwargs: fields names and proper value types :type kwargs: Any :return: asynchronous iterable generator of returned models :rtype: AsyncGenerator[Model]

Source code in ormar/queryset/queryset.py

async def iterate(  # noqa: A003
    self,
    *args: Any,
    **kwargs: Any,
) -> AsyncGenerator["T", None]:
    """
    Return async iterable generator for all rows from a database for given model.

    Passing args and/or kwargs is a shortcut and equals to calling
    `filter(*args, **kwargs).iterate()`.

    If there are no rows meeting the criteria an empty async generator is returned.

    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: asynchronous iterable generator of returned models
    :rtype: AsyncGenerator[Model]
    """

    if self._prefetch_related:
        raise QueryDefinitionError(
            "Prefetch related queries are not supported in iterators"
        )

    if kwargs or args:
        async for result in self.filter(*args, **kwargs).iterate():
            yield result
        return

    expr = self.build_select_expression()

    rows: list = []
    last_primary_key = None
    pk_alias = self.model.get_column_alias(self.model_config.pkname)

    async for row in self.database.iterate(query=expr):
        current_primary_key = row[pk_alias]
        if last_primary_key == current_primary_key or last_primary_key is None:
            last_primary_key = current_primary_key
            rows.append(row)
            continue

        yield (await self._process_query_result_rows(rows))[0]
        last_primary_key = current_primary_key
        rows = [row]

    if rows:
        yield (await self._process_query_result_rows(rows))[0]

limit(limit_count, limit_raw_sql=None)

You can limit the results to desired number of parent models.

To limit the actual number of database query rows instead of number of main models use the limit_raw_sql parameter flag, and set it to True.

:param limit_raw_sql: flag if raw sql should be limited :type limit_raw_sql: bool :param limit_count: number of models to limit :type limit_count: int :return: QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def limit(
    self, limit_count: int, limit_raw_sql: Optional[bool] = None
) -> "QuerySet[T]":
    """
    You can limit the results to desired number of parent models.

    To limit the actual number of database query rows instead of number of main
    models use the `limit_raw_sql` parameter flag, and set it to `True`.

    :param limit_raw_sql: flag if raw sql should be limited
    :type limit_raw_sql: bool
    :param limit_count: number of models to limit
    :type limit_count: int
    :return: QuerySet
    :rtype: QuerySet
    """
    limit_raw_sql = self.limit_sql_raw if limit_raw_sql is None else limit_raw_sql
    return self.rebuild_self(limit_count=limit_count, limit_raw_sql=limit_raw_sql)

max(columns) async

Returns max value of columns for rows matching the given criteria (applied with filter and exclude if set before).

:return: max value of column(s) :rtype: Any

Source code in ormar/queryset/queryset.py

async def max(self, columns: Union[str, List[str]]) -> Any:  # noqa: A003
    """
    Returns max value of columns for rows matching the given criteria
    (applied with `filter` and `exclude` if set before).

    :return: max value of column(s)
    :rtype: Any
    """
    if not isinstance(columns, list):
        columns = [columns]
    return await self._query_aggr_function(func_name="max", columns=columns)

min(columns) async

Returns min value of columns for rows matching the given criteria (applied with filter and exclude if set before).

:return: min value of column(s) :rtype: Any

Source code in ormar/queryset/queryset.py

async def min(self, columns: Union[str, List[str]]) -> Any:  # noqa: A003
    """
    Returns min value of columns for rows matching the given criteria
    (applied with `filter` and `exclude` if set before).

    :return: min value of column(s)
    :rtype: Any
    """
    if not isinstance(columns, list):
        columns = [columns]
    return await self._query_aggr_function(func_name="min", columns=columns)

offset(offset, limit_raw_sql=None)

You can also offset the results by desired number of main models.

To offset the actual number of database query rows instead of number of main models use the limit_raw_sql parameter flag, and set it to True.

:param limit_raw_sql: flag if raw sql should be offset :type limit_raw_sql: bool :param offset: numbers of models to offset :type offset: int :return: QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def offset(
    self, offset: int, limit_raw_sql: Optional[bool] = None
) -> "QuerySet[T]":
    """
    You can also offset the results by desired number of main models.

    To offset the actual number of database query rows instead of number of main
    models use the `limit_raw_sql` parameter flag, and set it to `True`.

    :param limit_raw_sql: flag if raw sql should be offset
    :type limit_raw_sql: bool
    :param offset: numbers of models to offset
    :type offset: int
    :return: QuerySet
    :rtype: QuerySet
    """
    limit_raw_sql = self.limit_sql_raw if limit_raw_sql is None else limit_raw_sql
    return self.rebuild_self(offset=offset, limit_raw_sql=limit_raw_sql)

order_by(columns)

With order_by() you can order the results from database based on your choice of fields.

You can provide a string with field name or list of strings with fields names.

Ordering in sql will be applied in order of names you provide in order_by.

By default if you do not provide ordering ormar explicitly orders by all primary keys

If you are sorting by nested models that causes that the result rows are unsorted by the main model ormar will combine those children rows into one main model.

The main model will never duplicate in the result

To order by main model field just provide a field name

To sort on nested models separate field names with dunder '__'.

You can sort this way across all relation types -> ForeignKey, reverse virtual FK and ManyToMany fields.

To sort in descending order provide a hyphen in front of the field name

:param columns: columns by which models should be sorted :type columns: Union[List, str] :return: QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def order_by(self, columns: Union[List, str, OrderAction]) -> "QuerySet[T]":
    """
    With `order_by()` you can order the results from database based on your
    choice of fields.

    You can provide a string with field name or list of strings with fields names.

    Ordering in sql will be applied in order of names you provide in order_by.

    By default if you do not provide ordering `ormar` explicitly orders by
    all primary keys

    If you are sorting by nested models that causes that the result rows are
    unsorted by the main model `ormar` will combine those children rows into
    one main model.

    The main model will never duplicate in the result

    To order by main model field just provide a field name

    To sort on nested models separate field names with dunder '__'.

    You can sort this way across all relation types -> `ForeignKey`,
    reverse virtual FK and `ManyToMany` fields.

    To sort in descending order provide a hyphen in front of the field name

    :param columns: columns by which models should be sorted
    :type columns: Union[List, str]
    :return: QuerySet
    :rtype: QuerySet
    """
    if not isinstance(columns, list):
        columns = [columns]

    orders_by = [
        (
            OrderAction(order_str=x, model_cls=self.model_cls)  # type: ignore
            if not isinstance(x, OrderAction)
            else x
        )
        for x in columns
    ]

    order_bys = self.order_bys + [x for x in orders_by if x not in self.order_bys]
    return self.rebuild_self(order_bys=order_bys)

paginate(page, page_size=20)

You can paginate the result which is a combination of offset and limit clauses. Limit is set to page size and offset is set to (page-1) * page_size.

:param page_size: numbers of items per page :type page_size: int :param page: page number :type page: int :return: QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def paginate(self, page: int, page_size: int = 20) -> "QuerySet[T]":
    """
    You can paginate the result which is a combination of offset and limit clauses.
    Limit is set to page size and offset is set to (page-1) * page_size.

    :param page_size: numbers of items per page
    :type page_size: int
    :param page: page number
    :type page: int
    :return: QuerySet
    :rtype: QuerySet
    """
    if page < 1 or page_size < 1:
        raise QueryDefinitionError("Page size and page have to be greater than 0.")

    limit_count = page_size
    query_offset = (page - 1) * page_size
    return self.rebuild_self(limit_count=limit_count, offset=query_offset)

prefetch_related(related)

Allows to prefetch related models during query - but opposite to select_related each subsequent model is fetched in a separate database query.

With prefetch_related always one query per Model is run against the database, meaning that you will have multiple queries executed one after another.

To fetch related model use ForeignKey names.

To chain related Models relation use double underscores between names.

:param related: list of relation field names, can be linked by '__' to nest :type related: Union[List, str] :return: QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def prefetch_related(
    self, related: Union[List, str, FieldAccessor]
) -> "QuerySet[T]":
    """
    Allows to prefetch related models during query - but opposite to
    `select_related` each subsequent model is fetched in a separate database query.

    **With `prefetch_related` always one query per Model is run against the
    database**, meaning that you will have multiple queries executed one
    after another.

    To fetch related model use `ForeignKey` names.

    To chain related `Models` relation use double underscores between names.

    :param related: list of relation field names, can be linked by '__' to nest
    :type related: Union[List, str]
    :return: QuerySet
    :rtype: QuerySet
    """
    if not isinstance(related, list):
        related = [related]
    related = [
        rel._access_chain if isinstance(rel, FieldAccessor) else rel
        for rel in related
    ]

    related = list(set(list(self._prefetch_related) + related))
    return self.rebuild_self(prefetch_related=related)

rebuild_self(filter_clauses=None, exclude_clauses=None, select_related=None, limit_count=None, offset=None, excludable=None, order_bys=None, prefetch_related=None, limit_raw_sql=None, proxy_source_model=None)

Method that returns new instance of queryset based on passed params, all not passed params are taken from current values.

Source code in ormar/queryset/queryset.py

def rebuild_self(  # noqa: CFQ002
    self,
    filter_clauses: Optional[List] = None,
    exclude_clauses: Optional[List] = None,
    select_related: Optional[List] = None,
    limit_count: Optional[int] = None,
    offset: Optional[int] = None,
    excludable: Optional["ExcludableItems"] = None,
    order_bys: Optional[List] = None,
    prefetch_related: Optional[List] = None,
    limit_raw_sql: Optional[bool] = None,
    proxy_source_model: Optional[Type["Model"]] = None,
) -> "QuerySet":
    """
    Method that returns new instance of queryset based on passed params,
    all not passed params are taken from current values.
    """
    overwrites = {
        "select_related": "_select_related",
        "offset": "query_offset",
        "excludable": "_excludable",
        "prefetch_related": "_prefetch_related",
        "limit_raw_sql": "limit_sql_raw",
    }
    passed_args = locals()

    def replace_if_none(arg_name: str) -> Any:
        if passed_args.get(arg_name) is None:
            return getattr(self, overwrites.get(arg_name, arg_name))
        return passed_args.get(arg_name)

    return self.__class__(
        model_cls=self.model_cls,
        filter_clauses=replace_if_none("filter_clauses"),
        exclude_clauses=replace_if_none("exclude_clauses"),
        select_related=replace_if_none("select_related"),
        limit_count=replace_if_none("limit_count"),
        offset=replace_if_none("offset"),
        excludable=replace_if_none("excludable"),
        order_bys=replace_if_none("order_bys"),
        prefetch_related=replace_if_none("prefetch_related"),
        limit_raw_sql=replace_if_none("limit_raw_sql"),
        proxy_source_model=replace_if_none("proxy_source_model"),
    )

select_all(follow=False)

By default adds only directly related models.

If follow=True is set it adds 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 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

Source code in ormar/queryset/queryset.py

def select_all(self, follow: bool = False) -> "QuerySet[T]":
    """
    By default adds only directly related models.

    If follow=True is set it adds 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 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.model.extract_related_names())
    if follow:
        relations = self.model._iterate_related_models()
    return self.rebuild_self(select_related=relations)

select_related(related)

Allows to prefetch related models during the same query.

With select_related always only one query is run against the database, meaning that one (sometimes complicated) join is generated and later nested models are processed in python.

To fetch related model use ForeignKey names.

To chain related Models relation use double underscores between names.

:param related: list of relation field names, can be linked by '__' to nest :type related: Union[List, str] :return: QuerySet :rtype: QuerySet

Source code in ormar/queryset/queryset.py

def select_related(self, related: Union[List, str, FieldAccessor]) -> "QuerySet[T]":
    """
    Allows to prefetch related models during the same query.

    **With `select_related` always only one query is run against the database**,
    meaning that one (sometimes complicated) join is generated and later nested
    models are processed in python.

    To fetch related model use `ForeignKey` names.

    To chain related `Models` relation use double underscores between names.

    :param related: list of relation field names, can be linked by '__' to nest
    :type related: Union[List, str]
    :return: QuerySet
    :rtype: QuerySet
    """
    if not isinstance(related, list):
        related = [related]
    related = [
        rel._access_chain if isinstance(rel, FieldAccessor) else rel
        for rel in related
    ]

    related = sorted(list(set(list(self._select_related) + related)))
    return self.rebuild_self(select_related=related)

sum(columns) async

Returns sum value of columns for rows matching the given criteria (applied with filter and exclude if set before).

:return: sum value of columns :rtype: int

Source code in ormar/queryset/queryset.py

async def sum(self, columns: Union[str, List[str]]) -> Any:  # noqa: A003
    """
    Returns sum value of columns for rows matching the given criteria
    (applied with `filter` and `exclude` if set before).

    :return: sum value of columns
    :rtype: int
    """
    if not isinstance(columns, list):
        columns = [columns]
    return await self._query_aggr_function(func_name="sum", columns=columns)

update(each=False, **kwargs) async

Updates the model table after applying the filters from kwargs.

You have to either pass a filter to narrow down a query or explicitly pass each=True flag to affect whole table.

:param each: flag if whole table should be affected if no filter is passed :type each: bool :param kwargs: fields names and proper value types :type kwargs: Any :return: number of updated rows :rtype: int

Source code in ormar/queryset/queryset.py

async def update(self, each: bool = False, **kwargs: Any) -> int:
    """
    Updates the model table after applying the filters from kwargs.

    You have to either pass a filter to narrow down a query or explicitly pass
    each=True flag to affect whole table.

    :param each: flag if whole table should be affected if no filter is passed
    :type each: bool
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: number of updated rows
    :rtype: int
    """
    if not each and not (self.filter_clauses or self.exclude_clauses):
        raise QueryDefinitionError(
            "You cannot update without filtering the queryset first. "
            "If you want to update all rows use update(each=True, **kwargs)"
        )

    self_fields = self.model.extract_db_own_fields().union(
        self.model.extract_related_names()
    )
    updates = {k: v for k, v in kwargs.items() if k in self_fields}
    updates = self.model.validate_enums(updates)
    updates = self.model.translate_columns_to_aliases(updates)

    expr = FilterQuery(filter_clauses=self.filter_clauses).apply(
        self.table.update().values(**updates)
    )
    expr = FilterQuery(filter_clauses=self.exclude_clauses, exclude=True).apply(
        expr
    )
    return await self.database.execute(expr)

update_or_create(**kwargs) async

Updates the model, or in case there is no match in database creates a new one.

:param kwargs: fields names and proper value types :type kwargs: Any :return: updated or created model :rtype: Model

Source code in ormar/queryset/queryset.py

async def update_or_create(self, **kwargs: Any) -> "T":
    """
    Updates the model, or in case there is no match in database creates a new one.

    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: updated or created model
    :rtype: Model
    """
    pk_name = self.model_config.pkname
    if "pk" in kwargs:
        kwargs[pk_name] = kwargs.pop("pk")
    if pk_name not in kwargs or kwargs.get(pk_name) is None:
        return await self.create(**kwargs)
    model = await self.get(pk=kwargs[pk_name])
    return await model.update(**kwargs)

values(fields=None, exclude_through=False, _as_dict=True, _flatten=False) async

Return a list of dictionaries with column values in order of the fields passed or all fields from queried models.

To filter for given row use filter/exclude methods before values, to limit number of rows use limit/offset or paginate before values.

Note that it always return a list even for one row from database.

:param exclude_through: flag if through models should be excluded :type exclude_through: bool :param _flatten: internal parameter to flatten one element tuples :type _flatten: bool :param _as_dict: internal parameter if return dict or tuples :type _as_dict: bool :param fields: field name or list of field names to extract from db :type fields: Union[List, str, Set, Dict]

Source code in ormar/queryset/queryset.py

async def values(
    self,
    fields: Union[List, str, Set, Dict, None] = None,
    exclude_through: bool = False,
    _as_dict: bool = True,
    _flatten: bool = False,
) -> List:
    """
    Return a list of dictionaries with column values in order of the fields
    passed or all fields from queried models.

    To filter for given row use filter/exclude methods before values,
    to limit number of rows use limit/offset or paginate before values.

    Note that it always return a list even for one row from database.

    :param exclude_through: flag if through models should be excluded
    :type exclude_through: bool
    :param _flatten: internal parameter to flatten one element tuples
    :type _flatten: bool
    :param _as_dict: internal parameter if return dict or tuples
    :type _as_dict: bool
    :param fields: field name or list of field names to extract from db
    :type fields:  Union[List, str, Set, Dict]
    """
    if fields:
        return await self.fields(columns=fields).values(
            _as_dict=_as_dict, _flatten=_flatten, exclude_through=exclude_through
        )
    expr = self.build_select_expression()
    rows = await self.database.fetch_all(expr)
    if not rows:
        return []
    alias_resolver = ReverseAliasResolver(
        select_related=self._select_related,
        excludable=self._excludable,
        model_cls=self.model_cls,  # type: ignore
        exclude_through=exclude_through,
    )
    column_map = alias_resolver.resolve_columns(
        columns_names=list(cast(LegacyRow, rows[0]).keys())
    )
    result = [
        {column_map.get(k): v for k, v in dict(x).items() if k in column_map}
        for x in rows
    ]
    if _as_dict:
        return result
    if _flatten and self._excludable.include_entry_count() != 1:
        raise QueryDefinitionError(
            "You cannot flatten values_list if more than one field is selected!"
        )
    tuple_result = [tuple(x.values()) for x in result]
    return tuple_result if not _flatten else [x[0] for x in tuple_result]

values_list(fields=None, flatten=False, exclude_through=False) async

Return a list of tuples with column values in order of the fields passed or all fields from queried models.

When one field is passed you can flatten the list of tuples into list of values of that single field.

To filter for given row use filter/exclude methods before values, to limit number of rows use limit/offset or paginate before values.

Note that it always return a list even for one row from database.

:param exclude_through: flag if through models should be excluded :type exclude_through: bool :param fields: field name or list of field names to extract from db :type fields: Union[str, List[str]] :param flatten: when one field is passed you can flatten the list of tuples :type flatten: bool

Source code in ormar/queryset/queryset.py

async def values_list(
    self,
    fields: Union[List, str, Set, Dict, None] = None,
    flatten: bool = False,
    exclude_through: bool = False,
) -> List:
    """
    Return a list of tuples with column values in order of the fields passed or
    all fields from queried models.

    When one field is passed you can flatten the list of tuples into list of values
    of that single field.

    To filter for given row use filter/exclude methods before values,
    to limit number of rows use limit/offset or paginate before values.

    Note that it always return a list even for one row from database.

    :param exclude_through: flag if through models should be excluded
    :type exclude_through: bool
    :param fields: field name or list of field names to extract from db
    :type fields: Union[str, List[str]]
    :param flatten: when one field is passed you can flatten the list of tuples
    :type flatten: bool
    """
    return await self.values(
        fields=fields,
        exclude_through=exclude_through,
        _as_dict=False,
        _flatten=flatten,
    )

ReferentialAction

Bases: Enum

Because the database management system(DBMS) enforces referential constraints, it must ensure data integrity if rows in a referenced table are to be deleted (or updated).

If dependent rows in referencing tables still exist, those references have to be considered.

SQL specifies 5 different referential actions that shall take place in such occurrences.

Source code in ormar/fields/referential_actions.py

class ReferentialAction(Enum):
    """
    Because the database management system(DBMS) enforces referential constraints,
    it must ensure data integrity
    if rows in a referenced table are to be deleted (or updated).

    If dependent rows in referencing tables still exist,
    those references have to be considered.

    SQL specifies 5 different referential actions
    that shall take place in such occurrences.
    """

    CASCADE: str = "CASCADE"
    RESTRICT: str = "RESTRICT"
    SET_NULL: str = "SET NULL"
    SET_DEFAULT: str = "SET DEFAULT"
    DO_NOTHING: str = "NO ACTION"

RelationType

Bases: Enum

Different types of relations supported by ormar:

• ForeignKey = PRIMARY
• reverse ForeignKey = REVERSE
• ManyToMany = MULTIPLE

Source code in ormar/relations/relation.py

class RelationType(Enum):
    """
    Different types of relations supported by ormar:

    *  ForeignKey = PRIMARY
    *  reverse ForeignKey = REVERSE
    *  ManyToMany = MULTIPLE
    """

    PRIMARY = 1
    REVERSE = 2
    MULTIPLE = 3
    THROUGH = 4

Signal

Signal that notifies all receiver functions. In ormar used by models to send pre_save, post_save etc. signals.

Source code in ormar/signals/signal.py

class Signal:
    """
    Signal that notifies all receiver functions.
    In ormar used by models to send pre_save, post_save etc. signals.
    """

    def __init__(self) -> None:
        self._receivers: Dict[Union[int, Tuple[int, int]], Callable] = {}

    def connect(self, receiver: Callable) -> None:
        """
        Connects given receiver function to the signal.

        :raises SignalDefinitionError: if receiver is not callable
        or not accept **kwargs
        :param receiver: receiver function
        :type receiver: Callable
        """
        if not callable(receiver):
            raise SignalDefinitionError("Signal receivers must be callable.")
        if not callable_accepts_kwargs(receiver):
            raise SignalDefinitionError(
                "Signal receivers must accept **kwargs argument."
            )
        new_receiver_key = make_id(receiver)
        if new_receiver_key not in self._receivers:
            self._receivers[new_receiver_key] = receiver

    def disconnect(self, receiver: Callable) -> bool:
        """
        Removes the receiver function from the signal.

        :param receiver: receiver function
        :type receiver: Callable
        :return: flag if receiver was removed
        :rtype: bool
        """
        new_receiver_key = make_id(receiver)
        receiver_func: Union[Callable, None] = self._receivers.pop(
            new_receiver_key, None
        )
        return True if receiver_func is not None else False

    async def send(self, sender: Type["Model"], **kwargs: Any) -> None:
        """
        Notifies all receiver functions with given kwargs
        :param sender: model that sends the signal
        :type sender: Type["Model"]
        :param kwargs: arguments passed to receivers
        :type kwargs: Any
        """
        receivers = [
            receiver_func(sender=sender, **kwargs)
            for receiver_func in self._receivers.values()
        ]
        await asyncio.gather(*receivers)

connect(receiver)

Connects given receiver function to the signal.

:raises SignalDefinitionError: if receiver is not callable or not accept **kwargs :param receiver: receiver function :type receiver: Callable

Source code in ormar/signals/signal.py

def connect(self, receiver: Callable) -> None:
    """
    Connects given receiver function to the signal.

    :raises SignalDefinitionError: if receiver is not callable
    or not accept **kwargs
    :param receiver: receiver function
    :type receiver: Callable
    """
    if not callable(receiver):
        raise SignalDefinitionError("Signal receivers must be callable.")
    if not callable_accepts_kwargs(receiver):
        raise SignalDefinitionError(
            "Signal receivers must accept **kwargs argument."
        )
    new_receiver_key = make_id(receiver)
    if new_receiver_key not in self._receivers:
        self._receivers[new_receiver_key] = receiver

disconnect(receiver)

Removes the receiver function from the signal.

:param receiver: receiver function :type receiver: Callable :return: flag if receiver was removed :rtype: bool

Source code in ormar/signals/signal.py

def disconnect(self, receiver: Callable) -> bool:
    """
    Removes the receiver function from the signal.

    :param receiver: receiver function
    :type receiver: Callable
    :return: flag if receiver was removed
    :rtype: bool
    """
    new_receiver_key = make_id(receiver)
    receiver_func: Union[Callable, None] = self._receivers.pop(
        new_receiver_key, None
    )
    return True if receiver_func is not None else False

send(sender, **kwargs) async

Notifies all receiver functions with given kwargs :param sender: model that sends the signal :type sender: Type["Model"] :param kwargs: arguments passed to receivers :type kwargs: Any

Source code in ormar/signals/signal.py

async def send(self, sender: Type["Model"], **kwargs: Any) -> None:
    """
    Notifies all receiver functions with given kwargs
    :param sender: model that sends the signal
    :type sender: Type["Model"]
    :param kwargs: arguments passed to receivers
    :type kwargs: Any
    """
    receivers = [
        receiver_func(sender=sender, **kwargs)
        for receiver_func in self._receivers.values()
    ]
    await asyncio.gather(*receivers)

SmallInteger

Bases: Integer, int

SmallInteger field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class SmallInteger(Integer, int):
    """
    SmallInteger field factory that construct Field classes and populated their values.
    """

    _type = int
    _sample = 0

    def __new__(  # type: ignore
        cls,
        *,
        minimum: Optional[int] = None,
        maximum: Optional[int] = None,
        multiple_of: Optional[int] = None,
        **kwargs: Any
    ) -> BaseField:
        autoincrement = kwargs.pop("autoincrement", None)
        autoincrement = (
            autoincrement
            if autoincrement is not None
            else kwargs.get("primary_key", False)
        )
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        kwargs["ge"] = kwargs["minimum"]
        kwargs["le"] = kwargs["maximum"]
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.SmallInteger()

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.SmallInteger()

String

Bases: ModelFieldFactory, str

String field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class String(ModelFieldFactory, str):
    """
    String field factory that construct Field classes and populated their values.
    """

    _type = str
    _sample = "string"

    def __new__(  # type: ignore # noqa CFQ002
        cls,
        *,
        max_length: int,
        min_length: Optional[int] = None,
        regex: Optional[str] = None,
        **kwargs: Any
    ) -> BaseField:  # type: ignore
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.String(length=kwargs.get("max_length"))

    @classmethod
    def validate(cls, **kwargs: Any) -> None:
        """
        Used to validate if all required parameters on a given field type are set.
        :param kwargs: all params passed during construction
        :type kwargs: Any
        """
        max_length = kwargs.get("max_length", None)
        if max_length <= 0:
            raise ModelDefinitionError(
                "Parameter max_length is required for field String"
            )

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.String(length=kwargs.get("max_length"))

validate(**kwargs) classmethod

Used to validate if all required parameters on a given field type are set. :param kwargs: all params passed during construction :type kwargs: Any

Source code in ormar/fields/model_fields.py

@classmethod
def validate(cls, **kwargs: Any) -> None:
    """
    Used to validate if all required parameters on a given field type are set.
    :param kwargs: all params passed during construction
    :type kwargs: Any
    """
    max_length = kwargs.get("max_length", None)
    if max_length <= 0:
        raise ModelDefinitionError(
            "Parameter max_length is required for field String"
        )

Text

Bases: ModelFieldFactory, str

Text field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class Text(ModelFieldFactory, str):
    """
    Text field factory that construct Field classes and populated their values.
    """

    _type = str
    _sample = "text"

    def __new__(cls, **kwargs: Any) -> BaseField:  # type: ignore
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.Text()

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.Text()

Time

Bases: ModelFieldFactory, time

Time field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class Time(ModelFieldFactory, datetime.time):
    """
    Time field factory that construct Field classes and populated their values.
    """

    _type = datetime.time
    _sample = "time"

    def __new__(  # type: ignore # noqa CFQ002
        cls, *, timezone: bool = False, **kwargs: Any
    ) -> BaseField:  # type: ignore
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }
        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        return sqlalchemy.Time(timezone=kwargs.get("timezone", False))

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    return sqlalchemy.Time(timezone=kwargs.get("timezone", False))

UUID

Bases: ModelFieldFactory, UUID

UUID field factory that construct Field classes and populated their values.

Source code in ormar/fields/model_fields.py

class UUID(ModelFieldFactory, uuid.UUID):
    """
    UUID field factory that construct Field classes and populated their values.
    """

    _type = uuid.UUID
    _sample = "uuid"

    def __new__(  # type: ignore # noqa CFQ002
        cls, *, uuid_format: str = "hex", **kwargs: Any
    ) -> BaseField:
        kwargs = {
            **kwargs,
            **{
                k: v
                for k, v in locals().items()
                if k not in ["cls", "__class__", "kwargs"]
            },
        }

        return super().__new__(cls, **kwargs)

    @classmethod
    def get_column_type(cls, **kwargs: Any) -> Any:
        """
        Return proper type of db column for given field type.
        Accepts required and optional parameters that each column type accepts.

        :param kwargs: key, value pairs of sqlalchemy options
        :type kwargs: Any
        :return: initialized column with proper options
        :rtype: sqlalchemy Column
        """
        uuid_format = kwargs.get("uuid_format", "hex")
        return sqlalchemy_uuid.UUID(uuid_format=uuid_format)

get_column_type(**kwargs) classmethod

Return proper type of db column for given field type. Accepts required and optional parameters that each column type accepts.

:param kwargs: key, value pairs of sqlalchemy options :type kwargs: Any :return: initialized column with proper options :rtype: sqlalchemy Column

Source code in ormar/fields/model_fields.py

@classmethod
def get_column_type(cls, **kwargs: Any) -> Any:
    """
    Return proper type of db column for given field type.
    Accepts required and optional parameters that each column type accepts.

    :param kwargs: key, value pairs of sqlalchemy options
    :type kwargs: Any
    :return: initialized column with proper options
    :rtype: sqlalchemy Column
    """
    uuid_format = kwargs.get("uuid_format", "hex")
    return sqlalchemy_uuid.UUID(uuid_format=uuid_format)

UniqueColumns

Bases: UniqueConstraint

Subclass of sqlalchemy.UniqueConstraint. Used to avoid importing anything from sqlalchemy by user.

Source code in ormar/fields/constraints.py

class UniqueColumns(UniqueConstraint):
    """
    Subclass of sqlalchemy.UniqueConstraint.
    Used to avoid importing anything from sqlalchemy by user.
    """

ForeignKey(to, *, name=None, unique=False, nullable=True, related_name=None, virtual=False, onupdate=None, ondelete=None, **kwargs)

Despite a name it's a function that returns constructed ForeignKeyField. This function is actually used in model declaration (as ormar.ForeignKey(ToModel)).

Accepts number of relation setting parameters as well as all BaseField ones.

:param to: target related ormar Model :type to: Model class :param name: name of the database field - later called alias :type name: str :param unique: parameter passed to sqlalchemy.ForeignKey, unique flag :type unique: bool :param nullable: marks field as optional/ required :type nullable: bool :param related_name: name of reversed FK relation populated for you on to model :type related_name: str :param virtual: marks if relation is virtual. It is for reversed FK and auto generated FK on through model in Many2Many relations. :type virtual: bool :param onupdate: parameter passed to sqlalchemy.ForeignKey. How to treat child rows on update of parent (the one where FK is defined) model. :type onupdate: Union[ReferentialAction, str] :param ondelete: parameter passed to sqlalchemy.ForeignKey. How to treat child rows on delete of parent (the one where FK is defined) model. :type ondelete: Union[ReferentialAction, str] :param kwargs: all other args to be populated by BaseField :type kwargs: Any :return: ormar ForeignKeyField with relation to selected model :rtype: ForeignKeyField

Source code in ormar/fields/foreign_key.py

def ForeignKey(  # type: ignore # noqa CFQ002
    to: Union[Type["T"], "ForwardRef"],
    *,
    name: Optional[str] = None,
    unique: bool = False,
    nullable: bool = True,
    related_name: Optional[str] = None,
    virtual: bool = False,
    onupdate: Union[ReferentialAction, str, None] = None,
    ondelete: Union[ReferentialAction, str, None] = None,
    **kwargs: Any,
) -> "T":
    """
    Despite a name it's a function that returns constructed ForeignKeyField.
    This function is actually used in model declaration (as ormar.ForeignKey(ToModel)).

    Accepts number of relation setting parameters as well as all BaseField ones.

    :param to: target related ormar Model
    :type to: Model class
    :param name: name of the database field - later called alias
    :type name: str
    :param unique: parameter passed to sqlalchemy.ForeignKey, unique flag
    :type unique: bool
    :param nullable: marks field as optional/ required
    :type nullable: bool
    :param related_name: name of reversed FK relation populated for you on to model
    :type related_name: str
    :param virtual: marks if relation is virtual.
    It is for reversed FK and auto generated FK on through model in Many2Many relations.
    :type virtual: bool
    :param onupdate: parameter passed to sqlalchemy.ForeignKey.
    How to treat child rows on update of parent (the one where FK is defined) model.
    :type onupdate: Union[ReferentialAction, str]
    :param ondelete: parameter passed to sqlalchemy.ForeignKey.
    How to treat child rows on delete of parent (the one where FK is defined) model.
    :type ondelete: Union[ReferentialAction, str]
    :param kwargs: all other args to be populated by BaseField
    :type kwargs: Any
    :return: ormar ForeignKeyField with relation to selected model
    :rtype: ForeignKeyField
    """

    onupdate = validate_referential_action(action=onupdate)
    ondelete = validate_referential_action(action=ondelete)

    owner = kwargs.pop("owner", None)
    self_reference = kwargs.pop("self_reference", False)

    orders_by = kwargs.pop("orders_by", None)
    related_orders_by = kwargs.pop("related_orders_by", None)

    skip_reverse = kwargs.pop("skip_reverse", False)
    skip_field = kwargs.pop("skip_field", False)

    sql_nullable = kwargs.pop("sql_nullable", None)
    sql_nullable = nullable if sql_nullable is None else sql_nullable

    index = kwargs.pop("index", False)

    validate_not_allowed_fields(kwargs)
    pk_only_model = None
    if to.__class__ == ForwardRef:
        __type__ = to if not nullable else Optional[to]
        constraints: List = []
        column_type = None
    else:
        (
            __type__,
            constraints,
            column_type,
            pk_only_model,
        ) = populate_fk_params_based_on_to_model(
            to=to,  # type: ignore
            nullable=nullable,
            ondelete=ondelete,
            onupdate=onupdate,
        )

    namespace = dict(
        __type__=__type__,
        to=to,
        to_pk_only=pk_only_model,
        through=None,
        alias=name,
        name=kwargs.pop("real_name", None),
        nullable=nullable,
        sql_nullable=sql_nullable,
        constraints=constraints,
        unique=unique,
        column_type=column_type,
        related_name=related_name,
        virtual=virtual,
        primary_key=False,
        index=index,
        default=None,
        server_default=None,
        onupdate=onupdate,
        ondelete=ondelete,
        owner=owner,
        self_reference=self_reference,
        is_relation=True,
        orders_by=orders_by,
        related_orders_by=related_orders_by,
        skip_reverse=skip_reverse,
        skip_field=skip_field,
    )

    Field = type("ForeignKey", (ForeignKeyField, BaseField), {})
    return Field(**namespace)

ManyToMany(to, through=None, *, name=None, unique=False, virtual=False, **kwargs)

Despite a name it's a function that returns constructed ManyToManyField. This function is actually used in model declaration (as ormar.ManyToMany(ToModel, through=ThroughModel)).

Accepts number of relation setting parameters as well as all BaseField ones.

:param to: target related ormar Model :type to: Model class :param through: through model for m2m relation :type through: Model class :param name: name of the database field - later called alias :type name: str :param unique: parameter passed to sqlalchemy.ForeignKey, unique flag :type unique: bool :param virtual: marks if relation is virtual. It is for reversed FK and auto generated FK on through model in Many2Many relations. :type virtual: bool :param kwargs: all other args to be populated by BaseField :type kwargs: Any :return: ormar ManyToManyField with m2m relation to selected model :rtype: ManyToManyField

Source code in ormar/fields/many_to_many.py

def ManyToMany(  # type: ignore
    to: Union[Type["T"], "ForwardRef"],
    through: Optional[Union[Type["T"], "ForwardRef"]] = None,
    *,
    name: Optional[str] = None,
    unique: bool = False,
    virtual: bool = False,
    **kwargs: Any,
) -> "RelationProxy[T]":
    """
    Despite a name it's a function that returns constructed ManyToManyField.
    This function is actually used in model declaration
    (as ormar.ManyToMany(ToModel, through=ThroughModel)).

    Accepts number of relation setting parameters as well as all BaseField ones.

    :param to: target related ormar Model
    :type to: Model class
    :param through: through model for m2m relation
    :type through: Model class
    :param name: name of the database field - later called alias
    :type name: str
    :param unique: parameter passed to sqlalchemy.ForeignKey, unique flag
    :type unique: bool
    :param virtual: marks if relation is virtual.
    It is for reversed FK and auto generated FK on through model in Many2Many relations.
    :type virtual: bool
    :param kwargs: all other args to be populated by BaseField
    :type kwargs: Any
    :return: ormar ManyToManyField with m2m relation to selected model
    :rtype: ManyToManyField
    """
    related_name = kwargs.pop("related_name", None)
    nullable = kwargs.pop("nullable", True)

    owner = kwargs.pop("owner", None)
    self_reference = kwargs.pop("self_reference", False)

    orders_by = kwargs.pop("orders_by", None)
    related_orders_by = kwargs.pop("related_orders_by", None)

    skip_reverse = kwargs.pop("skip_reverse", False)
    skip_field = kwargs.pop("skip_field", False)

    through_relation_name = kwargs.pop("through_relation_name", None)
    through_reverse_relation_name = kwargs.pop("through_reverse_relation_name", None)

    if through is not None and through.__class__ != ForwardRef:
        forbid_through_relations(cast(Type["Model"], through))

    validate_not_allowed_fields(kwargs)
    pk_only_model = None
    if to.__class__ == ForwardRef:
        __type__ = (
            Union[to, List[to]]  # type: ignore
            if not nullable
            else Optional[Union[to, List[to]]]  # type: ignore
        )
        column_type = None
    else:
        __type__, column_type, pk_only_model = populate_m2m_params_based_on_to_model(
            to=to, nullable=nullable  # type: ignore
        )
    namespace = dict(
        __type__=__type__,
        to=to,
        to_pk_only=pk_only_model,
        through=through,
        alias=name,
        name=name,
        nullable=nullable,
        unique=unique,
        column_type=column_type,
        related_name=related_name,
        virtual=virtual,
        primary_key=False,
        index=False,
        default=None,
        server_default=None,
        owner=owner,
        self_reference=self_reference,
        is_relation=True,
        is_multi=True,
        orders_by=orders_by,
        related_orders_by=related_orders_by,
        skip_reverse=skip_reverse,
        skip_field=skip_field,
        through_relation_name=through_relation_name,
        through_reverse_relation_name=through_reverse_relation_name,
    )

    Field = type("ManyToMany", (ManyToManyField, BaseField), {})
    return Field(**namespace)

and_(*args, **kwargs)

Construct and filter from nested groups and keyword arguments

:param args: nested filter groups :type args: Tuple[FilterGroup] :param kwargs: fields names and proper value types :type kwargs: Any :return: FilterGroup ready to be resolved :rtype: ormar.queryset.clause.FilterGroup

Source code in ormar/queryset/clause.py

def and_(*args: FilterGroup, **kwargs: Any) -> FilterGroup:
    """
    Construct and filter from nested groups and keyword arguments

    :param args: nested filter groups
    :type args: Tuple[FilterGroup]
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: FilterGroup ready to be resolved
    :rtype: ormar.queryset.clause.FilterGroup
    """
    return FilterGroup(_filter_type=FilterType.AND, *args, **kwargs)

or_(*args, **kwargs)

Construct or filter from nested groups and keyword arguments

:param args: nested filter groups :type args: Tuple[FilterGroup] :param kwargs: fields names and proper value types :type kwargs: Any :return: FilterGroup ready to be resolved :rtype: ormar.queryset.clause.FilterGroup

Source code in ormar/queryset/clause.py

def or_(*args: FilterGroup, **kwargs: Any) -> FilterGroup:
    """
    Construct or filter from nested groups and keyword arguments

    :param args: nested filter groups
    :type args: Tuple[FilterGroup]
    :param kwargs: fields names and proper value types
    :type kwargs: Any
    :return: FilterGroup ready to be resolved
    :rtype: ormar.queryset.clause.FilterGroup
    """
    return FilterGroup(_filter_type=FilterType.OR, *args, **kwargs)

post_bulk_update(senders)

Connect given function to all senders for post_bulk_update signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def post_bulk_update(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for post_bulk_update signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="post_bulk_update", senders=senders)

post_delete(senders)

Connect given function to all senders for post_delete signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def post_delete(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for post_delete signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="post_delete", senders=senders)

post_relation_add(senders)

Connect given function to all senders for post_relation_add signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def post_relation_add(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for post_relation_add signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="post_relation_add", senders=senders)

post_relation_remove(senders)

Connect given function to all senders for post_relation_remove signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def post_relation_remove(
    senders: Union[Type["Model"], List[Type["Model"]]]
) -> Callable:
    """
    Connect given function to all senders for post_relation_remove signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="post_relation_remove", senders=senders)

post_save(senders)

Connect given function to all senders for post_save signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def post_save(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for post_save signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="post_save", senders=senders)

post_update(senders)

Connect given function to all senders for post_update signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def post_update(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for post_update signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="post_update", senders=senders)

pre_delete(senders)

Connect given function to all senders for pre_delete signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def pre_delete(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for pre_delete signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="pre_delete", senders=senders)

pre_relation_add(senders)

Connect given function to all senders for pre_relation_add signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def pre_relation_add(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for pre_relation_add signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="pre_relation_add", senders=senders)

pre_relation_remove(senders)

Connect given function to all senders for pre_relation_remove signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def pre_relation_remove(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for pre_relation_remove signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="pre_relation_remove", senders=senders)

pre_save(senders)

Connect given function to all senders for pre_save signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def pre_save(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for pre_save signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="pre_save", senders=senders)

pre_update(senders)

Connect given function to all senders for pre_update signal.

:param senders: one or a list of "Model" classes that should have the signal receiver registered :type senders: Union[Type["Model"], List[Type["Model"]]] :return: returns the original function untouched :rtype: Callable

Source code in ormar/decorators/signals.py

def pre_update(senders: Union[Type["Model"], List[Type["Model"]]]) -> Callable:
    """
    Connect given function to all senders for pre_update signal.

    :param senders: one or a list of "Model" classes
    that should have the signal receiver registered
    :type senders: Union[Type["Model"], List[Type["Model"]]]
    :return: returns the original function untouched
    :rtype: Callable
    """
    return receiver(signal="pre_update", senders=senders)