BaseRepository

Base class for repositories, the persistence abstraction for aggregates. Repositories provide a collection-oriented interface (add, get, all) to load and persist aggregates, hiding database details behind a clean domain API.

See Repositories guide for practical usage and Repositories concept for design rationale.

Bases: Element, OptionsMixin

This is the baseclass for concrete Repository implementations.

The three methods in this baseclass to add, get or all entities are sufficient in most cases to handle application requirements. They have built-in support for handling child relationships and honor Unit of Work constructs. While they can be overridden, it is generally suggested to call the parent method first before writing custom code.

Repositories are strictly meant to be used in conjunction with Aggregate elements. It is always prudent to deal with persistence at the transaction boundary, which is at an Aggregate's level.

Design note: no delete/remove method. Repositories intentionally do not support hard deletion. Domain state changes (cancellation, deactivation, archival) should be modeled as explicit state transitions via commands and events, not as record erasure. Hard deletion is available at the infrastructure level (_dao.delete()) for projection rebuilds, test teardown, and compliance requirements (e.g. GDPR right to erasure).

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

def __init__(self, domain: "Domain", provider: BaseProvider) -> None:
    self._domain = domain
    self._provider = provider

query property

query: QuerySet

Return a QuerySet for fluent filtering on the aggregate's data store.

Use this inside custom repository methods instead of self._dao.query::

@domain.repository(part_of=Person)
class PersonRepository:
    def adults(self):
        return self.query.filter(age__gte=18).all().items

find_by

find_by(**kwargs: Any) -> Any

Find a single aggregate matching the given criteria.

Raises ObjectNotFoundError if no match is found. Raises TooManyObjectsError if multiple matches are found.

Example::

@domain.repository(part_of=Person)
class PersonRepository:
    def find_by_email(self, email: str) -> Person:
        return self.find_by(email=email)

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

def find_by(self, **kwargs: Any) -> Any:
    """Find a single aggregate matching the given criteria.

    Raises ``ObjectNotFoundError`` if no match is found.
    Raises ``TooManyObjectsError`` if multiple matches are found.

    Example::

        @domain.repository(part_of=Person)
        class PersonRepository:
            def find_by_email(self, email: str) -> Person:
                return self.find_by(email=email)
    """
    item = self._dao.find_by(**kwargs)
    self._prewarm_associations(item)
    return item

find

find(criteria: Q) -> ResultSet

Find all aggregates matching a Q criteria expression.

Returns a ResultSet containing the matching aggregates. Accepts composable Q objects, making it easy to build reusable, domain-named query functions::

from protean.utils.query import Q

def overdue_orders() -> Q:
    return Q(status="pending", due_date__lt=datetime.now())

results = repo.find(overdue_orders())
results = repo.find(overdue_orders() & Q(total__gte=5000))

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

def find(self, criteria: Q) -> "ResultSet":
    """Find all aggregates matching a Q criteria expression.

    Returns a `ResultSet` containing
    the matching aggregates. Accepts composable ``Q`` objects, making
    it easy to build reusable, domain-named query functions::

        from protean.utils.query import Q

        def overdue_orders() -> Q:
            return Q(status="pending", due_date__lt=datetime.now())

        results = repo.find(overdue_orders())
        results = repo.find(overdue_orders() & Q(total__gte=5000))
    """
    return self.query.filter(criteria).all()

exists

exists(criteria: Q) -> bool

Check if any aggregate matches the given Q criteria.

Returns True when at least one aggregate satisfies the criteria, False otherwise. Unlike find, this method does not load aggregate objects -- it only checks for existence::

if repo.exists(Q(email="john@example.com")):
    raise ValueError("Email already taken")

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

def exists(self, criteria: Q) -> bool:
    """Check if any aggregate matches the given Q criteria.

    Returns ``True`` when at least one aggregate satisfies the
    criteria, ``False`` otherwise.  Unlike ``find``, this method
    does not load aggregate objects -- it only checks for existence::

        if repo.exists(Q(email="john@example.com")):
            raise ValueError("Email already taken")
    """
    return self.query.filter(criteria).all().total > 0

add

add(item: Any) -> Any

This method helps persist or update aggregates or projections into the persistence store.

Returns the persisted item.

Protean adopts a collection-oriented design pattern to handle persistence. What this means is that the Repository interface does not hint in any way that there is an underlying persistence mechanism, avoiding any notion of saving or persisting data in the design layer. The task of syncing the data back into the persistence store is handled automatically.

To be specific, a Repository mimics a set collection. Whatever the implementation, the repository will not allow instances of the same object to be added twice. Also, when retrieving objects from a Repository and modifying them, you don't need to "re-save" them to the Repository.

If there is a Unit of Work in progress, then the changes are performed on the UoW's active session. They are committed whenever the entire UoW is committed. If there is no transaction in progress, changes are committed immediately to the persistence store. This mechanism is part of the DAO's design, and is automatically used wherever one tries to persist data.

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

def add(self, item: Any) -> Any:
    """This method helps persist or update aggregates or projections into the persistence store.

    Returns the persisted item.

    Protean adopts a collection-oriented design pattern to handle persistence. What this means is that
    the Repository interface does not hint in any way that there is an underlying persistence mechanism,
    avoiding any notion of saving or persisting data in the design layer. The task of syncing the data
    back into the persistence store is handled automatically.

    To be specific, a Repository mimics a `set` collection. Whatever the implementation, the repository
    will not allow instances of the same object to be added twice. Also, when retrieving objects from
    a Repository and modifying them, you don't need to "re-save" them to the Repository.

    If there is a Unit of Work in progress, then the changes are performed on the
    UoW's active session. They are committed whenever the entire UoW is committed. If there is no
    transaction in progress, changes are committed immediately to the persistence store. This mechanism
    is part of the DAO's design, and is automatically used wherever one tries to persist data.
    """
    # Increment access log repo save counter
    with contextlib.suppress(Exception):
        g._access_log_repo_saves = getattr(g, "_access_log_repo_saves", 0) + 1

    tracer = self._domain.tracer

    with tracer.start_as_current_span(
        "protean.repository.add",
        record_exception=False,
        set_status_on_exception=False,
    ) as span:
        span.set_attribute("protean.aggregate.type", item.__class__.__name__)
        span.set_attribute("protean.provider", self._provider.name)

        try:
            return self._do_add(item)
        except Exception as exc:
            set_span_error(span, exc)
            raise

get

get(identifier: Any) -> Any

This is a utility method to fetch data from the persistence store by its key identifier. All child objects, including enclosed entities, are returned as part of this call.

Returns the fetched object.

All other data filtering capabilities can be implemented by using the underlying DAO's BaseDAO.filter method.

Filter methods are typically implemented as domain-contextual queries, like find_adults(), find_residents_of_area(zipcode), etc. It is also possible to make use of more complicated, domain-friendly design patterns like the Specification pattern.

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

def get(self, identifier: Any) -> Any:
    """This is a utility method to fetch data from the persistence store by its key identifier. All child objects,
    including enclosed entities, are returned as part of this call.

    Returns the fetched object.

    All other data filtering capabilities can be implemented by using the underlying DAO's
    ``BaseDAO.filter`` method.

    Filter methods are typically implemented as domain-contextual queries, like `find_adults()`,
    `find_residents_of_area(zipcode)`, etc. It is also possible to make use of more complicated,
    domain-friendly design patterns like the `Specification` pattern.
    """
    # Increment access log repo load counter
    with contextlib.suppress(Exception):
        g._access_log_repo_loads = getattr(g, "_access_log_repo_loads", 0) + 1

    tracer = self._domain.tracer

    with tracer.start_as_current_span(
        "protean.repository.get",
        record_exception=False,
        set_status_on_exception=False,
    ) as span:
        span.set_attribute("protean.aggregate.type", self.meta_.part_of.__name__)
        span.set_attribute("protean.provider", self._provider.name)

        try:
            item = self._dao.get(identifier)
            self._prewarm_associations(item)
            return item
        except Exception as exc:
            set_span_error(span, exc)
            raise

get_or_none

get_or_none(identifier: Any) -> Any | None

Fetch data from the persistence store by its key identifier, same as get(), but return None instead of raising ObjectNotFoundError when nothing matches.

On a hit, the returned object has all child objects, including enclosed entities, pre-warmed exactly as get() does. On a miss, this method returns None and never raises ObjectNotFoundError. Any other error (for example TooManyObjectsError, or a provider/connection error) still propagates.

Use get() instead when a missing identifier should be treated as a hard failure.

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

def get_or_none(self, identifier: Any) -> Any | None:
    """Fetch data from the persistence store by its key identifier, same as `get()`,
    but return `None` instead of raising `ObjectNotFoundError` when nothing matches.

    On a hit, the returned object has all child objects, including enclosed entities,
    pre-warmed exactly as `get()` does. On a miss, this method returns `None` and never
    raises `ObjectNotFoundError`. Any other error (for example `TooManyObjectsError`, or
    a provider/connection error) still propagates.

    Use `get()` instead when a missing identifier should be treated as a hard failure.
    """
    # Increment access log repo load counter
    with contextlib.suppress(Exception):
        g._access_log_repo_loads = getattr(g, "_access_log_repo_loads", 0) + 1

    tracer = self._domain.tracer

    with tracer.start_as_current_span(
        "protean.repository.get_or_none",
        record_exception=False,
        set_status_on_exception=False,
    ) as span:
        span.set_attribute("protean.aggregate.type", self.meta_.part_of.__name__)
        span.set_attribute("protean.provider", self._provider.name)

        try:
            item = self._dao.get(identifier)
            self._prewarm_associations(item)
            return item
        except ObjectNotFoundError:
            return None
        except Exception as exc:
            set_span_error(span, exc)
            raise