QuerySet

Chainable query builder for repository data access. QuerySets support filtering, excluding, ordering, pagination, and lookup operations.

See Retrieve Aggregates guide for practical usage.

A chainable class to gather a bunch of criteria and preferences (resultset size, order etc.) before execution.

Internally, a QuerySet can be constructed, filtered, sliced, and generally passed around without actually fetching data. No data fetch actually occurs until you do something to evaluate the queryset.

Once evaluated, a QuerySet typically caches its results. If the data in the database might have changed, you can get updated results for the same query by calling all() on a previously evaluated QuerySet.

ATTRIBUTE	DESCRIPTION
offset	Number of records after which Results are fetched TYPE: QuerySet
limit	The size the recordset to be pulled from database TYPE: QuerySet
order_by	The list of parameters to be used for ordering the results. Use a - before the parameter name to sort in descending order and if not ascending order. TYPE: QuerySet
excludes_	Objects with these properties will be excluded from the results TYPE: QuerySet
filters	Filter criteria TYPE: QuerySet

:return Returns a ResultSet object that holds the query results

Initialize either with empty preferences (when invoked on an Entity) or carry forward filters and preferences when chained

Source code in src/protean/core/queryset.py

def __init__(
    self,
    owner_dao: "BaseDAO",
    domain: "Domain",
    entity_cls: type["BaseEntity"],
    criteria: Q | None = None,
    offset: int = 0,
    limit: int | None = None,  # No limit by default
    order_by: list[str] | str | None = None,
    only_fields: list[str] | None = None,
) -> None:
    """Initialize either with empty preferences (when invoked on an Entity)
    or carry forward filters and preferences when chained
    """
    self._owner_dao = owner_dao
    self._domain = domain
    self._entity_cls = entity_cls
    self._criteria = criteria or Q()
    self._result_cache: "ResultSet | None" = None
    self._offset = offset or 0

    # Field selection set via ``only()``. ``None`` means "fetch full
    # rows and materialize entities"; a list means "fetch only these
    # attributes and return read-only ``Record`` objects". Stored as
    # attribute (column) names so adapters can consume it directly.
    self._only_fields: list[str] | None = only_fields

    # If an explicit limit is not provided, use the limit from the entity class
    self._limit = limit or entity_cls.meta_.limit

    # `order_by` could be empty, or a string or a list.
    #   Initialize empty list if `order_by` is None
    #   Convert string to list if `order_by` is a String
    #   Safe-cast list to a list if `order_by` is already a list
    self._order_by: list[str]
    if order_by:
        self._order_by = [order_by] if isinstance(order_by, str) else order_by
    else:
        self._order_by = []

total property

total: int

Return the total number of records

items property

items: list[Any]

Return result values

first property

first: Any | None

Return the first result

last property

last: Any | None

Return the last result

has_next property

has_next: bool

Return True if there are more values present

has_prev property

has_prev: bool

Return True if there are previous values present

page property

page: int

Return the current page number

page_size property

page_size: int | None

Return the page size

total_pages property

total_pages: int

Return the total number of pages

filter

filter(*args: Any, **kwargs: Any) -> QuerySet

Return a new QuerySet instance with the args ANDed to the existing set.

Source code in src/protean/core/queryset.py

def filter(self, *args: Any, **kwargs: Any) -> "QuerySet":
    """
    Return a new QuerySet instance with the args ANDed to the existing
    set.
    """
    return self._filter_or_exclude(False, *args, **kwargs)

exclude

exclude(*args: Any, **kwargs: Any) -> QuerySet

Return a new QuerySet instance with NOT (args) ANDed to the existing set.

Source code in src/protean/core/queryset.py

def exclude(self, *args: Any, **kwargs: Any) -> "QuerySet":
    """
    Return a new QuerySet instance with NOT (args) ANDed to the existing
    set.
    """
    return self._filter_or_exclude(True, *args, **kwargs)

limit

limit(limit: int | None) -> QuerySet

Limit number of records

Source code in src/protean/core/queryset.py

def limit(self, limit: int | None) -> "QuerySet":
    """Limit number of records"""
    clone = self._clone()

    # Assign limit if it is an integer or None
    if isinstance(limit, int) or limit is None:
        clone._limit = limit

    return clone

offset

offset(offset: int) -> QuerySet

Fetch results after offset value

Source code in src/protean/core/queryset.py

def offset(self, offset: int) -> "QuerySet":
    """Fetch results after `offset` value"""
    clone = self._clone()

    if isinstance(offset, int):
        clone._offset = offset

    return clone

order_by

order_by(order_by: Union[list[str], str]) -> QuerySet

Update order_by setting for filter set

Source code in src/protean/core/queryset.py

def order_by(self, order_by: Union[list[str], str]) -> "QuerySet":
    """Update order_by setting for filter set"""
    clone = self._clone()

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

    # Get the attribute name of the field
    # We want to support both field name and attribute name in the query,
    #   so we look for the key name in both fields and attributes.
    #
    # If we don't find it in either, we raise an error.
    new_order_by = []

    for key in order_by:
        # If the key starts with a minus sign, it is a descending order
        reverse = False
        if key.startswith("-"):
            reverse = True
            cleaned_key = key[1:]
        else:
            cleaned_key = key

        attr_name = self._resolve_attribute_name(cleaned_key)

        if reverse:
            new_order_by.append(f"-{attr_name}")
        else:
            new_order_by.append(attr_name)

    clone._order_by.extend(
        item for item in new_order_by if item not in clone._order_by
    )

    return clone

only

only(*field_names: str) -> QuerySet

Restrict the query to a subset of persisted fields.

Returns a new QuerySet that, when evaluated, fetches only the requested columns (plus the identifier, which is always included) and yields read-only :class:Record objects instead of fully materialized domain entities. This avoids the I/O of loading large columns (e.g. JSON blobs) on read-optimized paths — counts, cleanups, statistics — that never need the whole record.

A Record is not a domain entity: it has no behavior, runs no invariants, and cannot be persisted. It is purely a read-side carrier of column values. Domain operations must continue to go through full entities; only() is for field-selection reads.

Calling only() again replaces the selection (selections do not compose). Calling only() with no arguments clears any selection and restores full-entity materialization.

:param field_names: Names of persisted fields to project. The identifier is always included automatically. :raises KeyError: if a name is not a field or attribute of the entity. :raises NotSupportedError: if a name resolves to a non-persisted field (e.g. an association), which cannot be projected.

Source code in src/protean/core/queryset.py

def only(self, *field_names: str) -> "QuerySet":
    """Restrict the query to a subset of persisted fields.

    Returns a new ``QuerySet`` that, when evaluated, fetches only the
    requested columns (plus the identifier, which is always included) and
    yields read-only :class:`Record` objects instead of fully materialized
    domain entities. This avoids the I/O of loading large columns (e.g.
    JSON blobs) on read-optimized paths — counts, cleanups, statistics —
    that never need the whole record.

    A ``Record`` is **not** a domain entity: it has no behavior, runs no
    invariants, and cannot be persisted. It is purely a read-side carrier
    of column values. Domain operations must continue to go through full
    entities; ``only()`` is for field-selection reads.

    Calling ``only()`` again **replaces** the selection (selections do not
    compose). Calling ``only()`` with no arguments clears any selection and
    restores full-entity materialization.

    :param field_names: Names of persisted fields to project. The
        identifier is always included automatically.
    :raises KeyError: if a name is not a field or attribute of the entity.
    :raises NotSupportedError: if a name resolves to a non-persisted field
        (e.g. an association), which cannot be projected.
    """
    clone = self._clone()

    # No arguments clears the selection (last call wins).
    if not field_names:
        clone._only_fields = None
        return clone

    entity_attributes = attributes(self._entity_cls)
    resolved: list[str] = []

    def _add(name: str) -> None:
        attr_name = self._resolve_attribute_name(name)

        # Only persisted attributes (real columns) can be projected.
        # Associations and other non-persisted fields have no column to
        # fetch and would make the selection meaningless.
        if attr_name not in entity_attributes:
            raise NotSupportedError(
                f"`.only()` cannot project '{name}' on "
                f"{self._entity_cls.__name__}: it is not a persisted field."
            )

        if attr_name not in resolved:
            resolved.append(attr_name)

    # The identifier is always included so every Record is addressable.
    id_field_obj = id_field(self._entity_cls)
    if id_field_obj is not None:
        # A registered identity field always has its ``field_name`` populated
        # (set during ``__set_name__``); ``None`` only occurs on an unbound Field.
        assert id_field_obj.field_name is not None
        _add(id_field_obj.field_name)

    for name in field_names:
        _add(name)

    clone._only_fields = resolved
    return clone

all

all(with_total: bool = True) -> ResultSet

Primary method to fetch data based on filters

Also trigged when the QuerySet is evaluated by calling one of the following methods

• len()
• bool()
• list()
• Iteration
• Slicing

When with_total is False the adapter may skip any expensive total-count computation (e.g. SQL's separate COUNT query); use this when only ResultSet.items is needed and ResultSet.total can be disregarded.

Source code in src/protean/core/queryset.py

def all(self, with_total: bool = True) -> "ResultSet":
    """Primary method to fetch data based on filters

    Also trigged when the QuerySet is evaluated by calling one of the following methods:
        * len()
        * bool()
        * list()
        * Iteration
        * Slicing

    When ``with_total`` is ``False`` the adapter may skip any expensive
    total-count computation (e.g. SQL's separate ``COUNT`` query); use this
    when only ``ResultSet.items`` is needed and ``ResultSet.total`` can be
    disregarded.
    """
    logger.debug(f"Query `{self.__class__.__name__}` objects with filters {self}")

    # Destroy any cached results
    self._result_cache = None

    # Call the read method of the dao
    results = self._owner_dao._filter(
        self._criteria,
        self._offset,
        self._limit,
        self._order_by,
        with_total=with_total,
        fields=self._only_fields,
    )

    if self._only_fields is not None:
        # Projection path: build inert, read-only Record objects. These are
        # not domain entities, so they are deliberately not retrieved-
        # marked, event-synced, or tracked in the Unit of Work.
        results.items = self._owner_dao.database_model_cls.to_records(
            results.items, self._only_fields
        )
        self._result_cache = results
        return results

    # Convert the returned results to entity and return it
    entity_items = []
    for item in results.items:
        entity = self._owner_dao.database_model_cls.to_entity(item)
        entity.state_.mark_retrieved()

        # Sync event position and register in UoW identity map
        self._owner_dao._sync_event_position(entity)
        self._owner_dao._track_in_uow(entity)

        entity_items.append(entity)

    results.items = entity_items

    # Cache results
    self._result_cache = results

    return results

count

count() -> int

Return the count of records matching the current criteria.

Issues a single SELECT COUNT(*) (or adapter equivalent) without projecting columns or materializing entities. Ignores offset, limit, and order_by since they do not affect the row count.

Source code in src/protean/core/queryset.py

def count(self) -> int:
    """Return the count of records matching the current criteria.

    Issues a single ``SELECT COUNT(*)`` (or adapter equivalent) without
    projecting columns or materializing entities. Ignores ``offset``,
    ``limit``, and ``order_by`` since they do not affect the row count.
    """
    return self._owner_dao._count(self._criteria)

update

update(*data: Any, **kwargs: Any) -> int

Updates all objects with details given if they match a set of conditions supplied.

This method updates each object individually, to fire callback methods and ensure validations are run.

Returns the number of objects matched (which may not be equal to the number of objects updated if objects rows already have the new value).

Source code in src/protean/core/queryset.py

def update(self, *data: Any, **kwargs: Any) -> int:
    """Updates all objects with details given if they match a set of conditions supplied.

    This method updates each object individually, to fire callback methods and ensure
    validations are run.

    Returns the number of objects matched (which may not be equal to the number of objects
        updated if objects rows already have the new value).
    """
    self._reject_if_projected("update")

    updated_item_count = 0

    try:
        items = self.all()

        for item in items:
            self._owner_dao.update(item, *data, **kwargs)
            updated_item_count += 1
    except Exception:
        raise

    return updated_item_count

raw

raw(query: Any, data: Any = None) -> ResultSet

Runs raw query directly on the database and returns Entity objects

Note that this method will raise an exception if the returned objects are not of the Entity type.

query is not checked for correctness or validity, and any errors thrown by the plugin or database are passed as-is. Data passed will be transferred as-is to the plugin.

All other query options like order_by, offset, limit, and any only() field selection are ignored for this action; raw() always returns full Entity objects, never Record objects.

Raises NotSupportedError if the provider does not support raw queries.

Source code in src/protean/core/queryset.py

def raw(self, query: Any, data: Any = None) -> "ResultSet":
    """Runs raw query directly on the database and returns Entity objects

    Note that this method will raise an exception if the returned objects
        are not of the Entity type.

    `query` is not checked for correctness or validity, and any errors thrown by the plugin or
        database are passed as-is. Data passed will be transferred as-is to the plugin.

    All other query options like `order_by`, `offset`, `limit`, and any
    `only()` field selection are ignored for this action; `raw()` always
    returns full Entity objects, never `Record` objects.

    Raises NotSupportedError if the provider does not support raw queries.
    """
    provider = self._owner_dao.provider
    if not provider.has_capability(DatabaseCapabilities.RAW_QUERIES):
        raise NotSupportedError(
            f"Provider '{provider.name}' ({provider.__class__.__name__}) "
            "does not support raw queries"
        )

    logger.debug(
        f"Query `{self.__class__.__name__}` objects with raw query {query}"
    )

    # Destroy any cached results
    self._result_cache = None

    try:
        # Call the raw method of the repository
        results = self._owner_dao._raw(query, data)

        # Convert the returned results to entity and return it
        entity_items = []
        for item in results.items:
            entity = self._owner_dao.database_model_cls.to_entity(item)
            entity.state_.mark_retrieved()

            # Sync event position and register in UoW identity map
            self._owner_dao._sync_event_position(entity)
            self._owner_dao._track_in_uow(entity)

            entity_items.append(entity)
        results.items = entity_items

        # Cache results
        self._result_cache = results
    except Exception:
        raise

    return results

delete

delete() -> int

Deletes matching objects from the Repository

Does not throw error if no objects are matched.

Returns the number of objects matched (which may not be equal to the number of objects deleted if objects rows already have the new value).

Source code in src/protean/core/queryset.py

def delete(self) -> int:
    """Deletes matching objects from the Repository

    Does not throw error if no objects are matched.

    Returns the number of objects matched (which may not be equal to the number of objects
        deleted if objects rows already have the new value).
    """
    self._reject_if_projected("delete")

    # Fetch Model class and connected repository from Domain
    deleted_item_count = 0

    try:
        items = self.all()

        for item in items:
            self._owner_dao.delete(item)
            deleted_item_count += 1
    except Exception:
        raise

    return deleted_item_count

__iter__

__iter__() -> Any

Return results on iteration

Source code in src/protean/core/queryset.py

def __iter__(self) -> Any:
    """Return results on iteration"""
    return iter(self._data)

__len__

__len__() -> int

Return length of results

Source code in src/protean/core/queryset.py

def __len__(self) -> int:
    """Return length of results"""
    return self._data.total

__bool__

__bool__() -> bool

Return True if query results have items

Source code in src/protean/core/queryset.py

def __bool__(self) -> bool:
    """Return True if query results have items"""
    return bool(self._data)

__repr__

__repr__() -> str

Support friendly print of query criteria

Source code in src/protean/core/queryset.py

def __repr__(self) -> str:
    """Support friendly print of query criteria"""
    return "<%s: entity: %s, criteria: %s, offset: %s, limit: %s, order_by: %s>" % (
        self.__class__.__name__,
        self._entity_cls,
        self._criteria.deconstruct(),
        self._offset,
        self._limit,
        self._order_by,
    )

__getitem__

__getitem__(k: Any) -> Any

Support slicing of results

Source code in src/protean/core/queryset.py

def __getitem__(self, k: Any) -> Any:
    """Support slicing of results"""
    return self._data.items[k]

__contains__

__contains__(k: Any) -> bool

Support in operations

Source code in src/protean/core/queryset.py

def __contains__(self, k: Any) -> bool:
    """Support `in` operations"""
    return k.id in [item.id for item in self._data.items]

---

Record

The read-only value type returned by QuerySet.only(). A Record carries a projected subset of a single result's fields. It is not a domain entity: it has no behavior, runs no invariants, and cannot be persisted.

A read-only selection of fields from a single result.

Returned by :meth:QuerySet.only instead of a fully materialized domain entity. A Record is intentionally inert: it is not a domain entity, it carries no behavior, runs no invariants, and cannot be persisted. It exists purely to carry a subset of column values on read-optimized paths (counts, cleanups, statistics) without the cost, or the validity guarantees, of a full entity. This keeps the domain model airtight: field selection never produces a partially-valid aggregate.

Access selected values by attribute (record.status) or item (record["status"]). Reading a field that was not selected raises :class:AttributeError / :class:KeyError rather than returning a silent None, so an unselected field is never mistaken for a null value.

Source code in src/protean/core/queryset.py

def __init__(self, entity_name: str, data: dict[str, Any]) -> None:
    object.__setattr__(self, "_entity_name", entity_name)
    object.__setattr__(self, "_data", dict(data))

keys

keys() -> KeysView[str]

Return the projected field names.

Source code in src/protean/core/queryset.py

def keys(self) -> KeysView[str]:
    """Return the projected field names."""
    return self._data.keys()

to_dict

to_dict() -> dict[str, Any]

Return the projected values as a plain dict.

Source code in src/protean/core/queryset.py

def to_dict(self) -> dict[str, Any]:
    """Return the projected values as a plain dict."""
    return dict(self._data)

---

F

A reference to another column of the same row, for use as the right-hand side of a lookup inside filter()/Q (e.g. filter(retry_count__lt=F("max_retries"))). Resolved natively by the in-memory and SQLAlchemy adapters; the Elasticsearch adapter raises NotImplementedError for F-bearing predicates. See the Retrieve Aggregates guide for usage.

Reference to another column of the same row, for use inside Q lookups.

F lets a filter compare two columns of the same row instead of comparing a column against a literal value::

# column against a literal
query.filter(retry_count__lt=3)

# column against another column
query.filter(retry_count__lt=F("max_retries"))

Only a bare column reference is supported. Arithmetic (F("a") + 1), function calls (Lower(F("email"))), and aggregations are intentionally out of scope.

Adapter support: the in-memory and SQLAlchemy adapters resolve F to the referenced column natively. The Elasticsearch adapter raises NotImplementedError for F-bearing predicates (column-to-column comparison there needs a Painless script query, which is not implemented); use the in-memory or SQLAlchemy backend for such filters.

Source code in src/protean/utils/query.py

def __init__(self, name: str) -> None:
    self.name = name