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
limit	The size the recordset to be pulled from database
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

Return the total number of records

items property

items

Return result values

first property

first

Return the first result

last property

last

Return the last result

has_next property

has_next

Return True if there are more values present

has_prev property

has_prev

Return True if there are previous values present

filter

filter(*args, **kwargs)

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, **kwargs):
    """
    Return a new QuerySet instance with the args ANDed to the existing
    set.
    """
    return self._filter_or_exclude(False, *args, **kwargs)

exclude

exclude(*args, **kwargs)

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, **kwargs):
    """
    Return a new QuerySet instance with NOT (args) ANDed to the existing
    set.
    """
    return self._filter_or_exclude(True, *args, **kwargs)

limit

limit(limit)

Limit number of records

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

def limit(self, limit):
    """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)

Fetch results after offset value

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

def offset(self, offset):
    """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()

        # If we are dealing with an aggregate, we should also update the last event position
        #   to make use of optimistic concurrency control. This event version will be used
        #   to check for conflicts when the aggregate is updated.
        # FIXME: This concurrency control applies only when events are generated. We should
        #   ensure the same control is applied when aggregates are updated without events.
        if entity.element_type == DomainObjects.AGGREGATE:
            # Fetch and sync events version
            id_f = id_field(entity)
            assert id_f is not None
            identifier = getattr(entity, id_f.field_name)
            last_message = self._domain.event_store.store.read_last_message(
                f"{entity.meta_.stream_category}-{identifier}"
            )
            if last_message:
                assert last_message.metadata is not None
                assert last_message.metadata.event_store is not None
                entity._event_position = last_message.metadata.event_store.position

        entity_items.append(entity)

        # Track aggregate at the UoW level, to be able to perform actions on UoW commit,
        #   like persisting events raised by the aggregate.
        if current_uow and entity.element_type == DomainObjects.AGGREGATE:
            current_uow._add_to_identity_map(entity)

    results.items = entity_items

    # Cache results
    self._result_cache = results

    return results

update

update(*data, **kwargs)

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, **kwargs):
    """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:
        # FIXME Log Exception
        raise

    return updated_item_count

raw

raw(query: Any, data: Any = None)

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.

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

def raw(self, query: Any, data: Any = None):
    """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.
    """
    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()
            entity_items.append(entity)
        results.items = entity_items

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

    return results

delete

delete()

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):
    """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:
        # FIXME Log Exception
        raise

    return deleted_item_count

update_all

update_all(*args, **kwargs)

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

This method forwards filters and updates directly to the repository. It does not instantiate entities and it does not trigger Entity callbacks or validations.

Update values can be specified either as a dict, or keyword arguments.

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_all(self, *args, **kwargs):
    """Updates all objects with details given if they match a set of conditions supplied.

    This method forwards filters and updates directly to the repository. It does not
    instantiate entities and it does not trigger Entity callbacks or validations.

    Update values can be specified either as a dict, or keyword arguments.

    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:
        updated_item_count = self._owner_dao._update_all(
            self._criteria, *args, **kwargs
        )
    except Exception:
        # FIXME Log Exception
        raise

    return updated_item_count

delete_all

delete_all(*args, **kwargs)

Deletes objects that match a set of conditions supplied.

This method forwards filters directly to the repository. It does not instantiate entities and it does not trigger Entity callbacks or validations.

Returns the number of objects matched and deleted.

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

def delete_all(self, *args, **kwargs):
    """Deletes objects that match a set of conditions supplied.

    This method forwards filters directly to the repository. It does not instantiate entities and
    it does not trigger Entity callbacks or validations.

    Returns the number of objects matched and deleted.
    """
    deleted_item_count = 0
    try:
        deleted_item_count = self._owner_dao._delete_all(self._criteria)
    except Exception:
        # FIXME Log Exception
        raise

    return deleted_item_count

__iter__

__iter__()

Return results on iteration

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

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

__len__

__len__()

Return length of results

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

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

__bool__

__bool__()

Return True if query results have items

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

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

__repr__

__repr__()

Support friendly print of query criteria

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

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

Support slicing of results

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

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

__contains__

__contains__(k)

Support in operations

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

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