In Memory Write Repository

in_memory_write_repository

Write-only repository backed by an in-memory dictionary.

Stores entities keyed by identifier, supporting insert, update, and
delete operations with optimistic concurrency via etag versioning.

InMemoryWriteRepository

Bases: WriteOnlyRepositoryPort[TEntity, TId]

In-memory write-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 MyEntity:
    def __init__(self, id: int, name: str) -> None:
        self.id = id
        self.name = name


repo = InMemoryWriteRepository[MyEntity, int]()
entity = MyEntity(id=1, name="alpha")
await repo.save(entity)
await repo.delete_by_id(1)

Source code in src/forging_blocks/infrastructure/repositories/in_memory_write_repository.py

class InMemoryWriteRepository[TEntity: Identified[Any], TId](WriteOnlyRepositoryPort[TEntity, TId]):
    """In-memory write-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 MyEntity:
            def __init__(self, id: int, name: str) -> None:
                self.id = id
                self.name = name


        repo = InMemoryWriteRepository[MyEntity, int]()
        entity = MyEntity(id=1, name="alpha")
        await repo.save(entity)
        await repo.delete_by_id(1)
        ```
    """

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

    async def delete_by_id(self, id: TId) -> None:
        """Delete an entity by ID.

        Args:
            id: Unique identifier of the entity.

        Raises:
            RepositoryError: If the ID is None, an empty string,
                or the boolean False.
            RepositoryNotFoundError: If no entity exists with the given ID.

        """
        self._validate_id(id)
        if id not in self._storage:
            raise RepositoryNotFoundError.for_id(id)
        del self._storage[id]

    async def save(self, aggregate: TEntity) -> None:
        """Persist an entity instance.

        Args:
            aggregate: The entity to save.

        Raises:
            RepositoryError: If the entity has no valid identifier
                (None, empty string, or boolean False).

        """
        entity_id: TId = cast(TId, aggregate.id)
        self._validate_id(entity_id)
        self._storage[entity_id] = aggregate

    @classmethod
    def _validate_id(cls, identifier: object) -> None:
        """Validate that an entity identifier is not None, empty, or False.

        Mirrors the validation performed by
        ``AggregateRoot._validate_identity`` so that the repository
        independently guards against invalid identifiers.

        Raises:
            RepositoryError: If *identifier* is ``None``, an empty string
                (``""``), or the boolean ``False``.

        """
        is_none = identifier is None
        is_empty_string = identifier == ""
        is_false = identifier is False

        if is_none or is_empty_string or is_false:
            raise RepositoryError(
                ErrorMessage(
                    "Invalid entity identifier (must not be None, empty string, or False)."
                )
            )

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

Initialize the write 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_write_repository.py

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

delete_by_id(id: TId) -> None async

Delete an entity by ID.

Parameters:

Name	Type	Description	Default
id	TId	Unique identifier of the entity.	required

Raises:

Type	Description
RepositoryError	If the ID is None, an empty string, or the boolean False.
RepositoryNotFoundError	If no entity exists with the given ID.

Source code in src/forging_blocks/infrastructure/repositories/in_memory_write_repository.py

async def delete_by_id(self, id: TId) -> None:
    """Delete an entity by ID.

    Args:
        id: Unique identifier of the entity.

    Raises:
        RepositoryError: If the ID is None, an empty string,
            or the boolean False.
        RepositoryNotFoundError: If no entity exists with the given ID.

    """
    self._validate_id(id)
    if id not in self._storage:
        raise RepositoryNotFoundError.for_id(id)
    del self._storage[id]

save(aggregate: TEntity) -> None async

Persist an entity instance.

Parameters:

Name	Type	Description	Default
aggregate	TEntity	The entity to save.	required

Raises:

Type	Description
RepositoryError	If the entity has no valid identifier (None, empty string, or boolean False).

Source code in src/forging_blocks/infrastructure/repositories/in_memory_write_repository.py

async def save(self, aggregate: TEntity) -> None:
    """Persist an entity instance.

    Args:
        aggregate: The entity to save.

    Raises:
        RepositoryError: If the entity has no valid identifier
            (None, empty string, or boolean False).

    """
    entity_id: TId = cast(TId, aggregate.id)
    self._validate_id(entity_id)
    self._storage[entity_id] = aggregate