Skip to content

Base Repository

base_repository

Base repository classes for common functionality.

Provides shared infrastructure for repository implementations.

BaseReadRepository

Bases: ReadOnlyRepositoryPort[TEntity, TId]

Base class for read-only repositories with common functionality.

Provides in-memory storage and basic query operations.

Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class BaseReadRepository[TEntity, TId](ReadOnlyRepositoryPort[TEntity, TId]):
    """Base class for read-only repositories with common functionality.

    Provides in-memory storage and basic query operations.
    """

    __slots__ = ("_storage",)

    def __init__(self, storage: Mapping[TId, TEntity] | None = None) -> None:
        """Initialize the read repository with optional external storage.

        Args:
            storage: An optional mapping to use as backing storage.
                If None, a new empty dictionary is used.
        """
        self._storage: dict[TId, TEntity] = dict(storage) if storage is not None else {}

    async def get_by_id(self, id: TId) -> TEntity | None:  # noqa: A002
        """Retrieve an entity by ID.

        Args:
            id: Unique identifier of the resource.

        Returns:
            The retrieved instance or None if not found.
        """
        return self._storage.get(id)

    async def list_all(self) -> Sequence[TEntity]:
        """Retrieve all resources in the repository.

        Returns:
            A sequence of entity instances.
        """
        return list(self._storage.values())

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

        Args:
            spec: The specification predicate to filter by.

        Returns:
            A list of matching entities.
        """
        return [v for v in self._storage.values() if spec.is_satisfied_by(v)]

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

        Args:
            spec: The specification predicate to filter by.

        Returns:
            The count of matching entities.
        """
        return sum(1 for v in self._storage.values() if spec.is_satisfied_by(v))

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

        Args:
            spec: The specification predicate to filter by.

        Returns:
            True if any entity matches, False otherwise.
        """
        return any(spec.is_satisfied_by(v) for v in self._storage.values())

__init__(storage: Mapping[TId, TEntity] | None = None) -> None

Initialize the read repository with optional external storage.

Parameters:

Name Type Description Default
storage Mapping[TId, TEntity] | None

An optional mapping to use as backing storage.
If None, a new empty dictionary is used.

 None
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
30
31
32
33
34
35
36
37
def __init__(self, storage: Mapping[TId, TEntity] | None = None) -> None:
    """Initialize the read repository with optional external storage.

    Args:
        storage: An optional mapping to use as backing storage.
            If None, a new empty dictionary is used.
    """
    self._storage: dict[TId, TEntity] = dict(storage) if storage is not None else {}

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

Retrieve an entity by ID.

Parameters:

Name Type Description Default
id TId

Unique identifier of the resource.

 required

Returns:

Type Description
TEntity | None

The retrieved instance or None if not found.
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
39
40
41
42
43
44
45
46
47
48
async def get_by_id(self, id: TId) -> TEntity | None:  # noqa: A002
    """Retrieve an entity by ID.

    Args:
        id: Unique identifier of the resource.

    Returns:
        The retrieved instance or None if not found.
    """
    return self._storage.get(id)

list_all() -> Sequence[TEntity] async

Retrieve all resources in the repository.

Returns:

Type Description
Sequence[TEntity]

A sequence of entity instances.
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
50
51
52
53
54
55
56
async def list_all(self) -> Sequence[TEntity]:
    """Retrieve all resources in the repository.

    Returns:
        A sequence of entity instances.
    """
    return list(self._storage.values())

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

Return all stored entities that satisfy the given specification.

Parameters:

Name Type Description Default
spec Specification[TEntity]

The specification predicate to filter by.

 required

Returns:

Type Description
Sequence[TEntity]

A list of matching entities.
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
58
59
60
61
62
63
64
65
66
67
async def find_matching(self, spec: Specification[TEntity]) -> Sequence[TEntity]:
    """Return all stored entities that satisfy the given specification.

    Args:
        spec: The specification predicate to filter by.

    Returns:
        A list of matching entities.
    """
    return [v for v in self._storage.values() if spec.is_satisfied_by(v)]

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

Return the count of entities satisfying the specification.

Parameters:

Name Type Description Default
spec Specification[TEntity]

The specification predicate to filter by.

 required

Returns:

Type Description
int

The count of matching entities.
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
69
70
71
72
73
74
75
76
77
78
async def count_matching(self, spec: Specification[TEntity]) -> int:
    """Return the count of entities satisfying the specification.

    Args:
        spec: The specification predicate to filter by.

    Returns:
        The count of matching entities.
    """
    return sum(1 for v in self._storage.values() if spec.is_satisfied_by(v))

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

Return True if at least one entity satisfies the specification.

Parameters:

Name Type Description Default
spec Specification[TEntity]

The specification predicate to filter by.

 required

Returns:

Type Description
bool

True if any entity matches, False otherwise.
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
80
81
82
83
84
85
86
87
88
89
async def exists_matching(self, spec: Specification[TEntity]) -> bool:
    """Return True if at least one entity satisfies the specification.

    Args:
        spec: The specification predicate to filter by.

    Returns:
        True if any entity matches, False otherwise.
    """
    return any(spec.is_satisfied_by(v) for v in self._storage.values())

BaseWriteRepository

Bases: WriteOnlyRepositoryPort[TEntity, TId]

Base class for write-only repositories with common functionality.

Provides in-memory storage and basic write operations with ID validation.

Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class BaseWriteRepository[TEntity: Identified[Any], TId](WriteOnlyRepositoryPort[TEntity, TId]):
    """Base class for write-only repositories with common functionality.

    Provides in-memory storage and basic write operations with ID validation.
    """

    __slots__ = ()

    def __init__(
        self,
        storage: MutableMapping[TId, TEntity] | None = None,
    ) -> None:
        """Initialize the write repository with optional external storage.

        Args:
            storage: An optional mutable mapping to use as backing storage.
                If None, a new empty dictionary is used.
        """
        self._storage: dict[TId, TEntity] = dict(storage) if storage is not None else {}

    async def delete_by_id(self, id: TId) -> None:  # noqa: A002
        """Delete an entity by ID.

        Args:
            id: Unique identifier of the entity.

        Raises:
            RepositoryError: If the ID is None, an empty string,
                or the boolean False.
            RepositoryNotFoundError: If no entity exists with the given ID.
        """
        self._validate_id(id)
        if id not in self._storage:
            raise RepositoryNotFoundError.for_id(id)
        del self._storage[id]

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

        Args:
            aggregate: The entity to save.

        Raises:
            RepositoryError: If the entity has no valid identifier
                (None, empty string, or boolean False).
        """
        entity_id: TId = aggregate.id  # type: ignore[assignment]
        self._validate_id(entity_id)
        self._storage[entity_id] = aggregate

    @staticmethod
    def _validate_id(identifier: object) -> None:
        """Validate that an entity identifier is not None, empty, or False.

        Mirrors the validation performed by
        ``AggregateRoot._validate_identity`` so that the repository
        independently guards against invalid identifiers.

        Raises:
            RepositoryError: If *identifier* is ``None``, an empty string
                (``""``), or the boolean ``False``.
        """
        is_none = identifier is None
        is_empty_string = identifier == ""
        is_false = identifier is False

        if is_none or is_empty_string or is_false:
            raise RepositoryError(
                ErrorMessage(
                    "Invalid entity identifier (must not be None, empty string, or False)."
                )
            )

__init__(storage: MutableMapping[TId, TEntity] | None = None) -> None

Initialize the write repository with optional external storage.

Parameters:

Name Type Description Default
storage MutableMapping[TId, TEntity] | None

An optional mutable mapping to use as backing storage.
If None, a new empty dictionary is used.

 None
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
100
101
102
103
104
105
106
107
108
109
110
def __init__(
    self,
    storage: MutableMapping[TId, TEntity] | None = None,
) -> None:
    """Initialize the write repository with optional external storage.

    Args:
        storage: An optional mutable mapping to use as backing storage.
            If None, a new empty dictionary is used.
    """
    self._storage: dict[TId, TEntity] = dict(storage) if storage is not None else {}

delete_by_id(id: TId) -> None async

Delete an entity by ID.

Parameters:

Name Type Description Default
id TId

Unique identifier of the entity.

 required

Raises:

Type Description
RepositoryError

If the ID is None, an empty string,
or the boolean False.
RepositoryNotFoundError

If no entity exists with the given ID.
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
async def delete_by_id(self, id: TId) -> None:  # noqa: A002
    """Delete an entity by ID.

    Args:
        id: Unique identifier of the entity.

    Raises:
        RepositoryError: If the ID is None, an empty string,
            or the boolean False.
        RepositoryNotFoundError: If no entity exists with the given ID.
    """
    self._validate_id(id)
    if id not in self._storage:
        raise RepositoryNotFoundError.for_id(id)
    del self._storage[id]

save(aggregate: TEntity) -> None async

Persist an entity instance.

Parameters:

Name Type Description Default
aggregate TEntity

The entity to save.

 required

Raises:

Type Description
RepositoryError

If the entity has no valid identifier
(None, empty string, or boolean False).
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
async def save(self, aggregate: TEntity) -> None:
    """Persist an entity instance.

    Args:
        aggregate: The entity to save.

    Raises:
        RepositoryError: If the entity has no valid identifier
            (None, empty string, or boolean False).
    """
    entity_id: TId = aggregate.id  # type: ignore[assignment]
    self._validate_id(entity_id)
    self._storage[entity_id] = aggregate

BaseRepository

Bases: BaseReadRepository[TEntity, TId], BaseWriteRepository[TEntity, TId]

Full CRUD repository base class combining read and write operations.

Suitable for non-CQRS applications or simplified contexts.

Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
class BaseRepository[TEntity: Identified[Any], TId](
    BaseReadRepository[TEntity, TId],
    BaseWriteRepository[TEntity, TId],
):
    """Full CRUD repository base class combining read and write operations.

    Suitable for non-CQRS applications or simplified contexts.
    """

    __slots__ = ()

    def __init__(
        self,
        storage: MutableMapping[TId, TEntity] | None = None,
    ) -> None:
        """Initialize the repository with optional external storage.

        Args:
            storage: An optional mutable mapping to use as backing storage.
                If None, a new empty dictionary is used.
        """
        super().__init__(storage)

__init__(storage: MutableMapping[TId, TEntity] | None = None) -> None

Initialize the repository with optional external storage.

Parameters:

Name Type Description Default
storage MutableMapping[TId, TEntity] | None

An optional mutable mapping to use as backing storage.
If None, a new empty dictionary is used.

 None
Source code in src/forging_blocks/infrastructure/repositories/base_repository.py 
177
178
179
180
181
182
183
184
185
186
187
def __init__(
    self,
    storage: MutableMapping[TId, TEntity] | None = None,
) -> None:
    """Initialize the repository with optional external storage.

    Args:
        storage: An optional mutable mapping to use as backing storage.
            If None, a new empty dictionary is used.
    """
    super().__init__(storage)