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.
excludes_	Objects with these properties will be excluded from the results
filters	Filter criteria

: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: "BaseEntity",
    criteria: Q = None,
    offset: int = 0,
    limit: int = None,  # No limit by default
    order_by: list = 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 = None
    self._offset = offset or 0

    # 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
    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

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])

Update order_by setting for filter set

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

def order_by(self, order_by: Union[list, str]):
    """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

        if cleaned_key in fields(self._entity_cls):
            attr_name = fields(self._entity_cls)[cleaned_key].attribute_name
        elif cleaned_key in attributes(self._entity_cls):
            attr_name = attributes(self._entity_cls)[cleaned_key].attribute_name
        else:
            raise KeyError(
                f"Key '{cleaned_key}' not found in either fields or attributes of {self._entity_cls}"
            )

        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

all

all() -> 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

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

def all(self) -> "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
    """
    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
    )

    # 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

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).
    """
    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 and limit are ignored for this action.

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` and `limit` are ignored for this action.

    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).
    """
    # 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]