Skip to content

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.

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: Generic[TReadAggregateRoot, TId], ReadOnlyRepository[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 
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
52
53
54
55
56
57
58
59
class InMemoryReadRepository(
    Generic[TReadAggregateRoot, TId],
    ReadOnlyRepository[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__ = ("_storage",)

    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.
        """
        self._storage: dict[TId, TReadAggregateRoot] = dict(storage) if storage is not None else {}

    async def get_by_id(self, id: TId) -> TReadAggregateRoot | None:  # noqa: A002
        """Retrieve an aggregate or read model 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[TReadAggregateRoot]:
        """Retrieve all resources in the repository.

        Returns:
            A sequence of aggregate or read model instances.
        """
        return list(self._storage.values())

__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 
33
34
35
36
37
38
39
40
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.
    """
    self._storage: dict[TId, TReadAggregateRoot] = dict(storage) if storage is not None else {}

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.
Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_repository.py 
42
43
44
45
46
47
48
49
50
51
async def get_by_id(self, id: TId) -> TReadAggregateRoot | None:  # noqa: A002
    """Retrieve an aggregate or read model 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[TReadAggregateRoot] async

Retrieve all resources in the repository.

Returns:

Type Description
Sequence[TReadAggregateRoot]

A sequence of aggregate or read model instances.
Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_repository.py 
53
54
55
56
57
58
59
async def list_all(self) -> Sequence[TReadAggregateRoot]:
    """Retrieve all resources in the repository.

    Returns:
        A sequence of aggregate or read model instances.
    """
    return list(self._storage.values())