Skip to content

Specification Repository Port

specification_repository_port

Specification-aware repository port.

Extends ReadOnlyRepository with query methods that accept
Specification predicates for in-memory filtering.

SpecificationRepositoryPort

Bases: ReadOnlyRepositoryPort[TEntity, TId]

Read-only repository port with Specification-based query support.

In addition to standard read-only operations, this interface
allows querying by Specification predicates.

Responsibilities
  • Find all entities matching a Specification predicate.
  • Count entities satisfying a specification.
  • Check whether any entity matches a specification.
Non-Responsibilities
  • Compile or optimize Specification predicates.
  • Paginate or sort specification results.
  • Provide indexing strategies — that belongs to infrastructure.
Example 
repo = MySpecRepo[Account, str]()
active = await repo.find_matching(IsActive())
count = await repo.count_matching(HasBalanceAbove(100.0))
Source code in src/forging_blocks/application/ports/outbound/specification_repository_port.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class SpecificationRepositoryPort[TEntity, TId](ReadOnlyRepositoryPort[TEntity, TId]):
    """Read-only repository port with Specification-based query support.

    In addition to standard read-only operations, this interface
    allows querying by ``Specification`` predicates.

    Responsibilities:
        - Find all entities matching a ``Specification`` predicate.
        - Count entities satisfying a specification.
        - Check whether any entity matches a specification.

    Non-Responsibilities:
        - Compile or optimize ``Specification`` predicates.
        - Paginate or sort specification results.
        - Provide indexing strategies — that belongs to infrastructure.

    Example:
        ```python
        repo = MySpecRepo[Account, str]()
        active = await repo.find_matching(IsActive())
        count = await repo.count_matching(HasBalanceAbove(100.0))
        ```
    """

    @abstractmethod
    async def find_matching(self, spec: Specification[TEntity]) -> Sequence[TEntity]:
        """Return all entities that satisfy the specification."""
        ...

    @abstractmethod
    async def count_matching(self, spec: Specification[TEntity]) -> int:
        """Return the count of entities satisfying the specification."""
        ...

    @abstractmethod
    async def exists_matching(self, spec: Specification[TEntity]) -> bool:
        """Return ``True`` if at least one entity satisfies the specification."""
        ...

find_matching(spec: Specification[TEntity]) -> Sequence[TEntity] abstractmethod async

Return all entities that satisfy the specification.

Source code in src/forging_blocks/application/ports/outbound/specification_repository_port.py 
38
39
40
41
@abstractmethod
async def find_matching(self, spec: Specification[TEntity]) -> Sequence[TEntity]:
    """Return all entities that satisfy the specification."""
    ...

count_matching(spec: Specification[TEntity]) -> int abstractmethod async

Return the count of entities satisfying the specification.

Source code in src/forging_blocks/application/ports/outbound/specification_repository_port.py 
43
44
45
46
@abstractmethod
async def count_matching(self, spec: Specification[TEntity]) -> int:
    """Return the count of entities satisfying the specification."""
    ...

exists_matching(spec: Specification[TEntity]) -> bool abstractmethod async

Return True if at least one entity satisfies the specification.

Source code in src/forging_blocks/application/ports/outbound/specification_repository_port.py 
48
49
50
51
@abstractmethod
async def exists_matching(self, spec: Specification[TEntity]) -> bool:
    """Return ``True`` if at least one entity satisfies the specification."""
    ...