Skip to content

In Memory Read Repository

in_memory_read_repository

In-memory read-only repository backed by a dictionary.

Provides a concrete implementation of ReadOnlyRepositoryPort for
query-side operations in CQRS architectures. Storage is a plain
dictionary keyed by entity identifier.

InMemoryReadRepository

Bases: ReadOnlyRepositoryPort[TEntity, TId]

In-memory read-only repository backed by a dictionary.

Stores entities in a dictionary keyed by their identifier.

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

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

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


storage = {1: {"id": 1, "name": "alpha"}, 2: {"id": 2, "name": "beta"}}
repo = InMemoryReadRepository[dict, int](storage=storage)

entity = await repo.get_by_id(1)
active = ExpressionSpecification(lambda e: e["name"].startswith("a"))
results = await repo.find_matching(active)
Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_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
 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
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class InMemoryReadRepository[TEntity, TId](ReadOnlyRepositoryPort[TEntity, TId]):
    """In-memory read-only repository backed by a dictionary.

    Stores entities in a dictionary keyed by their identifier.

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

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

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


        storage = {1: {"id": 1, "name": "alpha"}, 2: {"id": 2, "name": "beta"}}
        repo = InMemoryReadRepository[dict, int](storage=storage)

        entity = await repo.get_by_id(1)
        active = ExpressionSpecification(lambda e: e["name"].startswith("a"))
        results = await repo.find_matching(active)
        ```
    """

    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.

        """
        super().__init__()
        self._storage: dict[TId, TEntity] = dict(storage) if storage is not None else {}

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

        Args:
            entity_id: Unique identifier of the entity.

        Returns:
            The entity if found, otherwise None.

        """
        return self._storage.get(entity_id)

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

        Returns:
            A sequence of all stored entities.

        """
        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: Specification predicate to filter entities.

        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: Specification predicate to filter entities.

        Returns:
            The number 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: Specification predicate to filter entities.

        Returns:
            True if at least one 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/in_memory_read_repository.py 
41
42
43
44
45
46
47
48
49
50
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.

    """
    super().__init__()
    self._storage: dict[TId, TEntity] = dict(storage) if storage is not None else {}

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

Retrieve an entity by ID.

Parameters:

Name Type Description Default
entity_id TId

Unique identifier of the entity.

 required

Returns:

Type Description
TEntity | None

The entity if found, otherwise None.
Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_repository.py 
52
53
54
55
56
57
58
59
60
61
62
async def get_by_id(self, entity_id: TId) -> TEntity | None:
    """Retrieve an entity by ID.

    Args:
        entity_id: Unique identifier of the entity.

    Returns:
        The entity if found, otherwise None.

    """
    return self._storage.get(entity_id)

list_all() -> Sequence[TEntity] async

Retrieve all resources in the repository.

Returns:

Type Description
Sequence[TEntity]

A sequence of all stored entities.
Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_repository.py 
64
65
66
67
68
69
70
71
async def list_all(self) -> Sequence[TEntity]:
    """Retrieve all resources in the repository.

    Returns:
        A sequence of all stored entities.

    """
    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]

Specification predicate to filter entities.

 required

Returns:

Type Description
Sequence[TEntity]

A list of matching entities.
Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_repository.py 
73
74
75
76
77
78
79
80
81
82
83
async def find_matching(self, spec: Specification[TEntity]) -> Sequence[TEntity]:
    """Return all stored entities that satisfy the given specification.

    Args:
        spec: Specification predicate to filter entities.

    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]

Specification predicate to filter entities.

 required

Returns:

Type Description
int

The number of matching entities.
Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_repository.py 
85
86
87
88
89
90
91
92
93
94
95
async def count_matching(self, spec: Specification[TEntity]) -> int:
    """Return the count of entities satisfying the specification.

    Args:
        spec: Specification predicate to filter entities.

    Returns:
        The number 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]

Specification predicate to filter entities.

 required

Returns:

Type Description
bool

True if at least one entity matches, False otherwise.
Source code in src/forging_blocks/infrastructure/repositories/in_memory_read_repository.py 
 97
 98
 99
100
101
102
103
104
105
106
107
async def exists_matching(self, spec: Specification[TEntity]) -> bool:
    """Return True if at least one entity satisfies the specification.

    Args:
        spec: Specification predicate to filter entities.

    Returns:
        True if at least one entity matches, False otherwise.

    """
    return any(spec.is_satisfied_by(v) for v in self._storage.values())