Skip to content

In Memory Repository

in_memory_repository

In-memory full CRUD repository backed by a dictionary.

Combines InMemoryReadRepository and InMemoryWriteRepository into a single
class suitable for non-CQRS applications or simplified contexts.

InMemoryRepository

Bases: InMemoryReadRepository[TEntity, TId], InMemoryWriteRepository[TEntity, TId]

Full CRUD in-memory repository backed by a dictionary.

Combines read and write operations into a single class using shared
dictionary-based storage. Suitable for non-CQRS applications or
simplified single-process contexts.

Example 
class ExpressionSpecification:
    def __init__(self, predicate):
        self._predicate = predicate

    def is_satisfied_by(self, entity):
        return self._predicate(entity)


class MyEntity:
    def __init__(self, id: int, name: str) -> None:
        self.id = id
        self.name = name


repo = InMemoryRepository[MyEntity, int]()
entity = MyEntity(id=1, name="alpha")
await repo.save(entity)
retrieved = await repo.get_by_id(1)
matches = await repo.find_matching(ExpressionSpecification(lambda e: e.name == "alpha"))
await repo.delete_by_id(1)
Source code in src/forging_blocks/infrastructure/repositories/in_memory_repository.py 
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
60
61
62
63
64
65
66
class InMemoryRepository[TEntity: Identified[Any], TId](
    InMemoryReadRepository[TEntity, TId],
    InMemoryWriteRepository[TEntity, TId],
):
    """Full CRUD in-memory repository backed by a dictionary.

    Combines read and write operations into a single class using shared
    dictionary-based storage. Suitable for non-CQRS applications or
    simplified single-process contexts.

    Example:
        ```python
        class ExpressionSpecification:
            def __init__(self, predicate):
                self._predicate = predicate

            def is_satisfied_by(self, entity):
                return self._predicate(entity)


        class MyEntity:
            def __init__(self, id: int, name: str) -> None:
                self.id = id
                self.name = name


        repo = InMemoryRepository[MyEntity, int]()
        entity = MyEntity(id=1, name="alpha")
        await repo.save(entity)
        retrieved = await repo.get_by_id(1)
        matches = await repo.find_matching(ExpressionSpecification(lambda e: e.name == "alpha"))
        await repo.delete_by_id(1)
        ```
    """

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

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

Initialize the repository with optional external storage.

Parameters:

Name Type Description Default
storage Mapping[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/in_memory_repository.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
def __init__(
    self,
    storage: Mapping[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__()
    self._storage: dict[TId, TEntity] = dict(storage) if storage is not None else {}