QuerySet class

edgy.QuerySet

QuerySet(model_class=None, database=None, filter_clauses=None, or_clauses=None, select_related=None, prefetch_related=None, limit_count=None, limit_offset=None, order_by=None, group_by=None, distinct_on=None, only_fields=None, defer_fields=None, embed_parent=None, using_schema=None, table=None, exclude_secrets=False)

Bases: BaseQuerySet, QuerySetProtocol

QuerySet object used for query retrieving.

PARAMETER	DESCRIPTION
model_class	TYPE: Union[Type[Model], None] DEFAULT: None
database	TYPE: Union[Database, None] DEFAULT: None
filter_clauses	TYPE: Any DEFAULT: None
or_clauses	TYPE: Any DEFAULT: None
select_related	TYPE: Any DEFAULT: None
prefetch_related	TYPE: Any DEFAULT: None
limit_count	TYPE: Any DEFAULT: None
limit_offset	TYPE: Any DEFAULT: None
order_by	TYPE: Any DEFAULT: None
group_by	TYPE: Any DEFAULT: None
distinct_on	TYPE: Optional[Sequence[str]] DEFAULT: None
only_fields	TYPE: Any DEFAULT: None
defer_fields	TYPE: Any DEFAULT: None
embed_parent	TYPE: Any DEFAULT: None
using_schema	TYPE: Any DEFAULT: None
table	TYPE: Any DEFAULT: None
exclude_secrets	TYPE: Any DEFAULT: False

Source code in edgy/core/db/querysets/base.py

def __init__(
    self,
    model_class: Union[Type["Model"], None] = None,
    database: Union["Database", None] = None,
    filter_clauses: Any = None,
    or_clauses: Any = None,
    select_related: Any = None,
    prefetch_related: Any = None,
    limit_count: Any = None,
    limit_offset: Any = None,
    order_by: Any = None,
    group_by: Any = None,
    distinct_on: Optional[Sequence[str]] = None,
    only_fields: Any = None,
    defer_fields: Any = None,
    embed_parent: Any = None,
    using_schema: Any = None,
    table: Any = None,
    exclude_secrets: Any = False,
) -> None:
    super().__init__(model_class=model_class)
    self.model_class = cast("Type[Model]", model_class)
    self.filter_clauses = [] if filter_clauses is None else filter_clauses
    self.or_clauses = [] if or_clauses is None else or_clauses
    self.limit_count = limit_count
    self._select_related = [] if select_related is None else select_related
    self._prefetch_related = [] if prefetch_related is None else prefetch_related
    self._offset = limit_offset
    self._order_by = [] if order_by is None else order_by
    self._group_by = [] if group_by is None else group_by
    self.distinct_on = distinct_on
    self._only = [] if only_fields is None else only_fields
    self._defer = [] if defer_fields is None else defer_fields
    self._expression = None
    self._cache = None
    self.embed_parent = embed_parent
    self.using_schema = using_schema
    self._exclude_secrets = exclude_secrets or False
    self.extra: Dict[str, Any] = {}

    # Making sure the queryset always starts without any schema associated unless specified

    if table is not None:
        self.table = table
    if database is not None:
        self.database = database

model_class instance-attribute

model_class = cast('Type[Model]', model_class)

database instance-attribute

database = database

table instance-attribute

table = table

pkcolumns property

pkcolumns

filter_clauses instance-attribute

filter_clauses = [] if filter_clauses is None else filter_clauses

or_clauses instance-attribute

or_clauses = [] if or_clauses is None else or_clauses

limit_count instance-attribute

limit_count = limit_count

_select_related instance-attribute

_select_related = [] if select_related is None else select_related

_prefetch_related instance-attribute

_prefetch_related = [] if prefetch_related is None else prefetch_related

_offset instance-attribute

_offset = limit_offset

_order_by instance-attribute

_order_by = [] if order_by is None else order_by

_group_by instance-attribute

_group_by = [] if group_by is None else group_by

distinct_on instance-attribute

distinct_on = distinct_on

_only instance-attribute

_only = [] if only_fields is None else only_fields

_defer instance-attribute

_defer = [] if defer_fields is None else defer_fields

_expression instance-attribute

_expression = None

_cache instance-attribute

_cache = None

embed_parent instance-attribute

embed_parent = embed_parent

using_schema instance-attribute

using_schema = using_schema

_exclude_secrets instance-attribute

_exclude_secrets = exclude_secrets or False

extra instance-attribute

extra = {}

sql property writable

sql

execute async

execute()

Source code in edgy/core/db/querysets/protocols.py

async def execute(self) -> typing.Any:
    raise NotImplementedError()

_extract_model_references

_extract_model_references(extracted_values, model_class)

Exracts any possible model references from the EdgyModel and returns a dictionary.

PARAMETER	DESCRIPTION
extracted_values	TYPE: Any
model_class	TYPE: Optional[Type[Model]]

Source code in edgy/core/utils/models.py

def _extract_model_references(self, extracted_values: Any, model_class: Optional[Type["Model"]]) -> Any:
    """
    Exracts any possible model references from the EdgyModel and returns a dictionary.
    """
    model_references = {
        name: extracted_values.get(name, None)
        for name in model_class.meta.model_references.keys()  # type: ignore
        if extracted_values.get(name)
    }
    return model_references

_extract_values_from_field

_extract_values_from_field(extracted_values, model_class=None, is_update=False, is_partial=False)

Extracts all the default values from the given fields and returns the raw value corresponding to each field.

PARAMETER	DESCRIPTION
extracted_values	TYPE: Any
model_class	TYPE: Optional[Model] DEFAULT: None
is_update	TYPE: bool DEFAULT: False
is_partial	TYPE: bool DEFAULT: False

Source code in edgy/core/utils/models.py

def _extract_values_from_field(
    self,
    extracted_values: Any,
    model_class: Optional["Model"] = None,
    is_update: bool = False,
    is_partial: bool = False,
) -> Any:
    """
    Extracts all the default values from the given fields and returns the raw
    value corresponding to each field.
    """
    model_cls = model_class or self
    validated: Dict[str, Any] = {}
    # phase 1: transform when required
    if model_cls.meta.input_modifying_fields:
        extracted_values = {**extracted_values}
        for field_name in model_cls.meta.input_modifying_fields:
            model_cls.fields[field_name].modify_input(field_name, extracted_values)
    # phase 2: validate fields and set defaults for readonly
    for field_name, field in model_cls.fields.items():  # type: ignore
        if not is_partial and field.read_only:
            if field.has_default():
                if not is_update:
                    validated.update(field.get_default_values(field_name, validated))
                else:
                    # For datetimes with `auto_now` and `auto_now_add`
                    if not _has_auto_now_add(field):
                        validated.update(field.get_default_values(field_name, validated))
            continue
        if field_name in extracted_values:
            item = extracted_values[field_name]
            for sub_name, value in field.clean(field_name, item).items():
                if sub_name in validated:
                    raise ValueError(f"value set twice for key: {sub_name}")
                validated[sub_name] = value

    # phase 3: set defaults for the rest if not an update
    if not is_partial:
        for field_name, field in model_cls.fields.items():  # type: ignore
            # we need a second run
            if not field.read_only and field_name not in validated:
                if field.has_default():
                    validated.update(field.get_default_values(field_name, validated))
    # Update with any ModelRef
    validated.update(self._extract_model_references(extracted_values, model_cls))
    return validated

_update_auto_now_fields

_update_auto_now_fields(values, fields)

Updates the auto_now fields

PARAMETER	DESCRIPTION
values	TYPE: Any
fields	TYPE: Any

Source code in edgy/core/utils/models.py

def _update_auto_now_fields(self, values: Any, fields: Any) -> Any:
    """
    Updates the `auto_now` fields
    """
    for name, field in fields.items():
        if isinstance(field, BaseField) and _has_auto_now(field) and _is_datetime(field):
            values.update(field.get_default_values(name, values))
    return values

_resolve_value

_resolve_value(value)

PARAMETER	DESCRIPTION
value	TYPE: Any

Source code in edgy/core/utils/models.py

def _resolve_value(self, value: typing.Any) -> typing.Any:
    if isinstance(value, dict):
        return dumps(
            value,
            option=OPT_SERIALIZE_NUMPY | OPT_OMIT_MICROSECONDS,
        ).decode("utf-8")
    elif isinstance(value, Enum):
        return value.name
    return value

prefetch_related

prefetch_related(*prefetch)

Performs a reverse lookup for the foreignkeys. This is different from the select_related. Select related performs a follows the SQL foreign relation whereas the prefetch_related follows the python relations.

PARAMETER	DESCRIPTION
*prefetch	TYPE: Prefetch DEFAULT: ()

Source code in edgy/core/db/querysets/prefetch.py

def prefetch_related(self, *prefetch: Prefetch) -> "QuerySet":
    """
    Performs a reverse lookup for the foreignkeys. This is different
    from the select_related. Select related performs a follows the SQL foreign relation
    whereas the prefetch_related follows the python relations.
    """
    queryset: "QuerySet" = self._clone()

    if not isinstance(prefetch, (list, tuple)):
        prefetch = cast("List[Prefetch]", [prefetch])  # type: ignore

    if isinstance(prefetch, tuple):
        prefetch = list(prefetch)  # type: ignore

    if any(not isinstance(value, Prefetch) for value in prefetch):
        raise QuerySetError("The prefetch_related must have Prefetch type objects only.")

    prefetch = list(self._prefetch_related) + prefetch  # type: ignore
    queryset._prefetch_related = prefetch
    return queryset

using

using(schema)

Enables and switches the db schema.

Generates the registry object pointing to the desired schema using the same connection.

PARAMETER	DESCRIPTION
schema	TYPE: str

Source code in edgy/core/db/querysets/mixins.py

def using(self, schema: str) -> "QuerySet":
    """
    Enables and switches the db schema.

    Generates the registry object pointing to the desired schema
    using the same connection.
    """
    queryset = set_queryset_schema(self, self.model_class, value=schema)
    return queryset

using_with_db

using_with_db(connection_name, schema=None)

Enables and switches the database connection.

Generates the new queryset using the selected connection provided in the extra of the model registry.

PARAMETER	DESCRIPTION
connection_name	TYPE: str
schema	TYPE: Optional[str] DEFAULT: None

Source code in edgy/core/db/querysets/mixins.py

def using_with_db(self, connection_name: str, schema: Optional[str] = None) -> "QuerySet":
    """
    Enables and switches the database connection.

    Generates the new queryset using the selected connection provided in the extra of the model
    registry.
    """
    assert (
        connection_name in self.model_class.meta.registry.extra
    ), f"`{connection_name}` is not in the connections extra of the model`{self.model_class.__name__}` registry"

    connection: Type["Registry"] = self.model_class.meta.registry.extra[connection_name]
    if schema:
        return set_queryset_database(self, self.model_class, connection, schema)
    queryset = set_queryset_database(self, self.model_class, connection)
    return queryset

_build_order_by_expression

_build_order_by_expression(order_by, expression)

Builds the order by expression

PARAMETER	DESCRIPTION
order_by	TYPE: Any
expression	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def _build_order_by_expression(self, order_by: Any, expression: Any) -> Any:
    """Builds the order by expression"""
    order_by = list(map(self._prepare_order_by, order_by))
    expression = expression.order_by(*order_by)
    return expression

_build_group_by_expression

_build_group_by_expression(group_by, expression)

Builds the group by expression

PARAMETER	DESCRIPTION
group_by	TYPE: Any
expression	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def _build_group_by_expression(self, group_by: Any, expression: Any) -> Any:
    """Builds the group by expression"""
    group_by = list(map(self._prepare_group_by, group_by))
    expression = expression.group_by(*group_by)
    return expression

_build_filter_clauses_expression

_build_filter_clauses_expression(filter_clauses, expression)

Builds the filter clauses expression

PARAMETER	DESCRIPTION
filter_clauses	TYPE: Any
expression	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def _build_filter_clauses_expression(self, filter_clauses: Any, expression: Any) -> Any:
    """Builds the filter clauses expression"""
    if len(filter_clauses) == 1:
        clause = filter_clauses[0]
    else:
        clause = clauses_mod.and_(*filter_clauses)
    expression = expression.where(clause)
    return expression

_build_or_clauses_expression

_build_or_clauses_expression(or_clauses, expression)

Builds the filter clauses expression

PARAMETER	DESCRIPTION
or_clauses	TYPE: Any
expression	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def _build_or_clauses_expression(self, or_clauses: Any, expression: Any) -> Any:
    """Builds the filter clauses expression"""
    if len(or_clauses) == 1:
        clause = or_clauses[0]
    else:
        clause = clauses_mod.or_(*or_clauses)
    expression = expression.where(clause)
    return expression

_build_select_distinct

_build_select_distinct(distinct_on, expression)

Filters selects only specific fields. Leave empty to use simple distinct

PARAMETER	DESCRIPTION
distinct_on	TYPE: Optional[Sequence[str]]
expression	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def _build_select_distinct(self, distinct_on: Optional[Sequence[str]], expression: Any) -> Any:
    """Filters selects only specific fields. Leave empty to use simple distinct"""
    # using with columns is not supported by all databases
    if distinct_on:
        return expression.distinct(*map(self._prepare_fields_for_distinct, distinct_on))
    else:
        return expression.distinct()

_build_tables_select_from_relationship

_build_tables_select_from_relationship()

Builds the tables relationships and joins. When a table contains more than one foreign key pointing to the same destination table, a lookup for the related field is made to understand from which foreign key the table is looked up from.

Source code in edgy/core/db/querysets/base.py

def _build_tables_select_from_relationship(self) -> Any:
    """
    Builds the tables relationships and joins.
    When a table contains more than one foreign key pointing to the same
    destination table, a lookup for the related field is made to understand
    from which foreign key the table is looked up from.
    """
    queryset: "QuerySet" = self._clone()

    select_from = queryset.table
    tables = {select_from.name: select_from}

    # Select related
    for select_path in queryset._select_related:
        # For m2m relationships
        model_class = queryset.model_class
        select_from = queryset.table
        former_table = None
        while select_path:
            field_name = select_path.split("__", 1)[0]
            field = model_class.fields[field_name]
            if isinstance(field, RelationshipField):
                model_class, reverse_part, select_path = field.traverse_field(select_path)
            else:
                raise ValueError(f"{field_name}: invalid field type: {field!r}")
            if isinstance(field, BaseForeignKey):
                foreign_key = field
                reverse = False
            else:
                foreign_key = model_class.meta.fields_mapping[reverse_part]
                reverse = True
            if foreign_key.is_cross_db():
                raise NotImplementedError("We cannot cross databases yet, this feature is planned")
            table = model_class.table
            select_from = sqlalchemy.sql.join(  # type: ignore
                select_from,
                table,
                *self._select_from_relationship_clause_generator(select_from, foreign_key, table, reverse, former_table)
            )
            former_table = table
            tables[table.name] = table

    return tables.values(), select_from

_select_from_relationship_clause_generator staticmethod

_select_from_relationship_clause_generator(select_from, foreign_key, table, reverse, former_table=None)

PARAMETER	DESCRIPTION
select_from	TYPE: Any
foreign_key	TYPE: BaseForeignKey
table	TYPE: Any
reverse	TYPE: bool
former_table	TYPE: Any DEFAULT: None

Source code in edgy/core/db/querysets/base.py

@staticmethod
def _select_from_relationship_clause_generator(select_from: Any, foreign_key: BaseForeignKey, table: Any, reverse: bool, former_table: Any=None) -> Any:
    column_names = foreign_key.get_column_names(foreign_key.name)
    for col in column_names:
        if reverse:
            colname = foreign_key.from_fk_field_name(foreign_key.name, col)
        else:
            colname = col
        if former_table is None:
            former_table = select_from
        if reverse:
            yield getattr(former_table.c, colname) == getattr(table.c, col)
        else:
            yield getattr(former_table.c, colname) == getattr(table.c, foreign_key.from_fk_field_name(foreign_key.name, col))

_validate_only_and_defer

_validate_only_and_defer()

Source code in edgy/core/db/querysets/base.py

def _validate_only_and_defer(self) -> None:
    if self._only and self._defer:
        raise QuerySetError("You cannot use .only() and .defer() at the same time.")

_secret_recursive_names

_secret_recursive_names(model_class, columns=None)

Recursively gets the names of the fields excluding the secrets.

PARAMETER	DESCRIPTION
model_class	TYPE: Any
columns	TYPE: Union[List[str], None] DEFAULT: None

Source code in edgy/core/db/querysets/base.py

def _secret_recursive_names(self, model_class: Any, columns: Union[List[str], None] = None) -> List[str]:
    """
    Recursively gets the names of the fields excluding the secrets.
    """
    if columns is None:
        columns = []

    for name, field in model_class.fields.items():
        if isinstance(field, BaseForeignKey):
            # Making sure the foreign key is always added unless is a secret
            if not field.secret:
                columns.extend(field.get_column_names(name))
                columns.extend(self._secret_recursive_names(model_class=field.target, columns=columns))
            continue
        if not field.secret:
            columns.append(name)

    columns = list(set(columns))
    return columns

_build_select

_build_select()

Builds the query select based on the given parameters and filters.

Source code in edgy/core/db/querysets/base.py

def _build_select(self) -> Any:
    """
    Builds the query select based on the given parameters and filters.
    """
    queryset: "QuerySet" = self._clone()

    queryset._validate_only_and_defer()
    tables, select_from = queryset._build_tables_select_from_relationship()
    expression = sqlalchemy.sql.select(*tables)
    expression = expression.select_from(select_from)

    if queryset._only:
        expression = expression.with_only_columns(*queryset._only)

    if queryset._defer:
        columns = [column for column in select_from.columns if column.name not in queryset._defer]
        expression = expression.with_only_columns(*columns)

    if queryset._exclude_secrets:
        model_columns = queryset._secret_recursive_names(model_class=queryset.model_class)
        columns = [column for column in select_from.columns if column.name in model_columns]
        expression = expression.with_only_columns(*columns)

    if queryset.filter_clauses:
        expression = queryset._build_filter_clauses_expression(queryset.filter_clauses, expression=expression)

    if queryset.or_clauses:
        expression = queryset._build_or_clauses_expression(queryset.or_clauses, expression=expression)

    if queryset._order_by:
        expression = queryset._build_order_by_expression(queryset._order_by, expression=expression)

    if queryset.limit_count:
        expression = expression.limit(queryset.limit_count)

    if queryset._offset:
        expression = expression.offset(queryset._offset)

    if queryset._group_by:
        expression = queryset._build_group_by_expression(queryset._group_by, expression=expression)

    if queryset.distinct_on is not None:
        expression = queryset._build_select_distinct(queryset.distinct_on, expression=expression)

    queryset._expression = expression  # type: ignore
    return expression

_filter_query

_filter_query(exclude=False, or_=False, **kwargs)

PARAMETER	DESCRIPTION
exclude	TYPE: bool DEFAULT: False
or_	TYPE: bool DEFAULT: False
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def _filter_query(
    self,
    exclude: bool = False,
    or_: bool = False,
    **kwargs: Any,
) -> "QuerySet":
    from edgy.core.db.models import Model

    clauses = []
    filter_clauses = self.filter_clauses
    or_clauses = self.or_clauses
    select_related = list(self._select_related)
    prefetch_related = list(self._prefetch_related)

    # Making sure for queries we use the main class and not the proxy
    # And enable the parent
    if self.model_class.is_proxy_model:
        self.model_class = self.model_class.parent

    kwargs = clean_query_kwargs(self.model_class, kwargs)

    for key, value in kwargs.items():
        assert not isinstance(value, Model), f"should be parsed in clean: {key}: {value}"
        model_class, field_name, op, related_str, _ = crawl_relationship(self.model_class, key)
        if related_str and related_str not in select_related:
            select_related.append(related_str)
        column = model_class.table.columns[field_name]

        # Map the operation code onto SQLAlchemy's ColumnElement
        # https://docs.sqlalchemy.org/en/latest/core/sqlelement.html#sqlalchemy.sql.expression.ColumnElement
        op_attr = settings.filter_operators[op]
        has_escaped_character = False

        if op in ["contains", "icontains"]:
            has_escaped_character = any(c for c in self.ESCAPE_CHARACTERS if c in value)
            if has_escaped_character:
                # enable escape modifier
                for char in self.ESCAPE_CHARACTERS:
                    value = value.replace(char, f"\\{char}")
            value = f"%{value}%"

        clause = getattr(column, op_attr)(value)
        clause.modifiers["escape"] = "\\" if has_escaped_character else None
        clauses.append(clause)

    if exclude:
        if not or_:
            filter_clauses.append(clauses_mod.not_(clauses_mod.and_(*clauses)))
        else:
            or_clauses.append(clauses_mod.not_(clauses_mod.and_(*clauses)))
    else:
        if not or_:
            filter_clauses += clauses
        else:
            or_clauses += clauses

    return cast(
        "QuerySet",
        self.__class__(
            model_class=self.model_class,
            database=self._database,
            filter_clauses=filter_clauses,
            or_clauses=or_clauses,
            select_related=select_related,
            prefetch_related=prefetch_related,
            limit_count=self.limit_count,
            limit_offset=self._offset,
            order_by=self._order_by,
            only_fields=self._only,
            defer_fields=self._defer,
            embed_parent=self.embed_parent,
            table=self.table,
            exclude_secrets=self._exclude_secrets,
            using_schema=self.using_schema,
        ),
    )

_validate_kwargs

_validate_kwargs(**kwargs)

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def _validate_kwargs(self, **kwargs: Any) -> Any:
    return self._extract_values_from_field(kwargs, model_class=self.model_class)

_prepare_order_by

_prepare_order_by(order_by)

PARAMETER	DESCRIPTION
order_by	TYPE: str

Source code in edgy/core/db/querysets/base.py

def _prepare_order_by(self, order_by: str) -> Any:
    reverse = order_by.startswith("-")
    order_by = order_by.lstrip("-")
    order_col = self.table.columns[order_by]
    return order_col.desc() if reverse else order_col

_prepare_group_by

_prepare_group_by(group_by)

PARAMETER	DESCRIPTION
group_by	TYPE: str

Source code in edgy/core/db/querysets/base.py

def _prepare_group_by(self, group_by: str) -> Any:
    group_by = group_by.lstrip("-")
    group_col = self.table.columns[group_by]
    return group_col

_prepare_fields_for_distinct

_prepare_fields_for_distinct(distinct_on)

PARAMETER	DESCRIPTION
distinct_on	TYPE: str

Source code in edgy/core/db/querysets/base.py

def _prepare_fields_for_distinct(self, distinct_on: str) -> sqlalchemy.Column:
    return self.table.columns[distinct_on]

_clone

_clone()

Return a copy of the current QuerySet that's ready for another operation.

Source code in edgy/core/db/querysets/base.py

def _clone(self) -> Any:
    """
    Return a copy of the current QuerySet that's ready for another
    operation.
    """
    queryset = self.__class__.__new__(self.__class__)
    queryset.model_class = self.model_class

    # Making sure the registry schema takes precendent with
    # Any provided using
    if not self.model_class.meta.registry.db_schema:
        schema = get_schema()
        if self.using_schema is None and schema is not None:
            self.using_schema = schema
        queryset.model_class.table = self.model_class.build(self.using_schema)

    queryset.filter_clauses = copy.copy(self.filter_clauses)
    queryset.or_clauses = copy.copy(self.or_clauses)
    queryset.limit_count = copy.copy(self.limit_count)
    queryset._select_related = copy.copy(self._select_related)
    queryset._prefetch_related = copy.copy(self._prefetch_related)
    queryset._offset = copy.copy(self._offset)
    queryset._order_by = copy.copy(self._order_by)
    queryset._group_by = copy.copy(self._group_by)
    queryset.distinct_on = copy.copy(self.distinct_on)
    queryset._expression = copy.copy(self._expression)
    queryset.embed_parent = self.embed_parent
    queryset._only = copy.copy(self._only)
    queryset._defer = copy.copy(self._defer)
    queryset._database = self.database
    queryset.table = self.table
    queryset.extra = self.extra
    queryset._exclude_secrets = self._exclude_secrets
    queryset.using_schema = self.using_schema

    return queryset

__aiter__ async

__aiter__()

Source code in edgy/core/db/querysets/base.py

async def __aiter__(self) -> AsyncIterator[EdgyModel]:
    for value in await self:
        yield value

_set_query_expression

_set_query_expression(expression)

Sets the value of the sql property to the expression used.

PARAMETER	DESCRIPTION
expression	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def _set_query_expression(self, expression: Any) -> None:
    """
    Sets the value of the sql property to the expression used.
    """
    self.sql = expression
    self.model_class.raw_query = self.sql

_filter_or_exclude

_filter_or_exclude(clause=None, exclude=False, **kwargs)

Filters or excludes a given clause for a specific QuerySet.

PARAMETER	DESCRIPTION
clause	TYPE: Optional[BinaryExpression] DEFAULT: None
exclude	TYPE: bool DEFAULT: False
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def _filter_or_exclude(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    exclude: bool = False,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters or excludes a given clause for a specific QuerySet.
    """
    queryset: "QuerySet" = self._clone()
    if clause is None:
        return queryset._filter_query(exclude=exclude, **kwargs)

    queryset.filter_clauses.append(clause)
    return queryset

filter

filter(clause=None, **kwargs)

Filters the QuerySet by the given kwargs and clause.

PARAMETER	DESCRIPTION
clause	TYPE: Optional[BinaryExpression] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def filter(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters the QuerySet by the given kwargs and clause.
    """
    return self._filter_or_exclude(clause=clause, **kwargs)

or_

or_(clause=None, **kwargs)

Filters the QuerySet by the OR operand.

PARAMETER	DESCRIPTION
clause	TYPE: Optional[BinaryExpression] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def or_(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters the QuerySet by the OR operand.
    """
    queryset: "QuerySet" = self._clone()
    queryset = self.filter(clause=clause, or_=True, **kwargs)
    return queryset

and_

and_(clause=None, **kwargs)

Filters the QuerySet by the AND operand.

PARAMETER	DESCRIPTION
clause	TYPE: Optional[BinaryExpression] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def and_(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters the QuerySet by the AND operand.
    """
    queryset: "QuerySet" = self._clone()
    queryset = self.filter(clause=clause, **kwargs)
    return queryset

not_

not_(clause=None, **kwargs)

Filters the QuerySet by the NOT operand.

PARAMETER	DESCRIPTION
clause	TYPE: Optional[BinaryExpression] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def not_(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Filters the QuerySet by the NOT operand.
    """
    queryset: "QuerySet" = self._clone()
    queryset = queryset.exclude(clause=clause, **kwargs)
    return queryset

exclude

exclude(clause=None, **kwargs)

Exactly the same as the filter but for the exclude.

PARAMETER	DESCRIPTION
clause	TYPE: Optional[BinaryExpression] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def exclude(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Exactly the same as the filter but for the exclude.
    """
    queryset: "QuerySet" = self._clone()
    queryset = self._filter_or_exclude(clause=clause, exclude=True, **kwargs)
    return queryset

exclude_secrets

exclude_secrets(clause=None, **kwargs)

Excludes any field that contains the secret=True declared from being leaked.

PARAMETER	DESCRIPTION
clause	TYPE: Optional[BinaryExpression] DEFAULT: None
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def exclude_secrets(
    self,
    clause: Optional[sqlalchemy.sql.expression.BinaryExpression] = None,
    **kwargs: Any,
) -> "QuerySet":
    """
    Excludes any field that contains the `secret=True` declared from being leaked.
    """
    queryset: "QuerySet" = self._clone()
    queryset._exclude_secrets = True
    queryset = queryset.filter(clause=clause, **kwargs)
    return queryset

lookup

lookup(term)

Broader way of searching for a given term

PARAMETER	DESCRIPTION
term	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def lookup(self, term: Any) -> "QuerySet":
    """
    Broader way of searching for a given term
    """
    queryset: "QuerySet" = self._clone()
    if not term:
        return queryset

    filter_clauses = list(queryset.filter_clauses)
    value = f"%{term}%"

    search_fields = [
        name for name, field in queryset.model_class.fields.items() if isinstance(field, (CharField, TextField))
    ]
    search_clauses = [queryset.table.columns[name].ilike(value) for name in search_fields]

    if len(search_clauses) > 1:
        filter_clauses.append(sqlalchemy.sql.or_(*search_clauses))
    else:
        filter_clauses.extend(search_clauses)

    return queryset

order_by

order_by(*order_by)

Returns a QuerySet ordered by the given fields.

PARAMETER	DESCRIPTION
*order_by	TYPE: str DEFAULT: ()

Source code in edgy/core/db/querysets/base.py

def order_by(self, *order_by: str) -> "QuerySet":
    """
    Returns a QuerySet ordered by the given fields.
    """
    queryset: "QuerySet" = self._clone()
    queryset._order_by = order_by
    return queryset

limit

limit(limit_count)

Returns a QuerySet limited by.

PARAMETER	DESCRIPTION
limit_count	TYPE: int

Source code in edgy/core/db/querysets/base.py

def limit(self, limit_count: int) -> "QuerySet":
    """
    Returns a QuerySet limited by.
    """
    queryset: "QuerySet" = self._clone()
    queryset.limit_count = limit_count
    return queryset

offset

offset(offset)

Returns a Queryset limited by the offset.

PARAMETER	DESCRIPTION
offset	TYPE: int

Source code in edgy/core/db/querysets/base.py

def offset(self, offset: int) -> "QuerySet":
    """
    Returns a Queryset limited by the offset.
    """
    queryset: "QuerySet" = self._clone()
    queryset._offset = offset
    return queryset

group_by

group_by(*group_by)

Returns the values grouped by the given fields.

PARAMETER	DESCRIPTION
*group_by	TYPE: Sequence[str] DEFAULT: ()

Source code in edgy/core/db/querysets/base.py

def group_by(self, *group_by: Sequence[str]) -> "QuerySet":
    """
    Returns the values grouped by the given fields.
    """
    queryset: "QuerySet" = self._clone()
    queryset._group_by = group_by
    return queryset

distinct

distinct(*distinct_on)

Returns a queryset with distinct results.

PARAMETER	DESCRIPTION
*distinct_on	TYPE: str DEFAULT: ()

Source code in edgy/core/db/querysets/base.py

def distinct(self, *distinct_on: str) -> "QuerySet":
    """
    Returns a queryset with distinct results.
    """
    queryset: "QuerySet" = self._clone()
    queryset.distinct_on = distinct_on
    return queryset

only

only(*fields)

Returns a list of models with the selected only fields and always the primary key.

PARAMETER	DESCRIPTION
*fields	TYPE: Sequence[str] DEFAULT: ()

Source code in edgy/core/db/querysets/base.py

def only(self, *fields: Sequence[str]) -> "QuerySet":
    """
    Returns a list of models with the selected only fields and always the primary
    key.
    """
    only_fields = [sqlalchemy.text(field) for field in fields]
    missing = []
    if self.model_class.pknames:
        for pkname in self.model_class.pknames:
            if pkname not in fields:
                for pkcolumn in self.model_class.meta.get_columns_for_name(pkname):
                    missing.append(sqlalchemy.text(pkcolumn.key))
    else:
        for pkcolumn in self.model_class.pkcolumns:
            missing.append(sqlalchemy.text(pkcolumn.key))
    if missing:
        only_fields = missing + only_fields

    queryset: "QuerySet" = self._clone()
    queryset._only = only_fields
    return queryset

defer

defer(*fields)

Returns a list of models with the selected only fields and always the primary key.

PARAMETER	DESCRIPTION
*fields	TYPE: Sequence[str] DEFAULT: ()

Source code in edgy/core/db/querysets/base.py

def defer(self, *fields: Sequence[str]) -> "QuerySet":
    """
    Returns a list of models with the selected only fields and always the primary
    key.
    """
    queryset: "QuerySet" = self._clone()
    queryset._defer = fields
    return queryset

select_related

select_related(related)

Returns a QuerySet that will “follow” foreign-key relationships, selecting additional related-object data when it executes its query.

This is a performance booster which results in a single more complex query but means

later use of foreign-key relationships won’t require database queries.

PARAMETER	DESCRIPTION
related	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def select_related(self, related: Any) -> "QuerySet":
    """
    Returns a QuerySet that will “follow” foreign-key relationships, selecting additional
    related-object data when it executes its query.

    This is a performance booster which results in a single more complex query but means

    later use of foreign-key relationships won’t require database queries.
    """
    queryset: "QuerySet" = self._clone()
    if not isinstance(related, (list, tuple)):
        related = [related]

    related = list(queryset._select_related) + related
    queryset._select_related = related
    return queryset

values async

values(fields=None, exclude=None, exclude_none=False, flatten=False, **kwargs)

Returns the results in a python dictionary format.

PARAMETER	DESCRIPTION
fields	TYPE: Union[Sequence[str], str, None] DEFAULT: None
exclude	TYPE: Union[Sequence[str], Set[str]] DEFAULT: None
exclude_none	TYPE: bool DEFAULT: False
flatten	TYPE: bool DEFAULT: False
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def values(
    self,
    fields: Union[Sequence[str], str, None] = None,
    exclude: Union[Sequence[str], Set[str]] = None,
    exclude_none: bool = False,
    flatten: bool = False,
    **kwargs: Any,
) -> List[Any]:
    """
    Returns the results in a python dictionary format.
    """
    fields = fields or []
    queryset: "QuerySet" = self._clone()
    rows: List["Model"] = await queryset.all()

    if not isinstance(fields, list):
        raise QuerySetError(detail="Fields must be an iterable.")

    if not fields:
        rows = [row.model_dump(exclude=exclude, exclude_none=exclude_none) for row in rows]
    else:
        rows = [row.model_dump(exclude=exclude, exclude_none=exclude_none, include=fields) for row in rows]

    as_tuple = kwargs.pop("__as_tuple__", False)

    if not as_tuple:
        return rows

    if not flatten:
        rows = [tuple(row.values()) for row in rows]  # type: ignore
    else:
        try:
            rows = [row[fields[0]] for row in rows]  # type: ignore
        except KeyError:
            raise QuerySetError(detail=f"{fields[0]} does not exist in the results.") from None
    return rows

values_list async

values_list(fields=None, exclude=None, exclude_none=False, flat=False)

Returns the results in a python dictionary format.

PARAMETER	DESCRIPTION
fields	TYPE: Union[Sequence[str], str, None] DEFAULT: None
exclude	TYPE: Union[Sequence[str], Set[str]] DEFAULT: None
exclude_none	TYPE: bool DEFAULT: False
flat	TYPE: bool DEFAULT: False

Source code in edgy/core/db/querysets/base.py

async def values_list(
    self,
    fields: Union[Sequence[str], str, None] = None,
    exclude: Union[Sequence[str], Set[str]] = None,
    exclude_none: bool = False,
    flat: bool = False,
) -> List[Any]:
    """
    Returns the results in a python dictionary format.
    """
    queryset: "QuerySet" = self._clone()
    fields = fields or []
    if flat and len(fields) > 1:
        raise QuerySetError(
            detail=f"Maximum of 1 in fields when `flat` is enables, got {len(fields)} instead."
        ) from None

    if flat and isinstance(fields, str):
        fields = [fields]

    if isinstance(fields, str):
        fields = [fields]

    return await queryset.values(
        fields=fields,
        exclude=exclude,
        exclude_none=exclude_none,
        flatten=flat,
        __as_tuple__=True,
    )

exists async

exists(**kwargs)

Returns a boolean indicating if a record exists or not.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def exists(self, **kwargs: Any) -> bool:
    """
    Returns a boolean indicating if a record exists or not.
    """
    queryset: "QuerySet" = self._clone()
    expression = queryset._build_select()
    expression = sqlalchemy.exists(expression).select()
    queryset._set_query_expression(expression)
    _exists = await queryset.database.fetch_val(expression)
    return cast("bool", _exists)

count async

count(**kwargs)

Returns an indicating the total records.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def count(self, **kwargs: Any) -> int:
    """
    Returns an indicating the total records.
    """
    queryset: "QuerySet" = self._clone()
    expression = queryset._build_select().alias("subquery_for_count")
    expression = sqlalchemy.func.count().select().select_from(expression)
    queryset._set_query_expression(expression)
    _count = await queryset.database.fetch_val(expression)
    return cast("int", _count)

get_or_none async

get_or_none(**kwargs)

Fetch one object matching the parameters or returns None.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def get_or_none(self, **kwargs: Any) -> Union[EdgyModel, None]:
    """
    Fetch one object matching the parameters or returns None.
    """
    queryset: "QuerySet" = self.filter(**kwargs)
    expression = queryset._build_select().limit(2)
    queryset._set_query_expression(expression)
    rows = await queryset.database.fetch_all(expression)

    if not rows:
        return None
    if len(rows) > 1:
        raise MultipleObjectsReturned()
    return queryset.model_class.from_sqla_row(
        rows[0],
        select_related=queryset._select_related,
        exclude_secrets=queryset._exclude_secrets,
        using_schema=queryset.using_schema,
    )

embed_parent_in_result

embed_parent_in_result(result)

PARAMETER	DESCRIPTION
result	TYPE: Any

Source code in edgy/core/db/querysets/base.py

def embed_parent_in_result(self, result: Any) -> Any:
    if not self.embed_parent:
        return result
    new_result = getattr(result, self.embed_parent[0])
    if self.embed_parent[1]:
        setattr(new_result, self.embed_parent[1], result)
    return new_result

_all async

_all(**kwargs)

Executes the query.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def _all(self, **kwargs: Any) -> List[EdgyModel]:
    """
    Executes the query.
    """
    queryset: "QuerySet" = self._clone()
    if queryset.embed_parent:
        # activates distinct, not distinct on
        queryset.distinct_on = []

    if kwargs:
        return await queryset.filter(**kwargs).all()

    expression = queryset._build_select()
    queryset._set_query_expression(expression)

    rows = await queryset.database.fetch_all(expression)

    # Attach the raw query to the object
    queryset.model_class.raw_query = queryset.sql

    is_only_fields = True if queryset._only else False
    is_defer_fields = True if queryset._defer else False

    results = [
        queryset.model_class.from_sqla_row(
            row,
            select_related=queryset._select_related,
            prefetch_related=queryset._prefetch_related,
            is_only_fields=is_only_fields,
            only_fields=queryset._only,
            is_defer_fields=is_defer_fields,
            exclude_secrets=queryset._exclude_secrets,
            using_schema=queryset.using_schema,
        )
        for row in rows
    ]

    all_results = [self.embed_parent_in_result(result) for result in results]
    return all_results

all

all(**kwargs)

Returns the queryset records based on specific filters

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

def all(self, **kwargs: Any) -> "QuerySet":
    """
    Returns the queryset records based on specific filters
    """
    queryset: "QuerySet" = self._clone()
    queryset.extra = kwargs
    return queryset

get async

get(**kwargs)

Returns a single record based on the given kwargs.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def get(self, **kwargs: Any) -> EdgyModel:
    """
    Returns a single record based on the given kwargs.
    """
    queryset: "QuerySet" = self._clone()

    if kwargs:
        return await queryset.filter(**kwargs).get()

    expression = queryset._build_select().limit(2)
    rows = await queryset.database.fetch_all(expression)
    queryset._set_query_expression(expression)

    is_only_fields = True if queryset._only else False
    is_defer_fields = True if queryset._defer else False

    if not rows:
        raise ObjectNotFound()
    if len(rows) > 1:
        raise MultipleObjectsReturned()

    return self.embed_parent_in_result(queryset.model_class.from_sqla_row(
        rows[0],
        select_related=queryset._select_related,
        is_only_fields=is_only_fields,
        only_fields=queryset._only,
        is_defer_fields=is_defer_fields,
        prefetch_related=queryset._prefetch_related,
        exclude_secrets=queryset._exclude_secrets,
        using_schema=queryset.using_schema,
    ))

first async

first(**kwargs)

Returns the first record of a given queryset.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def first(self, **kwargs: Any) -> Union[EdgyModel, None]:
    """
    Returns the first record of a given queryset.
    """
    queryset: "QuerySet" = self._clone()
    if kwargs:
        return await queryset.filter(**kwargs).order_by("id").get()

    rows = await queryset.limit(1).order_by("id").all()
    if rows:
        return self.embed_parent_in_result(rows[0])
    return None

last async

last(**kwargs)

Returns the last record of a given queryset.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def last(self, **kwargs: Any) -> Union[EdgyModel, None]:
    """
    Returns the last record of a given queryset.
    """
    queryset: "QuerySet" = self._clone()
    if kwargs:
        return await queryset.filter(**kwargs).order_by("-id").get()

    rows = await queryset.order_by("-id").all()
    if rows:
        return self.embed_parent_in_result(rows[0])
    return None

create async

create(**kwargs)

Creates a record in a specific table.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def create(self, **kwargs: Any) -> EdgyModel:
    """
    Creates a record in a specific table.
    """
    queryset: "QuerySet" = self._clone()
    instance = queryset.model_class(**kwargs)
    instance.table = queryset.table
    instance = await instance.save(force_save=True)
    return self.embed_parent_in_result(instance)

bulk_create async

bulk_create(objs)

Bulk creates records in a table

PARAMETER	DESCRIPTION
objs	TYPE: List[Dict]

Source code in edgy/core/db/querysets/base.py

async def bulk_create(self, objs: List[Dict]) -> None:
    """
    Bulk creates records in a table
    """
    queryset: "QuerySet" = self._clone()
    new_objs = [queryset._validate_kwargs(**obj) for obj in objs]

    expression = queryset.table.insert().values(new_objs)
    queryset._set_query_expression(expression)
    await queryset.database.execute(expression)

bulk_update async

bulk_update(objs, fields)

Bulk updates records in a table.

A similar solution was suggested here: https://github.com/encode/orm/pull/148

It is thought to be a clean approach to a simple problem so it was added here and refactored to be compatible with Edgy.

PARAMETER	DESCRIPTION
objs	TYPE: List[EdgyModel]
fields	TYPE: List[str]

Source code in edgy/core/db/querysets/base.py

async def bulk_update(self, objs: List[EdgyModel], fields: List[str]) -> None:
    """
    Bulk updates records in a table.

    A similar solution was suggested here: https://github.com/encode/orm/pull/148

    It is thought to be a clean approach to a simple problem so it was added here and
    refactored to be compatible with Edgy.
    """
    queryset: "QuerySet" = self._clone()

    new_objs = []
    for obj in objs:
        new_obj = {}
        for key, value in obj.__dict__.items():
            if key in fields:
                new_obj[key] = self._resolve_value(value)
        new_objs.append(new_obj)

    new_objs = [queryset._extract_values_from_field(obj, queryset.model_class) for obj in new_objs]

    pks1 = (getattr(queryset.table.c, pkcol) == sqlalchemy.bindparam(pkcol) for pkcol in queryset.pkcolumns)
    expression = queryset.table.update().where(*pks1)
    kwargs: Dict[Any, Any] = {field: sqlalchemy.bindparam(field) for obj in new_objs for field in obj.keys()}
    pks2 = [{pkcol: getattr(obj, pkcol) for pkcol in queryset.pkcolumns} for obj in objs]

    query_list = []
    for pk, value in zip(pks2, new_objs):  # noqa
        query_list.append({**pk, **value})

    expression = expression.values(kwargs)
    queryset._set_query_expression(expression)
    await queryset.database.execute_many(str(expression), query_list)

delete async

delete()

Source code in edgy/core/db/querysets/base.py

async def delete(self) -> None:
    queryset: "QuerySet" = self._clone()

    await self.model_class.signals.pre_delete.send_async(self.__class__, instance=self)

    expression = queryset.table.delete()
    for filter_clause in queryset.filter_clauses:
        expression = expression.where(filter_clause)

    queryset._set_query_expression(expression)
    await queryset.database.execute(expression)

    await self.model_class.signals.post_delete.send_async(self.__class__, instance=self)

update async

update(**kwargs)

Updates a record in a specific table with the given kwargs.

PARAMETER	DESCRIPTION
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def update(self, **kwargs: Any) -> None:
    """
    Updates a record in a specific table with the given kwargs.
    """
    queryset: "QuerySet" = self._clone()

    extracted_fields = queryset._extract_values_from_field(kwargs, model_class=queryset.model_class)
    kwargs = queryset._update_auto_now_fields(extracted_fields, queryset.model_class.fields)

    # Broadcast the initial update details
    await self.model_class.signals.pre_update.send_async(self.__class__, instance=self, kwargs=kwargs)

    expression = queryset.table.update().values(**kwargs)

    for filter_clause in queryset.filter_clauses:
        expression = expression.where(filter_clause)

    queryset._set_query_expression(expression)
    await queryset.database.execute(expression)

    # Broadcast the update executed
    await self.model_class.signals.post_update.send_async(self.__class__, instance=self)

get_or_create async

get_or_create(defaults, **kwargs)

Creates a record in a specific table or updates if already exists.

PARAMETER	DESCRIPTION
defaults	TYPE: Dict[str, Any]
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def get_or_create(self, defaults: Dict[str, Any], **kwargs: Any) -> Tuple[EdgyModel, bool]:
    """
    Creates a record in a specific table or updates if already exists.
    """
    queryset: "QuerySet" = self._clone()

    try:
        instance = await queryset.get(**kwargs)
        return instance, False
    except ObjectNotFound:
        kwargs.update(defaults)
        instance = await queryset.create(**kwargs)
        return instance, True

update_or_create async

update_or_create(defaults, **kwargs)

Updates a record in a specific table or creates a new one.

PARAMETER	DESCRIPTION
defaults	TYPE: Dict[str, Any]
**kwargs	TYPE: Any DEFAULT: {}

Source code in edgy/core/db/querysets/base.py

async def update_or_create(self, defaults: Dict[str, Any], **kwargs: Any) -> Tuple[EdgyModel, bool]:
    """
    Updates a record in a specific table or creates a new one.
    """
    queryset: "QuerySet" = self._clone()
    try:
        instance = await queryset.get(**kwargs)
        await instance.update(**defaults)
        return instance, False
    except ObjectNotFound:
        kwargs.update(defaults)
        instance = await queryset.create(**kwargs)
        return instance, True

contains async

contains(instance)

Returns true if the QuerySet contains the provided object. False if otherwise.

PARAMETER	DESCRIPTION
instance	TYPE: EdgyModel

Source code in edgy/core/db/querysets/base.py

async def contains(self, instance: EdgyModel) -> bool:
    """Returns true if the QuerySet contains the provided object.
    False if otherwise.
    """
    queryset: "QuerySet" = self._clone()

    if getattr(instance, "pk", None) is None:
        raise ValueError("'obj' must be a model or reflect model instance.")
    # TODO: handle embed parent
    return await queryset.filter(pk=instance.pk).exists()

_execute async

_execute()

Source code in edgy/core/db/querysets/base.py

async def _execute(self) -> Any:
    queryset: "QuerySet" = self._clone()
    records = await queryset._all(**queryset.extra)
    return records