Skip to content

Repository

Generic repository interfaces for Domain-Driven Design.

This module defines repository interfaces used to persist and retrieve
aggregate roots. Repositories abstract infrastructure concerns and offer
type-safe contracts for both command-side and query-side data access in
CQRS or traditional architectures.

Responsibilities:
- Persist and retrieve aggregates.
- Abstract away storage mechanisms.

Non-Responsibilities:
- Implement business invariants (aggregate enforces them).
- Expose ORM models or database APIs directly.

repository

Generic repository interfaces for Domain-Driven Design.

This module defines repository interfaces used to persist and retrieve
aggregate roots. Repositories abstract infrastructure concerns and offer
type-safe contracts for both command-side and query-side data access in
CQRS or traditional architectures.

Responsibilities
  • Persist and retrieve aggregates.
  • Abstract away storage mechanisms.
Non-Responsibilities
  • Implement business invariants (aggregate enforces them).
  • Expose ORM models or database APIs directly.

ReadOnlyRepository

Bases: Generic[TReadAggregateRoot, TId], Protocol

Read-only repository abstraction for query operations.

This interface is optimized for query-side usage in CQRS architectures.
It provides type-safe retrieval of aggregates or read models.

Source code in src/forging_blocks/application/ports/outbound/repository.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class ReadOnlyRepository(Generic[TReadAggregateRoot, TId], Protocol):
    """Read-only repository abstraction for query operations.

    This interface is optimized for query-side usage in CQRS architectures.
    It provides type-safe retrieval of aggregates or read models.
    """

    async def get_by_id(self, id: TId) -> TReadAggregateRoot | None:
        """Retrieve an aggregate or read model by ID.

        Args:
            id: Unique identifier of the resource.

        Returns:
            The retrieved instance or None if not found.

        Notes:
            Implementations may read from:
                - read replicas,
                - projections,
                - cached read models.
        """
        ...

    async def list_all(self) -> Sequence[TReadAggregateRoot]:
        """Retrieve all resources.

        Returns:
            A sequence of aggregate or read model instances.

        Notes:
            Data sources may differ from command-side storage.
        """
        ...

get_by_id(id: TId) -> TReadAggregateRoot | None async

Retrieve an aggregate or read model by ID.

Parameters:

Name Type Description Default
id TId

Unique identifier of the resource.

 required

Returns:

Type Description
TReadAggregateRoot | None

The retrieved instance or None if not found.
Notes

Implementations may read from:
- read replicas,
- projections,
- cached read models.

Source code in src/forging_blocks/application/ports/outbound/repository.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
async def get_by_id(self, id: TId) -> TReadAggregateRoot | None:
    """Retrieve an aggregate or read model by ID.

    Args:
        id: Unique identifier of the resource.

    Returns:
        The retrieved instance or None if not found.

    Notes:
        Implementations may read from:
            - read replicas,
            - projections,
            - cached read models.
    """
    ...

list_all() -> Sequence[TReadAggregateRoot] async

Retrieve all resources.

Returns:

Type Description
Sequence[TReadAggregateRoot]

A sequence of aggregate or read model instances.
Notes

Data sources may differ from command-side storage.

Source code in src/forging_blocks/application/ports/outbound/repository.py 
56
57
58
59
60
61
62
63
64
65
async def list_all(self) -> Sequence[TReadAggregateRoot]:
    """Retrieve all resources.

    Returns:
        A sequence of aggregate or read model instances.

    Notes:
        Data sources may differ from command-side storage.
    """
    ...

WriteOnlyRepository

Bases: Protocol, Generic[TWriteAggregateRoot, TWriteId]

Write-only repository abstraction for command operations.

This interface supports command-side operations where writes are applied
independently from read-side storage.

Source code in src/forging_blocks/application/ports/outbound/repository.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class WriteOnlyRepository(Protocol, Generic[TWriteAggregateRoot, TWriteId]):
    """Write-only repository abstraction for command operations.

    This interface supports command-side operations where writes are applied
    independently from read-side storage.
    """

    async def delete_by_id(self, id: TWriteId) -> None:
        """Delete an aggregate by ID.

        Args:
            id: Unique identifier of the aggregate.

        Raises:
            RepositoryError: If deletion fails.
        """
        ...

    async def save(self, aggregate: TWriteAggregateRoot) -> None:
        """Persist an aggregate instance.

        Args:
            aggregate: The aggregate to save.
        """
        ...

delete_by_id(id: TWriteId) -> None async

Delete an aggregate by ID.

Parameters:

Name Type Description Default
id TWriteId

Unique identifier of the aggregate.

 required

Raises:

Type Description
RepositoryError

If deletion fails.
Source code in src/forging_blocks/application/ports/outbound/repository.py 
75
76
77
78
79
80
81
82
83
84
async def delete_by_id(self, id: TWriteId) -> None:
    """Delete an aggregate by ID.

    Args:
        id: Unique identifier of the aggregate.

    Raises:
        RepositoryError: If deletion fails.
    """
    ...

save(aggregate: TWriteAggregateRoot) -> None async

Persist an aggregate instance.

Parameters:

Name Type Description Default
aggregate TWriteAggregateRoot

The aggregate to save.

 required
Source code in src/forging_blocks/application/ports/outbound/repository.py 
86
87
88
89
90
91
92
async def save(self, aggregate: TWriteAggregateRoot) -> None:
    """Persist an aggregate instance.

    Args:
        aggregate: The aggregate to save.
    """
    ...

Repository

Bases: ReadOnlyRepository[TAggregateRoot, TId], WriteOnlyRepository[TAggregateRoot, TId], Protocol

Full CRUD repository abstraction.

Combines read and write capabilities into a single repository interface.
Suitable for non-CQRS applications or simplified contexts.

Source code in src/forging_blocks/application/ports/outbound/repository.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class Repository(
    ReadOnlyRepository[TAggregateRoot, TId],
    WriteOnlyRepository[TAggregateRoot, TId],
    Protocol,
):
    """Full CRUD repository abstraction.

    Combines read and write capabilities into a single repository interface.
    Suitable for non-CQRS applications or simplified contexts.
    """

    ...