Skip to content

In Memory Write Repository

in_memory_write_repository

In-memory write-only repository implementation.

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

InMemoryWriteRepository

Bases: BaseWriteRepository[TWriteAggregateRoot, TWriteId]

In-memory implementation of WriteOnlyRepository for command operations.

Stores aggregates in a dictionary keyed by their identifier. Designed
for command-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_write_repository.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
class InMemoryWriteRepository[TWriteAggregateRoot: Identified[Any], TWriteId](
    BaseWriteRepository[TWriteAggregateRoot, TWriteId],
):
    """In-memory implementation of WriteOnlyRepository for command operations.

    Stores aggregates in a dictionary keyed by their identifier. Designed
    for command-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: MutableMapping[TWriteId, TWriteAggregateRoot] | 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.
        """
        super().__init__(storage)

__init__(storage: MutableMapping[TWriteId, TWriteAggregateRoot] | None = None) -> None

Initialize the write repository with optional external storage.

Parameters:

Name Type Description Default
storage MutableMapping[TWriteId, TWriteAggregateRoot] | 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/in_memory_write_repository.py 
28
29
30
31
32
33
34
35
36
37
38
def __init__(
    self,
    storage: MutableMapping[TWriteId, TWriteAggregateRoot] | 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.
    """
    super().__init__(storage)