Skip to content

In Memory Read Repository

in_memory_read_repository

In-memory read-only repository implementation.

Provides a concrete implementation of ReadOnlyRepository backed by a
dictionary for query-side operations in CQRS architectures.

InMemoryReadRepository

Bases: BaseReadRepository[TReadAggregateRoot, TId]

In-memory implementation of ReadOnlyRepository for query operations.

Stores aggregates in a dictionary keyed by their identifier. Designed
for query-side usage in CQRS architectures.

The storage mapping is injected via the constructor and copied on init
to ensure independence from external mutation.

Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_repository.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class InMemoryReadRepository[TReadAggregateRoot, TId](
    BaseReadRepository[TReadAggregateRoot, TId],
):
    """In-memory implementation of ReadOnlyRepository for query operations.

    Stores aggregates in a dictionary keyed by their identifier. Designed
    for query-side usage in CQRS architectures.

    The storage mapping is injected via the constructor and copied on init
    to ensure independence from external mutation.
    """

    __slots__ = ()

    def __init__(self, storage: Mapping[TId, TReadAggregateRoot] | 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.
        """
        super().__init__(storage)

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

Initialize the read repository with optional external storage.

Parameters:

Name Type Description Default
storage Mapping[TId, TReadAggregateRoot] | 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/in_memory_read_repository.py 
26
27
28
29
30
31
32
33
def __init__(self, storage: Mapping[TId, TReadAggregateRoot] | 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.
    """
    super().__init__(storage)