QuerySet class

edgy.QuerySet

QuerySet(model_class=None, *, database=None, filter_clauses=_empty_set, select_related=_empty_set, prefetch_related=_empty_set, limit_count=None, limit=None, limit_offset=None, offset=None, batch_size=None, order_by=_empty_set, group_by=_empty_set, distinct_on=None, distinct=None, only_fields=None, only=_empty_set, defer_fields=None, defer=_empty_set, embed_parent=None, using_schema=Undefined, table=None, exclude_secrets=False, extra_select=None, reference_select=None)

Bases: BaseQuerySet

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

def __init__(
    self,
    model_class: type[BaseModelType] | None = None,
    *,
    database: Database | None = None,
    filter_clauses: Iterable[Any] = _empty_set,
    select_related: Iterable[str] = _empty_set,
    prefetch_related: Iterable[Prefetch] = _empty_set,
    limit_count: int | None = None,
    limit: int | None = None,
    limit_offset: int | None = None,
    offset: int | None = None,
    batch_size: int | None = None,
    order_by: Iterable[str] = _empty_set,
    group_by: Iterable[str] = _empty_set,
    distinct_on: None | Literal[True] | Iterable[str] = None,
    distinct: None | Literal[True] | Iterable[str] = None,
    only_fields: Iterable[str] | None = None,
    only: Iterable[str] = _empty_set,
    defer_fields: Sequence[str] | None = None,
    defer: Iterable[str] = _empty_set,
    embed_parent: tuple[str, str | str] | None = None,
    using_schema: str | None | Any = Undefined,
    table: sqlalchemy.Table | None = None,
    exclude_secrets: bool = False,
    extra_select: Iterable[sqlalchemy.ClauseElement] | None = None,
    reference_select: reference_select_type | None = None,
) -> None:
    if model_class.__is_proxy_model__:
        model_class = model_class.__parent__

    super().__init__(model_class=model_class)
    self.filter_clauses: list[Any] = list(filter_clauses)
    self.or_clauses: list[Any] = []
    self._aliases: dict[str, sqlalchemy.Alias] = {}
    if limit_count is not None:
        warnings.warn(
            "`limit_count` is deprecated use `limit`", DeprecationWarning, stacklevel=2
        )
        limit = limit_count
    self.limit_count = limit
    if limit_offset is not None:
        warnings.warn(
            "`limit_offset` is deprecated use `limit`", DeprecationWarning, stacklevel=2
        )
        offset = limit_offset
    self._offset = offset
    select_related = set(select_related)
    self._select_related: set[str] = set()
    self._select_related_weak: set[str] = set()
    if select_related:
        self._update_select_related(select_related)
    self._prefetch_related = list(prefetch_related)
    self._batch_size = batch_size
    self._order_by: tuple[str, ...] = tuple(order_by)
    self._group_by: tuple[str, ...] = tuple(group_by)
    if distinct_on is not None:
        warnings.warn(
            "`distinct_on` is deprecated use `distinct`", DeprecationWarning, stacklevel=2
        )
        distinct = distinct_on

    if distinct is True:
        distinct = _empty_set
    self.distinct_on = list(distinct) if distinct is not None else None
    if only_fields is not None:
        warnings.warn(
            "`only_fields` is deprecated use `only`", DeprecationWarning, stacklevel=2
        )
        only = only_fields
    self._only = set(only)
    if defer_fields is not None:
        warnings.warn(
            "`defer_fields` is deprecated use `defer`", DeprecationWarning, stacklevel=2
        )
        defer = defer_fields
    self._defer = set(defer)
    self.embed_parent = embed_parent
    self.embed_parent_filters: tuple[str, str | str] | None = None
    self.using_schema = using_schema
    self._extra_select = list(extra_select) if extra_select is not None else []
    self._reference_select = (
        reference_select.copy() if isinstance(reference_select, dict) else {}
    )
    self._exclude_secrets = exclude_secrets
    self._cache = QueryModelResultCache(attrs=self.model_class.pkcolumns)
    self._clear_cache(keep_result_cache=False)
    self._cached_select_related_expression: (
        tuple[Any, dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]]] | None
    ) = None
    self.active_schema = self.get_schema()
    self._for_update: dict[str, Any] | None = None

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

model_class instance-attribute

model_class = model_class

The model class associated with this QuerySet.

_table class-attribute instance-attribute

_table = None

_database class-attribute instance-attribute

_database = None

database instance-attribute

database = database

table instance-attribute

table = table

pkcolumns property

pkcolumns

Returns a sequence of primary key column names for the model class associated with this queryset.

This property directly delegates to the pkcolumns property of the model_class, providing access to the underlying database column names that constitute the primary key.

RETURNS	DESCRIPTION
Sequence[str]	Sequence[str]: A sequence (e.g., list or tuple) of strings, where each string is the name of a primary key column in the database.

filter_clauses instance-attribute

filter_clauses = list(filter_clauses)

or_clauses instance-attribute

or_clauses = []

_aliases instance-attribute

_aliases = {}

limit_count instance-attribute

limit_count = limit

_offset instance-attribute

_offset = offset

_select_related instance-attribute

_select_related = set()

_select_related_weak instance-attribute

_select_related_weak = set()

_prefetch_related instance-attribute

_prefetch_related = list(prefetch_related)

_batch_size instance-attribute

_batch_size = batch_size

_order_by instance-attribute

_order_by = tuple(order_by)

_group_by instance-attribute

_group_by = tuple(group_by)

distinct_on instance-attribute

distinct_on = list(distinct) if distinct is not None else None

_only instance-attribute

_only = set(only)

_defer instance-attribute

_defer = set(defer)

embed_parent instance-attribute

embed_parent = embed_parent

embed_parent_filters instance-attribute

embed_parent_filters = None

using_schema instance-attribute

using_schema = using_schema

_extra_select instance-attribute

_extra_select = list(extra_select) if extra_select is not None else []

_reference_select instance-attribute

_reference_select = copy() if isinstance(reference_select, dict) else {}

_exclude_secrets instance-attribute

_exclude_secrets = exclude_secrets

_cache instance-attribute

_cache = QueryModelResultCache(attrs=pkcolumns)

_cached_select_related_expression instance-attribute

_cached_select_related_expression = None

active_schema instance-attribute

active_schema = get_schema()

_for_update instance-attribute

_for_update = None

_has_dynamic_clauses cached property

_has_dynamic_clauses

_current_row property

_current_row

(Refactored: Delegates to the helper in executor.py)

sql cached property

sql

Get SQL select query as string with inserted blanks. For debugging only!

build_where_clause async

build_where_clause(_=None, tables_and_models=None)

(This method is now a simple forwarder to the Compiler. It's kept for API compatibility, e.g. for QuerySet(QuerySet) filters)

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

async def build_where_clause(
    self, _: Any = None, tables_and_models: tables_and_models_type | None = None
) -> Any:
    """
    (This method is now a simple forwarder to the Compiler.
     It's kept for API compatibility, e.g. for QuerySet(QuerySet) filters)
    """
    compiler = QueryCompiler(self)
    joins: Any | None = None
    if tables_and_models is None:
        joins, tables_and_models = self._get_join_graph_data()

    return await compiler.build_where_clause(tables_and_models, joins=joins)

using

using(_positional=_sentinel, *, database=Undefined, schema=Undefined)

Enables and switches the database schema and/or database connection for the queryset.

This method creates a new QuerySet instance that operates within the specified database and/or schema context. It's designed to support multi-tenancy by allowing dynamic selection of the database or schema without modifying the original QuerySet.

PARAMETER	DESCRIPTION
_positional	Deprecated positional argument for schema. This argument is maintained for backward compatibility but its use is discouraged. It will be treated as the schema argument. TYPE: Any DEFAULT: _sentinel
database	Specifies the database to use. - str: Name of an extra database connection registered in the model's registry. - Database: A Database instance to use directly. - None: Uses the default database from the model's registry. - Undefined (default): Retains the current database of the queryset. TYPE: str | Database | None DEFAULT: Undefined
schema	Specifies the database schema to use. - str: The schema name to activate. - False: Unsets the schema, reverting to the active default schema for the model. - None: Uses no specific schema. - Undefined (default): Retains the current schema of the queryset. TYPE: str | None | Literal[False] DEFAULT: Undefined

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance configured with the specified database and/or schema settings. This new instance allows chaining operations within the new context. TYPE: QuerySet

WARNS	DESCRIPTION
DeprecationWarning	If positional arguments are passed to this method for _positional. Users should explicitly use schema= keyword argument instead.

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

def using(
    self,
    _positional: Any = _sentinel,
    *,
    database: str | Any | None | Database = Undefined,
    schema: str | Any | None | Literal[False] = Undefined,
) -> QuerySet:
    """
    Enables and switches the database schema and/or database connection for
    the queryset.

    This method creates a new QuerySet instance that operates within the
    specified database and/or schema context. It's designed to support
    multi-tenancy by allowing dynamic selection of the database or schema
    without modifying the original QuerySet.

    Args:
        _positional (Any): Deprecated positional argument for `schema`. This
                           argument is maintained for backward compatibility but
                           its use is discouraged. It will be treated as the
                           `schema` argument.
        database (str | Database | None): Specifies the database to use.
          - `str`: Name of an extra database connection registered in the model's registry.
          - `Database`: A Database instance to use directly.
          - `None`: Uses the default database from the model's registry.
          - `Undefined` (default): Retains the current database of the queryset.
        schema (str | None | Literal[False]): Specifies the database schema to use.
          - `str`: The schema name to activate.
          - `False`: Unsets the schema, reverting to the active default schema for the model.
          - `None`: Uses no specific schema.
          - `Undefined` (default): Retains the current schema of the queryset.

    Returns:
        QuerySet: A new QuerySet instance configured with the specified database
                  and/or schema settings. This new instance allows chaining
                  operations within the new context.

    Warnings:
        DeprecationWarning: If positional arguments are passed to this method for
                            `_positional`. Users should explicitly use `schema=`
                            keyword argument instead.
    """
    # Check for deprecated positional argument usage.
    if _positional is not _sentinel:
        warnings.warn(
            "Passing positional arguments to using is deprecated. Use schema= instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        schema = _positional
        # If schema was not explicitly set as Undefined (meaning it was passed
        # positionally) and is still Undefined, set it to False to unset the schema.
        if schema is Undefined:
            schema = False

    # Create a clone of the current queryset to avoid modifying the original.
    queryset = cast("QuerySet", self._clone())

    # Process the 'database' argument.
    if database is not Undefined:
        if isinstance(database, Database):
            connection: Database = database
        elif database is None:
            # Use the default database from the model's registry.
            connection = self.model_class.meta.registry.database
        else:
            # Assert that the database name exists in the extra connections.
            assert database is None or database in self.model_class.meta.registry.extra, (
                f"`{database}` is not in the connections extra of the model"
                f"`{self.model_class.__name__}` registry"
            )
            # Retrieve the database from extra connections.
            connection = self.model_class.meta.registry.extra[database]
        # Assign the determined connection to the new queryset's database.
        queryset.database = connection

    # Process the 'schema' argument.
    if schema is not Undefined:
        # If schema is False, set using_schema to Undefined to unset it;
        # otherwise, use the provided schema.
        queryset.using_schema = schema if schema is not False else Undefined
        # Get the new schema based on the current queryset's configuration.
        new_schema = queryset.get_schema()
        # If the new schema is different from the active schema, update and
        # invalidate table cache.
        if new_schema != queryset.active_schema:
            queryset.active_schema = new_schema
            queryset.table = None  # Force regeneration of the table with the new schema.

    return queryset

prefetch_related

prefetch_related(*prefetch)

Performs a reverse lookup for foreign keys and other relationships, populating results onto the main model instances.

This method is distinct from select_related in that select_related performs a SQL JOIN to fetch related data in the same query, whereas prefetch_related executes separate queries for each relationship and then joins the results in Python. This is particularly useful for many-to-many relationships or reverse foreign key lookups, or when preloading related objects for a large set of parent objects.

PARAMETER	DESCRIPTION
*prefetch	One or more Prefetch objects, each defining a relationship to prefetch, including the related_name and the to_attr where results will be stored. An optional custom QuerySet can also be provided within the Prefetch object. TYPE: Prefetch DEFAULT: ()

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance with the specified prefetch relationships configured. This new QuerySet can then be further filtered, ordered, or executed. TYPE: QuerySet

RAISES	DESCRIPTION
QuerySetError	If any argument passed to prefetch is not an instance of the Prefetch class.

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

def prefetch_related(self, *prefetch: Prefetch) -> QuerySet:
    """
    Performs a reverse lookup for foreign keys and other relationships,
    populating results onto the main model instances.

    This method is distinct from `select_related` in that `select_related`
    performs a SQL JOIN to fetch related data in the same query, whereas
    `prefetch_related` executes separate queries for each relationship
    and then joins the results in Python. This is particularly useful for
    many-to-many relationships or reverse foreign key lookups, or when
    preloading related objects for a large set of parent objects.

    Args:
        *prefetch (Prefetch): One or more `Prefetch` objects, each defining
                               a relationship to prefetch, including the
                               `related_name` and the `to_attr` where results
                               will be stored. An optional custom `QuerySet`
                               can also be provided within the `Prefetch` object.

    Returns:
        QuerySet: A new `QuerySet` instance with the specified prefetch
                  relationships configured. This new QuerySet can then be
                  further filtered, ordered, or executed.

    Raises:
        QuerySetError: If any argument passed to `prefetch` is not an
                       instance of the `Prefetch` class.
    """
    queryset: QuerySet = self._clone()

    # Validate that all provided arguments are instances of Prefetch.
    if any(not isinstance(value, Prefetch) for value in prefetch):
        raise QuerySetError("The prefetch_related must have Prefetch type objects only.")

    # Append the new prefetch objects to the queryset's internal list.
    queryset._prefetch_related = [*self._prefetch_related, *prefetch]
    return queryset

using_with_db

using_with_db(connection_name, schema=Undefined)

Switches the database connection and optionally the schema for the queryset.

This method is deprecated in favor of the more flexible using method, which accepts both database and schema as keyword arguments.

PARAMETER	DESCRIPTION
connection_name	The name of the database connection (registered in the model's registry's extra connections) to switch to. TYPE: str
schema	The schema name to use. - str: The schema name to activate. - False: Unsets the schema, reverting to the active default schema. - None: Uses no specific schema. - Undefined (default): Retains the current schema. TYPE: str | None | Literal[False] DEFAULT: Undefined

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance configured with the specified database connection and schema. TYPE: QuerySet

WARNS	DESCRIPTION
DeprecationWarning	This method is deprecated. Users should migrate to using(database=..., schema=...) for future compatibility.

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

def using_with_db(
    self, connection_name: str, schema: str | Any | None | Literal[False] = Undefined
) -> QuerySet:
    """
    Switches the database connection and optionally the schema for the queryset.

    This method is deprecated in favor of the more flexible `using` method, which
    accepts both `database` and `schema` as keyword arguments.

    Args:
        connection_name (str): The name of the database connection (registered
                               in the model's registry's `extra` connections)
                               to switch to.
        schema (str | None | Literal[False]): The schema name to use.
          - `str`: The schema name to activate.
          - `False`: Unsets the schema, reverting to the active default schema.
          - `None`: Uses no specific schema.
          - `Undefined` (default): Retains the current schema.

    Returns:
        QuerySet: A new QuerySet instance configured with the specified database
                  connection and schema.

    Warnings:
        DeprecationWarning: This method is deprecated. Users should migrate to
                            `using(database=..., schema=...)` for future
                            compatibility.
    """
    warnings.warn(
        "'using_with_db' is deprecated in favor of 'using' with schema, database arguments.",
        DeprecationWarning,
        stacklevel=2,
    )
    # Delegate to the `using` method with the appropriate keyword arguments.
    return self.using(database=connection_name, schema=schema)

_clone

_clone()

This is core to the builder pattern

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

def _clone(self) -> QuerySet:
    """
    This is core to the builder pattern
    """
    queryset = self.__class__(
        self.model_class,
        database=getattr(self, "_database", None),
        filter_clauses=self.filter_clauses,
        prefetch_related=self._prefetch_related,
        limit=self.limit_count,
        offset=self._offset,
        batch_size=self._batch_size,
        order_by=self._order_by,
        group_by=self._group_by,
        distinct=self.distinct_on,
        only=self._only,
        defer=self._defer,
        embed_parent=self.embed_parent,
        using_schema=self.using_schema,
        table=getattr(self, "_table", None),
        exclude_secrets=self._exclude_secrets,
        reference_select=self._reference_select,
        extra_select=self._extra_select,
    )
    queryset.or_clauses.extend(self.or_clauses)
    queryset.embed_parent_filters = self.embed_parent_filters
    queryset._select_related.update(self._select_related)
    queryset._select_related_weak.update(self._select_related_weak)
    queryset._cached_select_related_expression = self._cached_select_related_expression
    queryset._for_update = self._for_update.copy() if self._for_update is not None else None
    return cast("QuerySet", queryset)

_as_select_with_tables async

_as_select_with_tables()

(This is the new internal method for the base class) Builds the query select by delegating to the QueryCompiler.

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

async def _as_select_with_tables(
    self,
) -> tuple[Any, tables_and_models_type]:
    """
    (This is the new internal method for the base class)
    Builds the query select by delegating to the QueryCompiler.
    """
    compiler = QueryCompiler(self)
    self._get_join_graph_data()
    expression, tables_and_models = await compiler.build_select()
    return expression, tables_and_models

_clear_cache

_clear_cache(*, keep_result_cache=False, keep_cached_selected=False)

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

def _clear_cache(
    self, *, keep_result_cache: bool = False, keep_cached_selected: bool = False
) -> None:
    if not keep_result_cache:
        self._cache.clear()
    if not keep_cached_selected:
        self._cached_select_with_tables: (
            tuple[Any, dict[str, tuple[sqlalchemy.Table, type[BaseModelType]]]] | None
        ) = None
    self._cache_count: int | None = None
    self._cache_first: tuple[BaseModelType, Any] | None = None
    self._cache_last: tuple[BaseModelType, Any] | None = None
    self._cache_fetch_all: bool = False

_build_order_by_iterable

_build_order_by_iterable(order_by, tables_and_models)

This is a helper for the compiler but is called by it, so it's okay for it to live here as it's part of the 'builder' logic.

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

def _build_order_by_iterable(
    self, order_by: Iterable[str], tables_and_models: tables_and_models_type
) -> Iterable:
    """
    This is a helper for the *compiler* but is called by it,
     so it's okay for it to live here as it's part of the 'builder' logic.
    """
    return (self._prepare_order_by(entry, tables_and_models) for entry in order_by)

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

_get_join_graph_data

_get_join_graph_data()

Gets the join graph, building it via the compiler if needed.

This is the new "bridge" that manages the _cached_select_related_expression variable to satisfy brittle tests, while keeping the compiler itself stateless.

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

def _get_join_graph_data(self) -> tuple[Any, tables_and_models_type]:
    """
    Gets the join graph, building it via the compiler if needed.

    This is the new "bridge" that manages the
    _cached_select_related_expression variable to satisfy brittle tests,
    while keeping the compiler itself stateless.
    """
    if self._cached_select_related_expression is None:
        # Create a compiler just to build the join graph
        compiler = QueryCompiler(self)

        # Call the compiler's build method and cache the result
        self._cached_select_related_expression = compiler.build_join_graph()
    return self._cached_select_related_expression

as_select_with_tables async

as_select_with_tables()

(Refactored: Now delegates to the Compiler)

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

async def as_select_with_tables(
    self,
) -> tuple[Any, tables_and_models_type]:
    """
    (Refactored: Now delegates to the Compiler)
    """
    if self._cached_select_with_tables is None:
        self._cached_select_with_tables = await self._as_select_with_tables()
    return self._cached_select_with_tables

as_select async

as_select()

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

async def as_select(
    self,
) -> Any:
    return (await self.as_select_with_tables())[0]

_kwargs_to_clauses

_kwargs_to_clauses(kwargs)

This is part of the 'filter' builder logic

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

def _kwargs_to_clauses(
    self,
    kwargs: Any,
) -> tuple[list[Any], set[str]]:
    """
    This is part of the 'filter' builder logic
    """
    clauses = []
    select_related: set[str] = set()
    cleaned_kwargs = clauses_mod.clean_query_kwargs(
        self.model_class, kwargs, self.embed_parent_filters, model_database=self.database
    )

    for key, value in cleaned_kwargs.items():
        model_class, field_name, op, related_str, _, cross_db_remainder = crawl_relationship(
            self.model_class, key
        )
        if related_str:
            select_related.add(related_str)
        field = model_class.meta.fields.get(field_name, clauses_mod.generic_field)
        if cross_db_remainder:
            assert field is not clauses_mod.generic_field
            fk_field = cast(BaseForeignKey, field)
            sub_query = (
                fk_field.target.query.filter(**{cross_db_remainder: value})
                .only(*fk_field.related_columns.keys())
                .values_list(fields=fk_field.related_columns.keys())
            )

            async def wrapper(
                queryset: QuerySet,
                tables_and_models: tables_and_models_type,
                *,
                _field: BaseFieldType = field,
                _sub_query: QuerySet = sub_query,
                _prefix: str = related_str,
            ) -> Any:
                table = tables_and_models[_prefix][0]
                fk_tuple = sqlalchemy.tuple_(
                    *(getattr(table.columns, colname) for colname in _field.get_column_names())
                )
                return fk_tuple.in_(await _sub_query)

            clauses.append(wrapper)
        else:
            assert not isinstance(value, BaseModelType), (
                f"should be parsed in clean: {key}: {value}"
            )

            async def wrapper(
                queryset: QuerySet,
                tables_and_models: tables_and_models_type,
                *,
                _field: BaseFieldType = field,
                _value: Any = value,
                _op: str | None = op,
                _prefix: str = related_str,
                _field_name: str = field_name,
            ) -> Any:
                _value = await clauses_mod.parse_clause_arg(
                    _value, queryset, tables_and_models
                )
                table = tables_and_models[_prefix][0]
                return _field.operator_to_clause(_field_name, _op, table, _value)

            wrapper._edgy_force_callable_queryset_filter = True
            clauses.append(wrapper)

    return clauses, select_related

_prepare_order_by

_prepare_order_by(order_by, tables_and_models)

(Helper for 'order_by' builder logic, but called by compiler)

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

def _prepare_order_by(self, order_by: str, tables_and_models: tables_and_models_type) -> Any:
    """
    (Helper for 'order_by' builder logic, but called by compiler)
    """
    reverse = order_by.startswith("-")
    order_by = order_by.lstrip("-")
    crawl_result = clauses_mod.clean_path_to_crawl_result(
        self.model_class,
        path=order_by,
        embed_parent=self.embed_parent_filters,
        model_database=self.database,
    )
    order_col = tables_and_models[crawl_result.forward_path][0].columns[
        crawl_result.field_name
    ]
    return order_col.desc() if reverse else order_col

_update_select_related_weak

_update_select_related_weak(fields, *, clear)

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

def _update_select_related_weak(self, fields: Iterable[str], *, clear: bool) -> bool:
    related: set[str] = set()
    for field_name in fields:
        field_name = field_name.lstrip("-")
        related_element = clauses_mod.clean_path_to_crawl_result(
            self.model_class,
            path=field_name,
            embed_parent=self.embed_parent_filters,
            model_database=self.database,
        ).forward_path
        if related_element:
            related.add(related_element)
    if related and not self._select_related.union(self._select_related_weak).issuperset(
        related
    ):
        self._cached_select_related_expression = None
        if clear:
            self._select_related_weak.clear()
        self._select_related_weak.update(related)
        return True
    return False

_update_select_related

_update_select_related(pathes)

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

def _update_select_related(self, pathes: Iterable[str]) -> None:
    related: set[str] = set()
    for path in pathes:
        path = path.lstrip("-")
        crawl_result = clauses_mod.clean_path_to_crawl_result(
            self.model_class,
            path=path,
            embed_parent=self.embed_parent_filters,
            model_database=self.database,
        )
        related_element = (
            crawl_result.field_name
            if not crawl_result.forward_path
            else f"{crawl_result.forward_path}__{crawl_result.field_name}"
        )
        if crawl_result.cross_db_remainder:
            raise QuerySetError(
                detail=f'Selected path "{related_element}" is on another database.'
            )
        if related_element:
            related.add(related_element)
    if related and not self._select_related.issuperset(related):
        self._cached_select_related_expression = None
        self._select_related.update(related)

_prepare_distinct

_prepare_distinct(distinct_on, tables_and_models)

Helper for 'distinct' builder, but called by compiler

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

def _prepare_distinct(
    self, distinct_on: str, tables_and_models: tables_and_models_type
) -> sqlalchemy.Column:
    """Helper for 'distinct' builder, but called by compiler"""
    crawl_result = clauses_mod.clean_path_to_crawl_result(
        self.model_class,
        path=distinct_on,
        embed_parent=self.embed_parent_filters,
        model_database=self.database,
    )
    return tables_and_models[crawl_result.forward_path][0].columns[crawl_result.field_name]

_embed_parent_in_result async

_embed_parent_in_result(result)

This is a result transformation, called by the Parser.

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

async def _embed_parent_in_result(
    self, result: EdgyModel | Awaitable[EdgyModel]
) -> tuple[EdgyModel, Any]:
    """
    This is a result transformation, called by the Parser.
    """
    if isawaitable(result):
        result = await result
    if not self.embed_parent:
        return result, result
    token = MODEL_GETATTR_BEHAVIOR.set("coro")
    try:
        new_result: Any = result
        for part in self.embed_parent[0].split("__"):
            new_result = getattr(new_result, part)
            if isawaitable(new_result):
                new_result = await new_result
    finally:
        MODEL_GETATTR_BEHAVIOR.reset(token)
    if self.embed_parent[1]:
        setattr(new_result, self.embed_parent[1], result)
    return result, new_result

get_schema

get_schema()

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

def get_schema(self) -> str | None:
    schema = self.using_schema
    if schema is Undefined:
        schema = get_schema()
    if schema is None:
        schema = self.model_class.get_db_schema()
    return schema

_execute_iterate async

_execute_iterate(fetch_all_at_once=False)

(Refactored: Now delegates to the Executor)

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

async def _execute_iterate(
    self, fetch_all_at_once: bool = False
) -> AsyncIterator[BaseModelType]:
    """
    (Refactored: Now delegates to the Executor)
    """
    # Create the specialists
    compiler = QueryCompiler(self)
    parser = ResultParser(self)
    executor = QueryExecutor(self, compiler, parser)

    # Delegate the work
    async for model in executor.iterate(fetch_all_at_once=fetch_all_at_once):  # type: ignore
        yield model

_execute_all async

_execute_all()

Still relies on _execute_iterate.

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

async def _execute_all(self) -> list[EdgyModel]:
    """
    Still relies on _execute_iterate.
    """
    return [result async for result in self._execute_iterate(fetch_all_at_once=True)]

_filter_or_exclude

_filter_or_exclude(kwargs, clauses, exclude=False, or_=False, allow_global_or=True)

This is the core 'filter' builder logic.

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

def _filter_or_exclude(
    self,
    kwargs: Any,
    clauses: Sequence[
        sqlalchemy.sql.expression.BinaryExpression
        | Callable[
            [QuerySetType],
            sqlalchemy.sql.expression.BinaryExpression
            | Awaitable[sqlalchemy.sql.expression.BinaryExpression],
        ]
        | dict[str, Any]
        | QuerySet
    ],
    exclude: bool = False,
    or_: bool = False,
    allow_global_or: bool = True,
) -> QuerySet:
    """
    This is the core 'filter' builder logic.
    """
    from edgy.core.db.querysets.queryset import QuerySet

    queryset: QuerySet = self._clone()
    if kwargs:
        clauses = [*clauses, kwargs]
    converted_clauses: Sequence[
        sqlalchemy.sql.expression.BinaryExpression
        | Callable[
            [QuerySetType],
            sqlalchemy.sql.expression.BinaryExpression
            | Awaitable[sqlalchemy.sql.expression.BinaryExpression],
        ]
    ] = []
    for raw_clause in clauses:
        if isinstance(raw_clause, dict):
            extracted_clauses, related = queryset._kwargs_to_clauses(kwargs=raw_clause)
            if not queryset._select_related.issuperset(related):
                queryset._select_related.update(related)
                queryset._cached_select_related_expression = None
            if or_ and extracted_clauses:
                wrapper_and = clauses_mod.and_(*extracted_clauses, no_select_related=True)

                if allow_global_or and len(clauses) == 1:
                    # Global OR mode: promote existing AND filters into the OR group.
                    # This turns:
                    #   qs.filter(A).or_(B)
                    # into:
                    #   OR( AND(A), AND(B) )
                    # instead of: OR(B) AND A.
                    assert not exclude

                    if queryset.filter_clauses:
                        # Wrap existing filters into a single AND group and move them to or_clauses
                        existing_and = clauses_mod.and_(
                            *queryset.filter_clauses, no_select_related=True
                        )
                        queryset.or_clauses.append(existing_and)
                        # Clear filter_clauses so they are not ANDed again later
                        queryset.filter_clauses = []

                    # Add the new OR operand
                    queryset.or_clauses.append(wrapper_and)
                    return queryset

                # Non-global OR (e.g. local_or) or multiple clauses:
                # just collect and handle them at the end as a local OR group.
                converted_clauses.append(wrapper_and)
            else:
                converted_clauses.extend(extracted_clauses)
        elif isinstance(raw_clause, QuerySet):
            assert raw_clause.model_class is queryset.model_class, (
                f"QuerySet arg has wrong model_class {raw_clause.model_class}"
            )
            converted_clauses.append(raw_clause.build_where_clause)
            if not queryset._select_related.issuperset(raw_clause._select_related):
                queryset._select_related.update(raw_clause._select_related)
                queryset._cached_select_related_expression = None
        else:
            clause = raw_clause

            # Support global OR mode for non-dict clauses (e.g. Q objects, raw callables)
            if or_ and allow_global_or and len(clauses) == 1:
                # Global OR only makes sense for non-exclude queries
                assert not exclude

                # If there are existing AND filters, promote them into the OR group
                if queryset.filter_clauses:
                    existing_and = clauses_mod.and_(
                        *queryset.filter_clauses,
                        no_select_related=True,
                    )
                    queryset.or_clauses.append(existing_and)
                    queryset.filter_clauses = []

                # Propagate select_related coming from this clause, if any
                if hasattr(clause, "_edgy_calculate_select_related"):
                    select_related_calculated = clause._edgy_calculate_select_related(queryset)
                    if not queryset._select_related.issuperset(select_related_calculated):
                        queryset._select_related.update(select_related_calculated)
                        queryset._cached_select_related_expression = None

                # Add this clause as a new OR branch and return immediately
                queryset.or_clauses.append(clause)
                return queryset

            # Normal path (no global OR promotion)
            converted_clauses.append(clause)
            if hasattr(clause, "_edgy_calculate_select_related"):
                select_related_calculated = clause._edgy_calculate_select_related(queryset)
                if not queryset._select_related.issuperset(select_related_calculated):
                    queryset._select_related.update(select_related_calculated)
                    queryset._cached_select_related_expression = None
    if not converted_clauses:
        return queryset

    if exclude:
        op = clauses_mod.and_ if not or_ else clauses_mod.or_

        queryset.filter_clauses.append(
            clauses_mod.not_(
                op(*converted_clauses, no_select_related=True), no_select_related=True
            )
        )
    elif or_:
        queryset.filter_clauses.append(
            clauses_mod.or_(*converted_clauses, no_select_related=True)
        )
    else:
        queryset.filter_clauses.extend(converted_clauses)
    return queryset

raw_delete async

raw_delete(use_models=False, remove_referenced_call=False)

(Refactored: Now delegates to the Executor)

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

async def raw_delete(
    self, use_models: bool = False, remove_referenced_call: str | bool = False
) -> int:
    """
    (Refactored: Now delegates to the Executor)
    """
    # We must create new specialists *every time* because the queryset
    # state might have changed (e.g., in _model_based_delete)
    compiler = QueryCompiler(self)
    parser = ResultParser(self)  # Delete doesn't use parser, but good practice
    executor = QueryExecutor(self, compiler, parser)

    return await executor.delete(
        use_models=use_models, remove_referenced_call=remove_referenced_call
    )

_get_raw async

_get_raw(**kwargs)

(Refactored: Builder logic stays, execution logic delegates)

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

async def _get_raw(self, **kwargs: Any) -> tuple[BaseModelType, Any]:
    """
    (Refactored: Builder logic stays, execution logic delegates)
    """
    if kwargs:
        cached = cast(
            tuple[BaseModelType, Any] | None, self._cache.get(self.model_class, kwargs)
        )
        if cached is not None:
            return cached
        filter_query = cast("BaseQuerySet", self.filter(**kwargs))
        filter_query._cache = self._cache
        return await filter_query._get_raw()
    elif self._cache_count == 1:
        if self._cache_first is not None:
            return self._cache_first
        elif self._cache_last is not None:
            return self._cache_last

    compiler = QueryCompiler(self)
    parser = ResultParser(self)
    executor = QueryExecutor(self, compiler, parser)
    return await executor.get_one()

_combine

_combine(other, op, *, all_=False)

Internal helper method used by the public interface (union, intersect, except_) to build and validate a CombinedQuerySet object.

This function enforces model class consistency and maps the requested set operation (with or without ALL) to the canonical operation name used by CombinedQuerySet.

PARAMETER	DESCRIPTION
other	The secondary QuerySet to combine results with. TYPE: QuerySet
op	The base set operation type ('union', 'intersect', or 'except'). TYPE: Literal['union', 'intersect', 'except']
all_	If True, requests the ALL variant of the set operation (e.g., UNION ALL). TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
CombinedQuerySet	A CombinedQuerySet instance configured for the specified set operation.

RAISES	DESCRIPTION
TypeError	If other is not an instance of QuerySet.
QuerySetError	If the models (self.model_class and other.model_class) of the two querysets do not match.

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

def _combine(
    self,
    other: QuerySet,
    op: Literal["union", "intersect", "except"],
    *,
    all_: bool = False,
) -> CombinedQuerySet:
    """
    Internal helper method used by the public interface (union, intersect, except_)
    to build and validate a CombinedQuerySet object.

    This function enforces model class consistency and maps the requested set operation
    (with or without ALL) to the canonical operation name used by `CombinedQuerySet`.

    Args:
        other: The secondary `QuerySet` to combine results with.
        op: The base set operation type ('union', 'intersect', or 'except').
        all_: If `True`, requests the ALL variant of the set operation (e.g., UNION ALL).

    Returns:
        A `CombinedQuerySet` instance configured for the specified set operation.

    Raises:
        TypeError: If `other` is not an instance of `QuerySet`.
        QuerySetError: If the models (`self.model_class` and `other.model_class`)
                       of the two querysets do not match.
    """
    from edgy.core.db.querysets.mixins.combined import CombinedQuerySet

    # Check type of the other object
    if not isinstance(other, QuerySet):
        raise TypeError("other must be a QuerySet")

    # Check model class consistency
    if self.model_class is not other.model_class:
        # QuerySetError detail string must be defined
        raise QuerySetError(detail="Both querysets must have the same model_class to combine.")

    # Map (op, all_) to a concrete set-op name understood by CombinedQuerySet
    if all_:
        if op == "union":
            op_name: str = "union_all"
        elif op == "intersect":
            op_name = "intersect_all"
        elif op == "except":
            op_name = "except_all"
        else:
            op_name = op  # type: ignore
    else:
        op_name = op

    return CombinedQuerySet(left=self, right=other, op=op_name)

union

union(other, *, all=False)

Returns a result set that is the UNION of the current QuerySet and another QuerySet.

By default, UNION returns only distinct rows (UNION DISTINCT). If all=True, it returns UNION ALL, which includes duplicate rows.

PARAMETER	DESCRIPTION
other	The other QuerySet to combine results with. TYPE: QuerySet
all	If True, performs UNION ALL; otherwise, performs UNION DISTINCT. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
CombinedQuerySet	A CombinedQuerySet instance representing the combined operation.

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

def union(self, other: QuerySet, *, all: bool = False) -> CombinedQuerySet:
    """
    Returns a result set that is the UNION of the current QuerySet and another QuerySet.

    By default, `UNION` returns only distinct rows (UNION DISTINCT).
    If `all=True`, it returns `UNION ALL`, which includes duplicate rows.

    Args:
        other: The other `QuerySet` to combine results with.
        all: If `True`, performs `UNION ALL`; otherwise, performs `UNION DISTINCT`.

    Returns:
        A `CombinedQuerySet` instance representing the combined operation.
    """
    return self._combine(other, "union", all_=all)

union_all

union_all(other)

Returns a result set that is the UNION ALL of the current QuerySet and another QuerySet.

This is a shortcut method equivalent to calling union(other, all=True).

PARAMETER	DESCRIPTION
other	The other QuerySet to combine results with. TYPE: QuerySet

RETURNS	DESCRIPTION
CombinedQuerySet	A CombinedQuerySet instance representing the UNION ALL operation.

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

def union_all(self, other: QuerySet) -> CombinedQuerySet:
    """
    Returns a result set that is the UNION ALL of the current QuerySet and another QuerySet.

    This is a shortcut method equivalent to calling `union(other, all=True)`.

    Args:
        other: The other `QuerySet` to combine results with.

    Returns:
        A `CombinedQuerySet` instance representing the UNION ALL operation.
    """
    return self._combine(other, "union", all_=True)

intersect

intersect(other, *, all=False)

Returns a result set that is the INTERSECT (intersection) of the current QuerySet and another QuerySet.

By default, INTERSECT returns only distinct rows (INTERSECT DISTINCT). If all=True, the behavior depends on the backend but typically implies INTERSECT ALL.

PARAMETER	DESCRIPTION
other	The other QuerySet to combine results with. TYPE: QuerySet
all	If True, attempts to perform INTERSECT ALL; otherwise, performs INTERSECT DISTINCT. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
CombinedQuerySet	A CombinedQuerySet instance representing the combined operation.

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

def intersect(self, other: QuerySet, *, all: bool = False) -> CombinedQuerySet:
    """
    Returns a result set that is the INTERSECT (intersection) of the current QuerySet
    and another QuerySet.

    By default, `INTERSECT` returns only distinct rows (INTERSECT DISTINCT).
    If `all=True`, the behavior depends on the backend but typically implies `INTERSECT ALL`.

    Args:
        other: The other `QuerySet` to combine results with.
        all: If `True`, attempts to perform `INTERSECT ALL`; otherwise, performs `INTERSECT DISTINCT`.

    Returns:
        A `CombinedQuerySet` instance representing the combined operation.
    """
    return self._combine(other, "intersect", all_=all)

except_

except_(other, *, all=False)

Returns a result set that is the EXCEPT (difference) of the current QuerySet and another QuerySet (rows in the first but not in the second).

By default, EXCEPT returns only distinct rows (EXCEPT DISTINCT). If all=True, performs EXCEPT ALL.

PARAMETER	DESCRIPTION
other	The other QuerySet to combine results with. TYPE: QuerySet
all	If True, performs EXCEPT ALL; otherwise, performs EXCEPT DISTINCT. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
CombinedQuerySet	A CombinedQuerySet instance representing the combined operation.

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

def except_(self, other: QuerySet, *, all: bool = False) -> CombinedQuerySet:
    """
    Returns a result set that is the EXCEPT (difference) of the current QuerySet
    and another QuerySet (rows in the first but not in the second).

    By default, `EXCEPT` returns only distinct rows (EXCEPT DISTINCT).
    If `all=True`, performs `EXCEPT ALL`.

    Args:
        other: The other `QuerySet` to combine results with.
        all: If `True`, performs `EXCEPT ALL`; otherwise, performs `EXCEPT DISTINCT`.

    Returns:
        A `CombinedQuerySet` instance representing the combined operation.
    """
    return self._combine(other, "except", all_=all)

_sql_helper async

_sql_helper()

Helper method to compile the SQL query represented by the current QuerySet into a string.

This is primarily used for debugging, introspection, or logging the generated SQL. The method ensures that literal bind values (parameters) are included in the final string.

RETURNS	DESCRIPTION
Any	A string containing the compiled SQL query.

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

async def _sql_helper(self) -> Any:
    """
    Helper method to compile the SQL query represented by the current QuerySet into a string.

    This is primarily used for debugging, introspection, or logging the generated SQL.
    The method ensures that literal bind values (parameters) are included in the final string.

    Returns:
        A string containing the compiled SQL query.
    """
    # Use the database context manager to ensure the engine is ready
    async with self.database:
        # 1. Get the SQLAlchemy Selectable object
        selectable = await self.as_select()

        # 2. Compile the selectable object using the database engine
        compiled_sql: Any = selectable.compile(
            self.database.engine,
            compile_kwargs={"literal_binds": True},  # Include parameter values directly
        )
        # Return the compiled statement (which often implicitly converts to a string)
        return compiled_sql

select_for_update

select_for_update(*, nowait=False, skip_locked=False, read=False, key_share=False, of=None)

Request row-level locks on the rows selected by this queryset, using dialect-appropriate SELECT ... FOR UPDATE semantics via SQLAlchemy's with_for_update().

PARAMETER	DESCRIPTION
nowait	Fail immediately if a lock cannot be acquired. TYPE: bool DEFAULT: False
skip_locked	Skip rows that are locked by other transactions (where supported). TYPE: bool DEFAULT: False
read	Shared lock variant (PostgreSQL's FOR SHARE). TYPE: bool DEFAULT: False
key_share	PostgreSQL's FOR KEY SHARE. TYPE: bool DEFAULT: False
of	Models whose tables should be explicitly locked (PostgreSQL's OF ...). The models must be part of the FROM/JOIN set for this query. TYPE: Sequence[type[BaseModelType]] | None DEFAULT: None

Notes

• Most databases require running inside an explicit transaction: async with database.transaction(): ...
• On unsupported dialects (e.g. SQLite), this is a no-op.
• For PostgreSQL, read=True maps to FOR SHARE and key_share=True to FOR KEY SHARE.
• of=[ModelA, ...] restricts locking to specific tables (PostgreSQL only). You should include related models in the query via select_related(...) if you plan to lock them with of=....

RETURNS	DESCRIPTION
QuerySet	A cloned queryset with locking enabled. TYPE: QuerySet

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

def select_for_update(
    self,
    *,
    nowait: bool = False,
    skip_locked: bool = False,
    read: bool = False,
    key_share: bool = False,
    of: Sequence[type[BaseModelType]] | None = None,
) -> QuerySet:
    """
    Request row-level locks on the rows selected by this queryset, using
    dialect-appropriate SELECT ... FOR UPDATE semantics via SQLAlchemy's
    `with_for_update()`.

    Args:
        nowait (bool): Fail immediately if a lock cannot be acquired.
        skip_locked (bool): Skip rows that are locked by other transactions (where supported).
        read (bool): Shared lock variant (PostgreSQL's FOR SHARE).
        key_share (bool):PostgreSQL's FOR KEY SHARE.
        of (Sequence[type[BaseModelType]] | None): Models whose tables should be explicitly locked
            (PostgreSQL's OF ...). The models must be part of the FROM/JOIN set for this query.

    Notes:
        - Most databases require running inside an explicit transaction:
              async with database.transaction():
                  ...
        - On unsupported dialects (e.g. SQLite), this is a no-op.
        - For PostgreSQL, `read=True` maps to FOR SHARE and `key_share=True` to FOR KEY SHARE.
        - `of=[ModelA, ...]` restricts locking to specific tables (PostgreSQL only).
          You should include related models in the query via `select_related(...)`
          if you plan to lock them with `of=...`.

    Returns:
        QuerySet: A cloned queryset with locking enabled.
    """
    queryset: QuerySet = self._clone()
    payload: dict[str, Any] = {
        "nowait": bool(nowait),
        "skip_locked": bool(skip_locked),
        "read": bool(read),
        "key_share": bool(key_share),
    }
    if of:
        # Store model classes here during compilation we map them to actual (possibly aliased) tables
        payload["of"] = tuple(of)
    queryset._for_update = payload
    return queryset

filter

filter(*clauses, **kwargs)

Filters the QuerySet by the given clauses and keyword arguments, combining them with the AND operand.

This is the primary method for constructing the WHERE clause of a query. Multiple clauses and kwargs are implicitly combined using AND.

PARAMETER	DESCRIPTION
*clauses	Positional arguments which can be: - SQLAlchemy Binary Expressions (e.g., Model.field == value). - Callables (sync/async) that accept the QuerySet and return a Binary Expression. - Dictionaries (Django-style lookups, e.g., {"field__gt": 10}). - Nested QuerySets (for subqueries). TYPE: BinaryExpression | Callable[[QuerySetType], BinaryExpression | Awaitable[BinaryExpression]] | dict[str, Any] | QuerySet DEFAULT: ()
**kwargs	Keyword arguments for Django-style lookups (e.g., field__gt=10). TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance with the additional filters applied.

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

def filter(
    self,
    *clauses: sqlalchemy.sql.expression.BinaryExpression
    | Callable[
        [QuerySetType],
        sqlalchemy.sql.expression.BinaryExpression
        | Awaitable[sqlalchemy.sql.expression.BinaryExpression],
    ]
    | dict[str, Any]
    | QuerySet,
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet by the given clauses and keyword arguments, combining them with the AND operand.

    This is the primary method for constructing the WHERE clause of a query. Multiple clauses
    and kwargs are implicitly combined using AND.

    Args:
        *clauses: Positional arguments which can be:
                  - SQLAlchemy Binary Expressions (e.g., `Model.field == value`).
                  - Callables (sync/async) that accept the QuerySet and return a Binary Expression.
                  - Dictionaries (Django-style lookups, e.g., `{"field__gt": 10}`).
                  - Nested QuerySets (for subqueries).
        **kwargs: Keyword arguments for Django-style lookups (e.g., `field__gt=10`).

    Returns:
        A new QuerySet instance with the additional filters applied.
    """
    return self._filter_or_exclude(clauses=clauses, kwargs=kwargs)

all

all(clear_cache=False)

Returns a cloned QuerySet instance, or simply clears the cache of the current instance.

PARAMETER	DESCRIPTION
clear_cache	If True, the method clears the internal cache and returns self. If False (default), it returns a fresh clone with an empty cache. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone or the current QuerySet instance (self).

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

def all(self, clear_cache: bool = False) -> QuerySet:
    """
    Returns a cloned QuerySet instance, or simply clears the cache of the current instance.

    Args:
        clear_cache: If `True`, the method clears the internal cache and returns `self`.
                     If `False` (default), it returns a fresh clone with an empty cache.

    Returns:
        A new QuerySet clone or the current QuerySet instance (`self`).
    """
    if clear_cache:
        self._clear_cache(keep_cached_selected=not self._has_dynamic_clauses)
        return self
    return self._clone()

or_

or_(*clauses, **kwargs)

Filters the QuerySet by the given clauses and keyword arguments, combining them with the OR operand.

This method is used to construct a disjunction (OR logic) for the WHERE clause.

PARAMETER	DESCRIPTION
*clauses	Positional arguments for filtering (same types as filter). TYPE: BinaryExpression | Callable[[QuerySetType], BinaryExpression | Awaitable[BinaryExpression]] | dict[str, Any] | QuerySet DEFAULT: ()
**kwargs	Keyword arguments for filtering (Django-style lookups). TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance with the OR filters applied.

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

def or_(
    self,
    *clauses: sqlalchemy.sql.expression.BinaryExpression
    | Callable[
        [QuerySetType],
        sqlalchemy.sql.expression.BinaryExpression
        | Awaitable[sqlalchemy.sql.expression.BinaryExpression],
    ]
    | dict[str, Any]
    | QuerySet,
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet by the given clauses and keyword arguments, combining them with the OR operand.

    This method is used to construct a disjunction (OR logic) for the WHERE clause.

    Args:
        *clauses: Positional arguments for filtering (same types as `filter`).
        **kwargs: Keyword arguments for filtering (Django-style lookups).

    Returns:
        A new QuerySet instance with the OR filters applied.
    """
    return self._filter_or_exclude(clauses=clauses, or_=True, kwargs=kwargs)

local_or

local_or(*clauses, **kwargs)

Filters the QuerySet using the OR operand, but only applies the OR logic locally.

This prevents the OR operation from becoming a global OR that overrides existing, unrelated AND filters, often by applying the OR clause within parentheses.

PARAMETER	DESCRIPTION
*clauses	Positional arguments for filtering. TYPE: BinaryExpression | Callable[[QuerySetType], BinaryExpression | Awaitable[BinaryExpression]] | dict[str, Any] | QuerySet DEFAULT: ()
**kwargs	Keyword arguments for filtering. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance with the local OR filters applied.

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

def local_or(
    self,
    *clauses: sqlalchemy.sql.expression.BinaryExpression
    | Callable[
        [QuerySetType],
        sqlalchemy.sql.expression.BinaryExpression
        | Awaitable[sqlalchemy.sql.expression.BinaryExpression],
    ]
    | dict[str, Any]
    | QuerySet,
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet using the OR operand, but only applies the OR logic locally.

    This prevents the OR operation from becoming a global OR that overrides existing,
    unrelated AND filters, often by applying the OR clause within parentheses.

    Args:
        *clauses: Positional arguments for filtering.
        **kwargs: Keyword arguments for filtering.

    Returns:
        A new QuerySet instance with the local OR filters applied.
    """
    return self._filter_or_exclude(
        clauses=clauses, or_=True, kwargs=kwargs, allow_global_or=False
    )

and_

and_(*clauses, **kwargs)

Filters the QuerySet by the given clauses and keyword arguments, using the AND operand.

This method is an alias for filter(), explicitly stating the AND logic.

PARAMETER	DESCRIPTION
*clauses	Positional arguments for filtering. TYPE: BinaryExpression | Callable[[QuerySetType], BinaryExpression | Awaitable[BinaryExpression]] | dict[str, Any] DEFAULT: ()
**kwargs	Keyword arguments for filtering. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance with the AND filters applied.

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

def and_(
    self,
    *clauses: sqlalchemy.sql.expression.BinaryExpression
    | Callable[
        [QuerySetType],
        sqlalchemy.sql.expression.BinaryExpression
        | Awaitable[sqlalchemy.sql.expression.BinaryExpression],
    ]
    | dict[str, Any],
    **kwargs: Any,
) -> QuerySet:
    """
    Filters the QuerySet by the given clauses and keyword arguments, using the AND operand.

    This method is an alias for `filter()`, explicitly stating the AND logic.

    Args:
        *clauses: Positional arguments for filtering.
        **kwargs: Keyword arguments for filtering.

    Returns:
        A new QuerySet instance with the AND filters applied.
    """
    return self._filter_or_exclude(clauses=clauses, kwargs=kwargs)

not_

not_(*clauses, **kwargs)

Excludes results from the QuerySet by negating the given clauses and keyword arguments.

This method is an alias for exclude(), explicitly stating the NOT logic.

PARAMETER	DESCRIPTION
*clauses	Positional arguments for exclusion. TYPE: BinaryExpression | Callable[[QuerySetType], BinaryExpression | Awaitable[BinaryExpression]] | dict[str, Any] | QuerySet DEFAULT: ()
**kwargs	Keyword arguments for exclusion. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance with the exclusion applied.

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

def not_(
    self,
    *clauses: sqlalchemy.sql.expression.BinaryExpression
    | Callable[
        [QuerySetType],
        sqlalchemy.sql.expression.BinaryExpression
        | Awaitable[sqlalchemy.sql.expression.BinaryExpression],
    ]
    | dict[str, Any]
    | QuerySet,
    **kwargs: Any,
) -> QuerySet:
    """
    Excludes results from the QuerySet by negating the given clauses and keyword arguments.

    This method is an alias for `exclude()`, explicitly stating the NOT logic.

    Args:
        *clauses: Positional arguments for exclusion.
        **kwargs: Keyword arguments for exclusion.

    Returns:
        A new QuerySet instance with the exclusion applied.
    """
    return self.exclude(*clauses, **kwargs)

exclude

exclude(*clauses, **kwargs)

Excludes results from the QuerySet by negating the given clauses and keyword arguments.

The exclusion logic is typically implemented by calling _filter_or_exclude with the exclude=True flag.

PARAMETER	DESCRIPTION
*clauses	Positional arguments for exclusion (same types as filter). TYPE: BinaryExpression | Callable[[QuerySetType], BinaryExpression | Awaitable[BinaryExpression]] | dict[str, Any] | QuerySet DEFAULT: ()
**kwargs	Keyword arguments for exclusion (Django-style lookups). TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance with the exclusion applied.

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

def exclude(
    self,
    *clauses: sqlalchemy.sql.expression.BinaryExpression
    | Callable[
        [QuerySetType],
        sqlalchemy.sql.expression.BinaryExpression
        | Awaitable[sqlalchemy.sql.expression.BinaryExpression],
    ]
    | dict[str, Any]
    | QuerySet,
    **kwargs: Any,
) -> QuerySet:
    """
    Excludes results from the QuerySet by negating the given clauses and keyword arguments.

    The exclusion logic is typically implemented by calling `_filter_or_exclude`
    with the `exclude=True` flag.

    Args:
        *clauses: Positional arguments for exclusion (same types as `filter`).
        **kwargs: Keyword arguments for exclusion (Django-style lookups).

    Returns:
        A new QuerySet instance with the exclusion applied.
    """
    return self._filter_or_exclude(clauses=clauses, exclude=True, kwargs=kwargs)

exclude_secrets

exclude_secrets(exclude_secrets=True)

Marks the QuerySet to exclude any model fields declared with secret=True from being leaked or serialized in the final result set.

PARAMETER	DESCRIPTION
exclude_secrets	If True (default), secrets are excluded; if False, they are included. TYPE: bool DEFAULT: True

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the _exclude_secrets flag set.

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

def exclude_secrets(
    self,
    exclude_secrets: bool = True,
) -> QuerySet:
    """
    Marks the QuerySet to exclude any model fields declared with `secret=True` from being leaked
    or serialized in the final result set.

    Args:
        exclude_secrets: If `True` (default), secrets are excluded; if `False`, they are included.

    Returns:
        A new QuerySet clone with the `_exclude_secrets` flag set.
    """
    queryset = self._clone()
    queryset._exclude_secrets = exclude_secrets
    return queryset

extra_select

extra_select(*extra)

Adds extra columns or expressions to the SELECT statement.

PARAMETER	DESCRIPTION
*extra	Additional SQLAlchemy column clauses or expressions. TYPE: ColumnClause DEFAULT: ()

RETURNS	DESCRIPTION
QuerySetType	A new QuerySet with the extra select clauses. TYPE: QuerySetType

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

def extra_select(
    self,
    *extra: sqlalchemy.ColumnClause,
) -> QuerySetType:
    """
    Adds extra columns or expressions to the SELECT statement.

    Args:
        *extra (sqlalchemy.ColumnClause): Additional SQLAlchemy column clauses or expressions.

    Returns:
        QuerySetType: A new QuerySet with the extra select clauses.
    """
    queryset = self._clone()
    queryset._extra_select.extend(extra)
    return queryset

reference_select

reference_select(references)

Adds references to the SELECT statement, used for specific model field handling.

PARAMETER	DESCRIPTION
references	A dictionary defining reference selections. TYPE: reference_select_type

RETURNS	DESCRIPTION
QuerySetType	A new QuerySet with the added reference selections. TYPE: QuerySetType

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

def reference_select(self, references: reference_select_type) -> QuerySetType:
    """
    Adds references to the SELECT statement, used for specific model field handling.

    Args:
        references (reference_select_type): A dictionary defining reference selections.

    Returns:
        QuerySetType: A new QuerySet with the added reference selections.
    """
    queryset = self._clone()
    queryset._reference_select.update(references)
    return queryset

batch_size

batch_size(batch_size=None)

Sets the batch or chunk size for results. This is primarily used in conjunction with iteration methods (like QuerySet.iterate()) to control memory usage.

PARAMETER	DESCRIPTION
batch_size	The number of records to fetch in each database round-trip. If None, iteration may fetch all results at once (depending on the backend). TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the batch size set.

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

def batch_size(
    self,
    batch_size: int | None = None,
) -> QuerySet:
    """
    Sets the batch or chunk size for results. This is primarily used in conjunction
    with iteration methods (like `QuerySet.iterate()`) to control memory usage.

    Args:
        batch_size: The number of records to fetch in each database round-trip.
                    If `None`, iteration may fetch all results at once (depending on the backend).

    Returns:
        A new QuerySet clone with the batch size set.
    """
    queryset = self._clone()
    queryset._batch_size = batch_size
    return queryset

lookup

lookup(term)

Performs a broader, case-insensitive search for a given term across all CharField and TextField instances defined on the model.

The search uses the SQL LIKE operator with wildcards (%) and typically maps to ILIKE on PostgreSQL for case-insensitivity.

PARAMETER	DESCRIPTION
term	The search term to look for. TYPE: Any

RETURNS	DESCRIPTION
QuerySet	A new QuerySet instance with the combined OR filter applied to all searchable fields.

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

def lookup(self, term: Any) -> QuerySet:
    """
    Performs a broader, case-insensitive search for a given term across all
    CharField and TextField instances defined on the model.

    The search uses the SQL `LIKE` operator with wildcards (`%`) and typically
    maps to `ILIKE` on PostgreSQL for case-insensitivity.

    Args:
        term: The search term to look for.

    Returns:
        A new QuerySet instance with the combined OR filter applied to all searchable fields.
    """
    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.meta.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.

Sorting is applied in ascending order by default. Prepending a field name with a hyphen (-) specifies descending order (e.g., -created_at).

PARAMETER	DESCRIPTION
*order_by	One or more field names to sort by. TYPE: str DEFAULT: ()

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the specified ordering.

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

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

    Sorting is applied in ascending order by default. Prepending a field name with
    a hyphen (`-`) specifies descending order (e.g., `-created_at`).

    Args:
        *order_by: One or more field names to sort by.

    Returns:
        A new QuerySet clone with the specified ordering.
    """
    queryset: QuerySet = self._clone()
    queryset._order_by = order_by
    if queryset._update_select_related_weak(order_by, clear=True):
        queryset._update_select_related_weak(queryset._group_by, clear=False)
    return queryset

reverse

reverse()

Reverses the established order of the QuerySet.

If an order_by is already set, it negates the sorting direction of every field. If no order_by is set, it defaults to reversing the primary key columns. It also adjusts internal cache pointers for consistency if caching is active.

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the reversed ordering.

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

def reverse(self) -> QuerySet:
    """
    Reverses the established order of the QuerySet.

    If an `order_by` is already set, it negates the sorting direction of every field.
    If no `order_by` is set, it defaults to reversing the primary key columns.
    It also adjusts internal cache pointers for consistency if caching is active.

    Returns:
        A new QuerySet clone with the reversed ordering.
    """
    if not self._order_by:
        queryset = self.order_by(*self.model_class.pkcolumns)
    else:
        queryset = self._clone()
    queryset._order_by = tuple(
        el[1:] if el.startswith("-") else f"-{el}" for el in queryset._order_by
    )
    queryset._cache_last = self._cache_first
    queryset._cache_first = self._cache_last
    queryset._cache_count = self._cache_count
    if self._cache_fetch_all:
        # we may have embedded active
        cache_keys = []
        values = []
        for k, v in reversed(self._cache.get_category(self.model_class).items()):
            cache_keys.append(k)
            values.append(v)
        queryset._cache.update(self.model_class, values, cache_keys=cache_keys)
        queryset._cache_fetch_all = True
    return queryset

limit

limit(limit_count)

Limits the number of results returned by the QuerySet (SQL LIMIT clause).

PARAMETER	DESCRIPTION
limit_count	The maximum number of rows to return. TYPE: int

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the limit applied.

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

def limit(self, limit_count: int) -> QuerySet:
    """
    Limits the number of results returned by the QuerySet (SQL LIMIT clause).

    Args:
        limit_count: The maximum number of rows to return.

    Returns:
        A new QuerySet clone with the limit applied.
    """
    queryset: QuerySet = self._clone()
    queryset.limit_count = limit_count
    return queryset

offset

offset(offset)

Skips the specified number of results before starting to return rows (SQL OFFSET clause).

PARAMETER	DESCRIPTION
offset	The number of rows to skip. TYPE: int

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the offset applied.

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

def offset(self, offset: int) -> QuerySet:
    """
    Skips the specified number of results before starting to return rows (SQL OFFSET clause).

    Args:
        offset: The number of rows to skip.

    Returns:
        A new QuerySet clone with the offset applied.
    """
    queryset: QuerySet = self._clone()
    queryset._offset = offset
    return queryset

group_by

group_by(*group_by)

Groups the results of the QuerySet by the given fields (SQL GROUP BY clause).

PARAMETER	DESCRIPTION
*group_by	One or more field names to group the results by. TYPE: str DEFAULT: ()

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the grouping applied.

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

def group_by(self, *group_by: str) -> QuerySet:
    """
    Groups the results of the QuerySet by the given fields (SQL GROUP BY clause).

    Args:
        *group_by: One or more field names to group the results by.

    Returns:
        A new QuerySet clone with the grouping applied.
    """
    queryset: QuerySet = self._clone()
    queryset._group_by = group_by
    if queryset._update_select_related_weak(group_by, clear=True):
        queryset._update_select_related_weak(queryset._order_by, clear=False)
    return queryset

distinct

distinct(first=True, *distinct_on)

Returns a queryset with distinct results.

PARAMETER	DESCRIPTION
first	If True, applies DISTINCT. If False, removes any distinct clause. If a string, applies DISTINCT ON to that field. TYPE: bool | str DEFAULT: True
*distinct_on	Additional fields for DISTINCT ON. TYPE: str DEFAULT: ()

RETURNS	DESCRIPTION
QuerySet	A new QuerySet with the distinct clause applied. TYPE: QuerySet

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

def distinct(self, first: bool | str = True, *distinct_on: str) -> QuerySet:
    """
    Returns a queryset with distinct results.

    Args:
        first (bool | str): If True, applies `DISTINCT`. If False, removes any distinct clause.
                            If a string, applies `DISTINCT ON` to that field.
        *distinct_on (str): Additional fields for `DISTINCT ON`.

    Returns:
        QuerySet: A new QuerySet with the distinct clause applied.
    """
    queryset: QuerySet = self._clone()
    if first is False:
        queryset.distinct_on = None
    elif first is True:
        queryset.distinct_on = []
    else:
        queryset.distinct_on = [first, *distinct_on]
    return queryset

only

only(*fields)

Restricts the QuerySet to retrieve only the specified fields from the database for the model instances, along with the primary key field(s).

This method is used for performance optimization when only a subset of columns is needed. The primary key is automatically included to ensure object identity and saving functionality.

PARAMETER	DESCRIPTION
*fields	The names of the model fields (columns) to include in the SELECT statement. TYPE: str DEFAULT: ()

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the _only set attribute containing the selected fields.

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

def only(self, *fields: str) -> QuerySet:
    """
    Restricts the QuerySet to retrieve **only** the specified fields from the database
    for the model instances, along with the primary key field(s).

    This method is used for performance optimization when only a subset of columns is needed.
    The primary key is automatically included to ensure object identity and saving functionality.

    Args:
        *fields: The names of the model fields (columns) to include in the SELECT statement.

    Returns:
        A new QuerySet clone with the `_only` set attribute containing the selected fields.
    """
    queryset: QuerySet = self._clone()
    only_fields = set(fields)
    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):
                    only_fields.add(pkcolumn.key)
    else:
        for pkcolumn in self.model_class.pkcolumns:
            only_fields.add(pkcolumn.key)
    queryset._only = only_fields
    return queryset

defer

defer(*fields)

Excludes the specified fields from being retrieved from the database.

When accessing a deferred field on a model instance, a subsequent database query will be triggered to fetch its value. This is useful for large, rarely accessed fields (e.g., large text blobs).

PARAMETER	DESCRIPTION
*fields	The names of the model fields (columns) to exclude from the SELECT statement. TYPE: str DEFAULT: ()

RETURNS	DESCRIPTION
QuerySet	A new QuerySet clone with the _defer set attribute containing the fields to skip.

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

def defer(self, *fields: str) -> QuerySet:
    """
    Excludes the specified fields from being retrieved from the database.

    When accessing a deferred field on a model instance, a subsequent database query
    will be triggered to fetch its value. This is useful for large, rarely accessed
    fields (e.g., large text blobs).

    Args:
        *fields: The names of the model fields (columns) to **exclude** from the SELECT statement.

    Returns:
        A new QuerySet clone with the `_defer` set attribute containing the fields to skip.
    """
    queryset: QuerySet = self._clone()

    queryset._defer = set(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.

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

def select_related(self, *related: str) -> 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 len(related) >= 1 and not isinstance(cast(Any, related[0]), str):
        warnings.warn(
            "use `select_related` with variadic str arguments instead of a Sequence",
            DeprecationWarning,
            stacklevel=2,
        )
        related = cast(tuple[str, ...], related[0])
    queryset._update_select_related(related)
    return queryset

values async

values(fields=None, exclude=None, exclude_none=False)

Executes the query and returns the results as a list of Python dictionaries, rather than model instances.

PARAMETER	DESCRIPTION
fields	A sequence of field names to include in the resulting dictionaries. If None, all model fields are included. TYPE: Sequence[str] | str | None DEFAULT: None
exclude	A sequence of field names to exclude from the resulting dictionaries. TYPE: Sequence[str] | set[str] DEFAULT: None
exclude_none	If True, fields with a value of None are omitted from the dictionaries. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[dict]	A list of dictionaries representing the selected data rows.

RAISES	DESCRIPTION
QuerySetError	If the fields argument is not a suitable sequence.

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

async def values(
    self,
    fields: Sequence[str] | str | None = None,
    exclude: Sequence[str] | set[str] = None,
    exclude_none: bool = False,
) -> list[dict]:
    """
    Executes the query and returns the results as a list of Python dictionaries,
    rather than model instances.

    Args:
        fields: A sequence of field names to include in the resulting dictionaries.
                If `None`, all model fields are included.
        exclude: A sequence of field names to exclude from the resulting dictionaries.
        exclude_none: If `True`, fields with a value of `None` are omitted from the dictionaries.

    Returns:
        A list of dictionaries representing the selected data rows.

    Raises:
        QuerySetError: If the `fields` argument is not a suitable sequence.
    """

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

    if fields is not None and not isinstance(fields, Iterable):
        raise QuerySetError(detail="Fields must be a suitable sequence of strings or unset.")

    rows: list[BaseModelType] = await self
    if fields:
        return [
            row.model_dump(exclude=exclude, exclude_none=exclude_none, include=fields)
            for row in rows
        ]
    else:
        return [row.model_dump(exclude=exclude, exclude_none=exclude_none) for row in rows]

values_list async

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

Executes the query and returns the results as a list of tuples or, if flat=True, a flat list of values for a single field.

PARAMETER	DESCRIPTION
fields	A sequence of field names to include. Must contain exactly one field if flat=True. TYPE: Sequence[str] | str | None DEFAULT: None
exclude	A sequence of field names to exclude. TYPE: Sequence[str] | set[str] DEFAULT: None
exclude_none	If True, fields with a value of None are omitted during dictionary creation. TYPE: bool DEFAULT: False
flat	If True and only one field is selected, returns a list of values instead of tuples/dictionaries. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[Any]	A list of tuples, or a flat list of values if flat=True.

RAISES	DESCRIPTION
QuerySetError	If flat=True but more than one field is selected, or if the selected field does not exist.

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

async def values_list(
    self,
    fields: Sequence[str] | str | None = None,
    exclude: Sequence[str] | set[str] = None,
    exclude_none: bool = False,
    flat: bool = False,
) -> list[Any]:
    """
    Executes the query and returns the results as a list of tuples or, if `flat=True`,
    a flat list of values for a single field.

    Args:
        fields: A sequence of field names to include. Must contain exactly one field if `flat=True`.
        exclude: A sequence of field names to exclude.
        exclude_none: If `True`, fields with a value of `None` are omitted during dictionary creation.
        flat: If `True` and only one field is selected, returns a list of values instead of tuples/dictionaries.

    Returns:
        A list of tuples, or a flat list of values if `flat=True`.

    Raises:
        QuerySetError: If `flat=True` but more than one field is selected, or if the selected field does not exist.
    """
    if isinstance(fields, str):
        fields = [fields]
    rows = await self.values(
        fields=fields,
        exclude=exclude,
        exclude_none=exclude_none,
    )
    if not rows:
        return []
    if not flat:
        return [tuple(row.values()) for row in rows]
    else:
        try:
            return [row[fields[0]] for row in rows]
        except KeyError:
            raise QuerySetError(detail=f"{fields[0]} does not exist in the results.") from None

exists async

exists(**kwargs)

Returns a boolean indicating if one or more records matching the QuerySet's criteria exists.

If keyword arguments are provided, it first checks the cache for an existence match.

PARAMETER	DESCRIPTION
**kwargs	Optional filters to apply before checking for existence (e.g., pk=1). TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
bool	True if at least one record exists, False otherwise.

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

async def exists(self, **kwargs: Any) -> bool:
    """
    Returns a boolean indicating if one or more records matching the QuerySet's criteria exists.

    If keyword arguments are provided, it first checks the cache for an existence match.

    Args:
        **kwargs: Optional filters to apply before checking for existence (e.g., `pk=1`).

    Returns:
        True if at least one record exists, False otherwise.
    """
    if kwargs:
        # check cache for existance
        cached = self._cache.get(self.model_class, kwargs)
        if cached is not None:
            return True
        filter_query = self.filter(**kwargs)
        filter_query._cache = self._cache
        return await filter_query.exists()
    queryset: QuerySet = self
    expression = (await queryset.as_select()).exists().select()
    check_db_connection(queryset.database)
    async with queryset.database as database:
        _exists = await database.fetch_val(expression)
    return cast(bool, _exists)

count async

count()

Executes a SELECT COUNT statement and returns the total number of records matching the query.

The result is cached internally to prevent redundant database calls for subsequent counts.

RETURNS	DESCRIPTION
int	The total number of records as an integer.

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

async def count(self) -> int:
    """
    Executes a SELECT COUNT statement and returns the total number of records matching the query.

    The result is cached internally to prevent redundant database calls for subsequent counts.

    Returns:
        The total number of records as an integer.
    """
    if self._cache_count is not None:
        return self._cache_count
    queryset: QuerySet = self
    expression = (await queryset.as_select()).alias("subquery_for_count")
    expression = sqlalchemy.func.count().select().select_from(expression)
    check_db_connection(queryset.database)
    async with queryset.database as database:
        self._cache_count = count = cast("int", await database.fetch_val(expression))
    return count

get_or_none async

get_or_none(**kwargs)

Fetches a single object matching the parameters.

If no object is found (raises ObjectNotFound), returns None.

PARAMETER	DESCRIPTION
**kwargs	Filters to identify the single object. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
EdgyEmbedTarget | None	The matching model instance, or None.

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

async def get_or_none(self, **kwargs: Any) -> EdgyEmbedTarget | None:
    """
    Fetches a single object matching the parameters.

    If no object is found (raises `ObjectNotFound`), returns `None`.

    Args:
        **kwargs: Filters to identify the single object.

    Returns:
        The matching model instance, or `None`.
    """
    try:
        return await self.get(**kwargs)
    except ObjectNotFound:
        return None

get async

get(**kwargs)

Fetches a single object matching the parameters.

PARAMETER	DESCRIPTION
**kwargs	Filters to identify the single object. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
EdgyEmbedTarget	The matching model instance.

RAISES	DESCRIPTION
ObjectNotFound	If no object is found.
MultipleObjectsReturned	If more than one object is found (implicitly handled by underlying _get_raw).

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

async def get(self, **kwargs: Any) -> EdgyEmbedTarget:
    """
    Fetches a single object matching the parameters.

    Args:
        **kwargs: Filters to identify the single object.

    Returns:
        The matching model instance.

    Raises:
        ObjectNotFound: If no object is found.
        MultipleObjectsReturned: If more than one object is found (implicitly handled by underlying `_get_raw`).
    """
    return cast(EdgyEmbedTarget, (await self._get_raw(**kwargs))[1])

first async

first()

Returns the first record from the QuerySet, respecting any ordering and limits.

If the count is zero or the first record is already cached, it returns the cached result.

RETURNS	DESCRIPTION
EdgyEmbedTarget | None	The first model instance, or None if the QuerySet is empty.

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

async def first(self) -> EdgyEmbedTarget | None:
    """
    Returns the first record from the QuerySet, respecting any ordering and limits.

    If the count is zero or the first record is already cached, it returns the cached result.

    Returns:
        The first model instance, or `None` if the QuerySet is empty.
    """
    if self._cache_count is not None and self._cache_count == 0:
        return None
    if self._cache_first is not None:
        return cast(EdgyEmbedTarget, self._cache_first[1])
    queryset = self
    if not queryset._order_by:
        queryset = queryset.order_by(*self.model_class.pkcolumns)
    expression, tables_and_models = await queryset.as_select_with_tables()
    self._cached_select_related_expression = queryset._cached_select_related_expression
    check_db_connection(queryset.database)
    async with queryset.database as database:
        row = await database.fetch_one(expression, pos=0)
    if row:
        parser = ResultParser(self)
        result_tuple: tuple[Any, EdgyEmbedTarget] = await parser.row_to_model(
            row, tables_and_models
        )
        self._cache_first = result_tuple
        return result_tuple[1]
    return None

last async

last()

Returns the last record from the QuerySet...

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

async def last(self) -> EdgyEmbedTarget | None:
    """
    Returns the last record from the QuerySet...
    """
    if self._cache_count is not None and self._cache_count == 0:
        return None
    if self._cache_last is not None:
        return cast(EdgyEmbedTarget, self._cache_last[1])
    queryset = self
    if not queryset._order_by:
        queryset = queryset.order_by(*self.model_class.pkcolumns)
    queryset = queryset.reverse()
    expression, tables_and_models = await queryset.as_select_with_tables()
    self._cached_select_related_expression = queryset._cached_select_related_expression
    check_db_connection(queryset.database)
    async with queryset.database as database:
        row = await database.fetch_one(expression, pos=0)
    if row:
        # NEW FIXED LINES:
        parser = ResultParser(self)
        result_tuple: tuple[Any, EdgyEmbedTarget] = await parser.row_to_model(
            row, tables_and_models
        )
        self._cache_last = result_tuple
        return result_tuple[1]
    return None

create async

create(*args, **kwargs)

Creates and saves a single record in the database table associated with the QuerySet's model.

PARAMETER	DESCRIPTION
*args	Positional arguments for model instantiation. TYPE: Any DEFAULT: ()
**kwargs	Keyword arguments for model instantiation and field values. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
EdgyEmbedTarget	The newly created model instance.

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

async def create(self, *args: Any, **kwargs: Any) -> EdgyEmbedTarget:
    """
    Creates and saves a single record in the database table associated with the QuerySet's model.

    Args:
        *args: Positional arguments for model instantiation.
        **kwargs: Keyword arguments for model instantiation and field values.

    Returns:
        The newly created model instance.
    """
    # for tenancy
    queryset: QuerySet = self._clone()
    check_db_connection(queryset.database)
    token = CHECK_DB_CONNECTION_SILENCED.set(True)
    try:
        instance = queryset.model_class(*args, **kwargs)
        # apply_instance_extras filters out table Alias
        apply_instance_extras(
            instance,
            self.model_class,
            schema=self.using_schema,
            table=queryset.table,
            database=queryset.database,
        )
        # values=set(kwargs.keys()) is required for marking the provided kwargs as explicit provided kwargs
        token2 = CURRENT_INSTANCE.set(self)
        try:
            instance = await instance.real_save(force_insert=True, values=set(kwargs.keys()))
        finally:
            CURRENT_INSTANCE.reset(token2)
        result = await self._embed_parent_in_result(instance)
        self._clear_cache(keep_result_cache=True)
        self._cache.update(
            self.model_class,
            [result],
            cache_keys=[self._cache.create_cache_key(self.model_class, result[0])],
        )
        return cast(EdgyEmbedTarget, result[1])
    finally:
        CHECK_DB_CONNECTION_SILENCED.reset(token)

bulk_create async

bulk_create(objs)

Bulk creates multiple records in a single batch operation.

This method bypasses model-level save hooks (except for pre/post-save) for efficiency, and returns None.

PARAMETER	DESCRIPTION
objs	An iterable of dictionaries or model instances to be created. TYPE: Iterable[dict[str, Any] | EdgyModel]

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

async def bulk_create(self, objs: Iterable[dict[str, Any] | EdgyModel]) -> None:
    """
    Bulk creates multiple records in a single batch operation.

    This method bypasses model-level save hooks (except for pre/post-save) for efficiency,
    and returns `None`.

    Args:
        objs: An iterable of dictionaries or model instances to be created.
    """
    queryset: QuerySet = self._clone()

    new_objs: list[EdgyModel] = []

    async def _iterate(obj_or_dict: EdgyModel | dict[str, Any]) -> dict[str, Any]:
        if isinstance(obj_or_dict, dict):
            obj: EdgyModel = queryset.model_class(**obj_or_dict)
            if (
                self.model_class.meta.post_save_fields
                and not self.model_class.meta.post_save_fields.isdisjoint(obj_or_dict.keys())
            ):
                new_objs.append(obj)
        else:
            obj = obj_or_dict
            if self.model_class.meta.post_save_fields:
                new_objs.append(obj)
        original = obj.extract_db_fields()
        col_values: dict[str, Any] = obj.extract_column_values(
            original, phase="prepare_insert", instance=self, model_instance=obj
        )
        col_values.update(
            await obj.execute_pre_save_hooks(col_values, original, is_update=False)
        )
        return col_values

    check_db_connection(queryset.database)
    token = CURRENT_INSTANCE.set(self)
    try:
        async with queryset.database as database, database.transaction():
            expression = queryset.table.insert().values([await _iterate(obj) for obj in objs])
            await database.execute_many(expression)
        self._clear_cache(keep_result_cache=True)
        if new_objs:
            keys = self.model_class.meta.fields.keys()
            await run_concurrently(
                [obj.execute_post_save_hooks(keys, is_update=False) for obj in new_objs],
                limit=1 if getattr(queryset.database, "force_rollback", False) else None,
            )
    finally:
        CURRENT_INSTANCE.reset(token)

bulk_update async

bulk_update(objs, fields)

Bulk updates records in a table based on the provided list of model instances and fields.

The primary key of each model instance is used to identify the record to update. This operation is performed within a database transaction.

PARAMETER	DESCRIPTION
objs	A list of existing model instances to update. TYPE: list[EdgyModel]
fields	A list of field names that should be updated across all instances. TYPE: list[str]

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

async def bulk_update(self, objs: list[EdgyModel], fields: list[str]) -> None:
    """
    Bulk updates records in a table based on the provided list of model instances and fields.

    The primary key of each model instance is used to identify the record to update.
    This operation is performed within a database transaction.

    Args:
        objs: A list of existing model instances to update.
        fields: A list of field names that should be updated across all instances.
    """
    queryset: QuerySet = self._clone()
    fields = list(fields)

    pk_query_placeholder = (
        getattr(queryset.table.c, pkcol)
        == sqlalchemy.bindparam(
            "__id" if pkcol == "id" else pkcol, type_=getattr(queryset.table.c, pkcol).type
        )
        for pkcol in queryset.pkcolumns
    )
    expression = queryset.table.update().where(*pk_query_placeholder)

    update_list = []
    fields_plus_pk = {*fields, *queryset.model_class.pkcolumns}
    check_db_connection(queryset.database)
    token = CURRENT_INSTANCE.set(self)
    try:
        async with queryset.database as database, database.transaction():
            for obj in objs:
                extracted = obj.extract_db_fields(fields_plus_pk)
                update = queryset.model_class.extract_column_values(
                    extracted,
                    is_update=True,
                    is_partial=True,
                    phase="prepare_update",
                    instance=self,
                    model_instance=obj,
                )
                update.update(
                    await obj.execute_pre_save_hooks(update, extracted, is_update=True)
                )
                if "id" in update:
                    update["__id"] = update.pop("id")
                update_list.append(update)

            values_placeholder: Any = {
                pkcol: sqlalchemy.bindparam(pkcol, type_=getattr(queryset.table.c, pkcol).type)
                for field in fields
                for pkcol in queryset.model_class.meta.field_to_column_names[field]
            }
            expression = expression.values(values_placeholder)
            await database.execute_many(expression, update_list)
        self._clear_cache()
        if (
            self.model_class.meta.post_save_fields
            and not self.model_class.meta.post_save_fields.isdisjoint(fields)
        ):
            await run_concurrently(
                [obj.execute_post_save_hooks(fields, is_update=True) for obj in objs],
                limit=1 if getattr(queryset.database, "force_rollback", False) else None,
            )
    finally:
        CURRENT_INSTANCE.reset(token)

bulk_get_or_create async

bulk_get_or_create(objs, unique_fields=None)

Bulk gets or creates records in a table.

If records exist based on unique fields, they are retrieved. Otherwise, new records are created.

PARAMETER	DESCRIPTION
objs	A list of objects or dictionaries. TYPE: list[Union[dict[str, Any], EdgyModel]]
unique_fields	Fields that determine uniqueness. If None, all records are treated as new. TYPE: list[str] | None DEFAULT: None

RETURNS	DESCRIPTION
list[EdgyModel]	list[EdgyModel]: A list of retrieved or newly created objects.

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

async def bulk_get_or_create(
    self,
    objs: list[dict[str, Any] | EdgyModel],
    unique_fields: list[str] | None = None,
) -> list[EdgyModel]:
    """
    Bulk gets or creates records in a table.

    If records exist based on unique fields, they are retrieved.
    Otherwise, new records are created.

    Args:
        objs (list[Union[dict[str, Any], EdgyModel]]): A list of objects or dictionaries.
        unique_fields (list[str] | None): Fields that determine uniqueness. If None, all records are treated as new.

    Returns:
        list[EdgyModel]: A list of retrieved or newly created objects.
    """
    queryset: QuerySet = self._clone()
    new_objs: list[EdgyModel] = []
    retrieved_objs: list[EdgyModel] = []
    check_db_connection(queryset.database)

    if unique_fields:
        existing_records: dict[tuple, EdgyModel] = {}
        for obj in objs:
            filter_kwargs = {}
            dict_fields = {}
            if isinstance(obj, dict):
                for field in unique_fields:
                    if field in obj:
                        value = obj[field]
                        if isinstance(value, dict):
                            dict_fields[field] = value
                        else:
                            filter_kwargs[field] = value
            else:
                for field in unique_fields:
                    value = getattr(obj, field)
                    if isinstance(value, dict):
                        dict_fields[field] = value
                    else:
                        filter_kwargs[field] = value
            lookup_key = _extract_unique_lookup_key(obj, unique_fields)
            if lookup_key is not None and lookup_key in existing_records:
                continue
            found = False
            # This fixes edgy-guardian bug when using databasez.iterate indirectly and
            # is safe in case force_rollback is active
            # Models can also issue loads by accessing attrs for building unique_fields
            # For limiting use something like QuerySet.limit(100).bulk_get_or_create(...)
            for model in await queryset.filter(**filter_kwargs):
                if all(getattr(model, k) == expected for k, expected in dict_fields.items()):
                    lookup_key = _extract_unique_lookup_key(model, unique_fields)
                    assert lookup_key is not None, "invalid fields/attributes in unique_fields"
                    if lookup_key not in existing_records:
                        existing_records[lookup_key] = model
                    found = True
                    break
            if found is False:
                new_objs.append(queryset.model_class(**obj) if isinstance(obj, dict) else obj)

        retrieved_objs.extend(existing_records.values())
    else:
        new_objs.extend(
            [queryset.model_class(**obj) if isinstance(obj, dict) else obj for obj in objs]
        )

    async def _iterate(obj: EdgyModel) -> dict[str, Any]:
        original = obj.extract_db_fields()
        col_values: dict[str, Any] = obj.extract_column_values(
            original, phase="prepare_insert", instance=self
        )
        col_values.update(
            await obj.execute_pre_save_hooks(col_values, original, is_update=False)
        )
        return col_values

    token = CURRENT_INSTANCE.set(self)

    try:
        async with queryset.database as database, database.transaction():
            if new_objs:
                new_obj_values = [await _iterate(obj) for obj in new_objs]
                expression = queryset.table.insert().values(new_obj_values)
                await database.execute_many(expression)
                retrieved_objs.extend(new_objs)

            self._clear_cache()
            keys = self.model_class.meta.fields.keys()
            await run_concurrently(
                [obj.execute_post_save_hooks(keys, is_update=False) for obj in new_objs],
                limit=1 if getattr(queryset.database, "force_rollback", False) else None,
            )
    finally:
        CURRENT_INSTANCE.reset(token)

    return retrieved_objs

delete async

delete(use_models=False)

Deletes records from the database.

This method triggers pre_delete and post_delete signals. If use_models is True or the model has specific deletion requirements, it performs a model-based deletion.

PARAMETER	DESCRIPTION
use_models	If True, deletion is performed by iterating and deleting individual model instances. Defaults to False. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
int	The number of rows deleted. TYPE: int

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

async def delete(self, use_models: bool = False) -> int:
    """
    Deletes records from the database.

    This method triggers `pre_delete` and `post_delete` signals.
    If `use_models` is True or the model has specific deletion requirements,
    it performs a model-based deletion.

    Args:
        use_models (bool): If True, deletion is performed by iterating and
                           deleting individual model instances. Defaults to False.

    Returns:
        int: The number of rows deleted.
    """
    await self.model_class.meta.signals.pre_delete.send_async(
        self.model_class, instance=self, model_instance=None
    )
    row_count = await self.raw_delete(use_models=use_models, remove_referenced_call=False)
    await self.model_class.meta.signals.post_delete.send_async(
        self.model_class, instance=self, model_instance=None, row_count=row_count
    )
    return row_count

update async

update(**kwargs)

Updates records in a specific table with the given keyword arguments, matching the QuerySet's filters.

This performs a database-level update operation without fetching and saving model instances.

Warning: - Does not execute instance-level pre_save_callback/post_save_callback hooks. - Values are processed directly for column mapping and validation but do not pass through model instance saving.

PARAMETER	DESCRIPTION
**kwargs	The field names and new values to apply to the matching records. TYPE: Any DEFAULT: {}

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

async def update(self, **kwargs: Any) -> None:
    """
    Updates records in a specific table with the given keyword arguments, matching the QuerySet's filters.

    This performs a database-level update operation without fetching and saving model instances.

    Warning:
    - **Does not** execute instance-level `pre_save_callback`/`post_save_callback` hooks.
    - Values are processed directly for column mapping and validation but do not pass through model instance saving.

    Args:
        **kwargs: The field names and new values to apply to the matching records.
    """

    column_values = self.model_class.extract_column_values(
        kwargs, is_update=True, is_partial=True, phase="prepare_update", instance=self
    )

    # Broadcast the initial update details
    # add is_update to match save
    await self.model_class.meta.signals.pre_update.send_async(
        self.model_class,
        instance=self,
        model_instance=None,
        values=kwargs,
        column_values=column_values,
        is_update=True,
        is_migration=False,
    )

    expression = self.table.update().values(**column_values)
    expression = expression.where(await self.build_where_clause())
    check_db_connection(self.database)
    async with self.database as database:
        await database.execute(expression)

    # Broadcast the update executed
    # add is_update to match save
    await self.model_class.meta.signals.post_update.send_async(
        self.model_class,
        instance=self,
        model_instance=None,
        values=kwargs,
        column_values=column_values,
        is_update=True,
        is_migration=False,
    )
    self._clear_cache()

get_or_create async

get_or_create(defaults=None, *args, **kwargs)

Fetches a single object matching kwargs. If found, returns the object and False. If not found, creates a new object using kwargs (with defaults applied) and returns it with True.

PARAMETER	DESCRIPTION
defaults	Optional dictionary of values to use if a new object must be created. Can also be a ModelRef instance to set a related object upon creation. TYPE: dict[str, Any] | Any | None DEFAULT: None
*args	Positional arguments for model creation if object is not found. TYPE: Any DEFAULT: ()
**kwargs	Filters used to attempt retrieval, and primary creation arguments if not found. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
EdgyEmbedTarget	A tuple containing the fetched/created model instance and a boolean indicating
bool	if the object was created (True) or fetched (False).

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

async def get_or_create(
    self, defaults: dict[str, Any] | Any | None = None, *args: Any, **kwargs: Any
) -> tuple[EdgyEmbedTarget, bool]:
    """
    Fetches a single object matching `kwargs`. If found, returns the object and `False`.
    If not found, creates a new object using `kwargs` (with `defaults` applied) and returns it with `True`.

    Args:
        defaults: Optional dictionary of values to use if a new object must be created.
                  Can also be a `ModelRef` instance to set a related object upon creation.
        *args: Positional arguments for model creation if object is not found.
        **kwargs: Filters used to attempt retrieval, and primary creation arguments if not found.

    Returns:
        A tuple containing the fetched/created model instance and a boolean indicating
        if the object was created (`True`) or fetched (`False`).
    """
    if not isinstance(defaults, dict):
        # can be a ModelRef so pass it
        args = (defaults, *args)
        defaults = {}

    try:
        raw_instance, get_instance = await self._get_raw(**kwargs)
    except ObjectNotFound:
        kwargs.update(defaults)
        instance: EdgyEmbedTarget = await self.create(*args, **kwargs)
        return instance, True
    for arg in args:
        if isinstance(arg, ModelRef):
            relation_field = self.model_class.meta.fields[arg.__related_name__]
            extra_params = {}
            try:
                # m2m or foreign key
                target_model_class = relation_field.target
            except AttributeError:
                # reverse m2m or foreign key
                target_model_class = relation_field.related_from
            if not relation_field.is_m2m:
                # sometimes the foreign key is required, so set it already
                extra_params[relation_field.foreign_key.name] = raw_instance
            model = target_model_class(
                **arg.model_dump(exclude={"__related_name__"}),
                **extra_params,
            )
            relation = getattr(raw_instance, arg.__related_name__)
            await relation.add(model)
    return cast(EdgyEmbedTarget, get_instance), False

update_or_create async

update_or_create(defaults=None, *args, **kwargs)

Updates a single object matching kwargs using defaults. If not found, creates a new object.

PARAMETER	DESCRIPTION
defaults	Optional dictionary of values to apply as updates if the object is found, or to use as creation values if the object is new. Can also be a ModelRef. TYPE: dict[str, Any] | Any | None DEFAULT: None
*args	Positional arguments for model creation if object is not found. TYPE: Any DEFAULT: ()
**kwargs	Filters used to attempt retrieval. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
EdgyEmbedTarget	A tuple containing the fetched/created model instance and a boolean indicating
bool	if the object was created (True) or updated (False).

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

async def update_or_create(
    self, defaults: dict[str, Any] | Any | None = None, *args: Any, **kwargs: Any
) -> tuple[EdgyEmbedTarget, bool]:
    """
    Updates a single object matching `kwargs` using `defaults`. If not found, creates a new object.

    Args:
        defaults: Optional dictionary of values to apply as updates if the object is found,
                  or to use as creation values if the object is new. Can also be a `ModelRef`.
        *args: Positional arguments for model creation if object is not found.
        **kwargs: Filters used to attempt retrieval.

    Returns:
        A tuple containing the fetched/created model instance and a boolean indicating
        if the object was created (`True`) or updated (`False`).
    """
    if not isinstance(defaults, dict):
        # can be a ModelRef so pass it
        args = (defaults, *args)
        defaults = {}
    try:
        raw_instance, get_instance = await self._get_raw(**kwargs)
    except ObjectNotFound:
        kwargs.update(defaults)
        instance: EdgyEmbedTarget = await self.create(*args, **kwargs)
        return instance, True
    await get_instance.update(**defaults)
    for arg in args:
        if isinstance(arg, ModelRef):
            relation_field = self.model_class.meta.fields[arg.__related_name__]
            extra_params = {}
            try:
                # m2m or foreign key
                target_model_class = relation_field.target
            except AttributeError:
                # reverse m2m or foreign key
                target_model_class = relation_field.related_from
            if not relation_field.is_m2m:
                # sometimes the foreign key is required, so set it already
                extra_params[relation_field.foreign_key.name] = raw_instance
            model = target_model_class(
                **arg.model_dump(exclude={"__related_name__"}),
                **extra_params,
            )
            relation = getattr(raw_instance, arg.__related_name__)
            await relation.add(model)
    self._clear_cache()
    return cast(EdgyEmbedTarget, get_instance), False

contains async

contains(instance)

Checks if the QuerySet contains a specific model instance by verifying its existence in the database using its primary key(s).

PARAMETER	DESCRIPTION
instance	The model instance to check for containment. TYPE: BaseModelType

RETURNS	DESCRIPTION
bool	True if the record exists in the database and matches the QuerySet filters, False otherwise.

RAISES	DESCRIPTION
ValueError	If the provided object is not a model instance or has a missing primary key.

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

async def contains(self, instance: BaseModelType) -> bool:
    """
    Checks if the QuerySet contains a specific model instance by verifying its existence
    in the database using its primary key(s).

    Args:
        instance: The model instance to check for containment.

    Returns:
        True if the record exists in the database and matches the QuerySet filters, False otherwise.

    Raises:
        ValueError: If the provided object is not a model instance or has a missing primary key.
    """
    query: Any = {}
    try:
        pkcolumns = instance.pkcolumns
        for pkcolumn in pkcolumns:
            query[pkcolumn] = getattr(instance, pkcolumn)
            if query[pkcolumn] is None:
                raise ValueError("'obj' has incomplete/missing primary key.") from None
    except AttributeError:
        raise ValueError("'obj' must be a model or reflect model instance.") from None
    return await self.exists(**query)

transaction

transaction(*, force_rollback=False, **kwargs)

Returns a database transaction context manager for the assigned database.

PARAMETER	DESCRIPTION
force_rollback	If True, the transaction will always rollback when the context manager exits, regardless of whether an exception occurred. Useful for testing. TYPE: bool DEFAULT: False
**kwargs	Additional keyword arguments passed to the underlying database transaction factory. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
Transaction	A Transaction context manager.

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

def transaction(self, *, force_rollback: bool = False, **kwargs: Any) -> Transaction:
    """
    Returns a database transaction context manager for the assigned database.

    Args:
        force_rollback: If `True`, the transaction will always rollback when the context manager exits,
                        regardless of whether an exception occurred. Useful for testing.
        **kwargs: Additional keyword arguments passed to the underlying database transaction factory.

    Returns:
        A `Transaction` context manager.
    """
    return self.database.transaction(force_rollback=force_rollback, **kwargs)

__aiter__ async

__aiter__()

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

async def __aiter__(self) -> AsyncIterator[Any]:
    async for value in self._execute_iterate():
        yield value