InMemoryBackend

Bases: TransactionalBackend

In-memory backend for Event Sourcery.

Provides a fully configured backend for in-memory event store.

Useful for testing, development, and scenarios where persistence is not required. Ensures multi-tenancy and transactional event handling using in-memory implementations.

Source code in event_sourcery/_event_store/in_memory.py

class InMemoryBackend(TransactionalBackend):
    """
    In-memory backend for Event Sourcery.

    Provides a fully configured backend for in-memory event store.

    Useful for testing, development, and scenarios where persistence is not required.
    Ensures multi-tenancy and transactional event handling using in-memory
    implementations.
    """

    def __init__(self) -> None:
        super().__init__()
        self[InMemoryConfig] = not_configured(
            "Configure backend with `.configure(config)`"
        )
        self[Storage] = Storage()
        self[StorageStrategy] = lambda c: InMemoryStorageStrategy(
            c[Storage],
            c[Dispatcher],
            outbox_strategy=c.get(InMemoryOutboxStorageStrategy),
        ).scoped_for_tenant(c[TenantId])
        self[SubscriptionStrategy] = lambda c: InMemorySubscriptionStrategy(c[Storage])

    def configure(self, config: InMemoryConfig | None = None) -> Self:
        """
        Sets the backend configuration for outbox behavior.

        If no config is provided, the default configuration is used.
        This method must be called before using the backend.

        Args:
            config (InMemoryConfig | None):
                Optional custom configuration. If None, uses default KurrentDBConfig().

        Returns:
            Self: The configured backend instance (for chaining).
        """
        self[InMemoryConfig] = config or InMemoryConfig()
        return self

    def with_outbox(self, filterer: OutboxFiltererStrategy = no_filter) -> Self:
        self[OutboxFiltererStrategy] = filterer  # type: ignore[type-abstract]
        self[InMemoryOutboxStorageStrategy] = singleton(
            lambda c: InMemoryOutboxStorageStrategy(
                c[OutboxFiltererStrategy],  # type: ignore[type-abstract]
                c[InMemoryConfig].outbox_attempts,
            )
        )
        self[OutboxStorageStrategy] = lambda c: c[InMemoryOutboxStorageStrategy]
        return self

configure(config=None)

Sets the backend configuration for outbox behavior.

If no config is provided, the default configuration is used. This method must be called before using the backend.

Parameters:

Name	Type	Description	Default
config	InMemoryConfig | None	Optional custom configuration. If None, uses default KurrentDBConfig().	None

Returns:

Name	Type	Description
Self	Self	The configured backend instance (for chaining).

Source code in event_sourcery/_event_store/in_memory.py

def configure(self, config: InMemoryConfig | None = None) -> Self:
    """
    Sets the backend configuration for outbox behavior.

    If no config is provided, the default configuration is used.
    This method must be called before using the backend.

    Args:
        config (InMemoryConfig | None):
            Optional custom configuration. If None, uses default KurrentDBConfig().

    Returns:
        Self: The configured backend instance (for chaining).
    """
    self[InMemoryConfig] = config or InMemoryConfig()
    return self

InMemoryConfig

Bases: BaseModel

Configuration for InMemoryBackend event store integration.

Attributes:

Name	Type	Description
outbox_attempts	PositiveInt	Maximum number of outbox delivery attempts per event before giving up.

Source code in event_sourcery/_event_store/in_memory.py

class InMemoryConfig(BaseModel):
    """
    Configuration for InMemoryBackend event store integration.

    Attributes:
        outbox_attempts (PositiveInt):
            Maximum number of outbox delivery attempts per event before giving up.
    """

    model_config = ConfigDict(extra="forbid", frozen=True)
    outbox_attempts: PositiveInt = 3

InMemoryKeyStorage

Bases: EncryptionKeyStorageStrategy

In-memory implementation of encryption key storage strategy.

Stores encryption keys for data subjects in memory. Suitable for testing and development where persistent key storage is not required.

Source code in event_sourcery/_event_store/in_memory.py

@dataclass
class InMemoryKeyStorage(EncryptionKeyStorageStrategy):
    """
    In-memory implementation of encryption key storage strategy.

    Stores encryption keys for data subjects in memory. Suitable for testing and
    development where persistent key storage is not required.
    """

    _keys: dict[tuple[TenantId, str], bytes] = field(default_factory=dict)
    _tenant_id: TenantId = DEFAULT_TENANT

    def get(self, subject_id: str) -> bytes | None:
        return self._keys.get((self._tenant_id, subject_id))

    def store(self, subject_id: str, key: bytes) -> None:
        self._keys[(self._tenant_id, subject_id)] = key

    def delete(self, subject_id: str) -> None:
        self._keys.pop((self._tenant_id, subject_id), None)

    def scoped_for_tenant(self, tenant_id: TenantId) -> "InMemoryKeyStorage":
        return InMemoryKeyStorage(self._keys, tenant_id)