Backend

Bases: _Container

Dependency Injection container for Event Sourcery components.

Dict-like object resolving object instance during access by type. Providers are passed setting the type as key and provider as a value.

By default, basic implementations are registered or an exception is raised if backend configuration (e.g., storage strategy) is required.

Core components are exposed by properties.

Source code in event_sourcery/_event_store/backend.py

class Backend(_Container):
    """
    Dependency Injection container for Event Sourcery components.

    Dict-like object resolving object instance during access by type.
    Providers are passed setting the type as key and provider as a value.

    By default, basic implementations are registered or an exception is raised
    if backend configuration (e.g., storage strategy) is required.

    Core components are exposed by properties.
    """

    def __init__(self) -> None:
        super().__init__()
        self[TenantId] = DEFAULT_TENANT
        self[EventRegistry] = EventRegistry()
        self[EncryptionStrategy] = NoEncryptionStrategy()
        self[EncryptionKeyStorageStrategy] = (
            lambda c: NoKeyStorageStrategy().scoped_for_tenant(c[TenantId])
        )
        self[Encryption] = lambda c: Encryption(
            registry=c[EventRegistry],
            strategy=c[EncryptionStrategy],
            key_storage=c[EncryptionKeyStorageStrategy],
        )
        self[Serde] = lambda c: Serde(
            registry=c[EventRegistry],
            encryption=c[Encryption],
        )
        self[StorageStrategy] = not_configured(
            "Use one of pyES backends: SQLAlchemy, Django or KurrentDB",
        )
        self[EventStore] = lambda c: EventStore(
            storage_strategy=c[StorageStrategy],
            serde=c[Serde],
        )
        self[Outbox] = lambda c: Outbox(
            strategy=c[OutboxStorageStrategy],
            serde=c[Serde],
        )
        self[OutboxStorageStrategy] = lambda _: NoOutboxStorageStrategy()
        self[SubscriptionStrategy] = not_configured(
            "Use one of pyES backends: SQLAlchemy, Django or KurrentDB",
        )
        self[PositionPhase] = lambda c: SubscriptionBuilder(
            c[Serde],
            c[SubscriptionStrategy],
        )

    @property
    def event_store(self) -> EventStore:
        """
        Returns the current instance of `EventStore`.
        """
        return self[EventStore]

    @property
    def outbox(self) -> Outbox:
        """
        Returns the current instance of `Outbox`.
        """
        return self[Outbox]

    @property
    def subscriber(self) -> PositionPhase:
        """
        Returns the current instance of `SubscriptionBuilder` (as `PositionPhase`).
        """
        return self[PositionPhase]

    def in_tenant_mode(self, tenant_id: TenantId) -> Self:
        """
        Returns a copy of the backend with the specified tenant ID set.
        """
        in_tenant_mode = self.copy()
        in_tenant_mode[TenantId] = tenant_id
        return in_tenant_mode

    def with_outbox(self, filterer: OutboxFiltererStrategy = no_filter) -> Self:
        """
        Configure the outbox with a custom filter.
        """
        raise NotImplementedError()

    def with_encryption(
        self,
        strategy: EncryptionStrategy,
        key_storage: EncryptionKeyStorageStrategy,
    ) -> Self:
        """
        Configures event encryption with the provided strategy and key storage.
        """
        self[EncryptionStrategy] = strategy
        self[EncryptionKeyStorageStrategy] = lambda c: key_storage.scoped_for_tenant(
            c[TenantId],
        )
        return self

event_store: EventStore property

Returns the current instance of EventStore.

outbox: Outbox property

Returns the current instance of Outbox.

subscriber: PositionPhase property

Returns the current instance of SubscriptionBuilder (as PositionPhase).

in_tenant_mode(tenant_id)

Returns a copy of the backend with the specified tenant ID set.

Source code in event_sourcery/_event_store/backend.py

def in_tenant_mode(self, tenant_id: TenantId) -> Self:
    """
    Returns a copy of the backend with the specified tenant ID set.
    """
    in_tenant_mode = self.copy()
    in_tenant_mode[TenantId] = tenant_id
    return in_tenant_mode

with_encryption(strategy, key_storage)

Configures event encryption with the provided strategy and key storage.

Source code in event_sourcery/_event_store/backend.py

def with_encryption(
    self,
    strategy: EncryptionStrategy,
    key_storage: EncryptionKeyStorageStrategy,
) -> Self:
    """
    Configures event encryption with the provided strategy and key storage.
    """
    self[EncryptionStrategy] = strategy
    self[EncryptionKeyStorageStrategy] = lambda c: key_storage.scoped_for_tenant(
        c[TenantId],
    )
    return self

with_outbox(filterer=no_filter)

Configure the outbox with a custom filter.

Source code in event_sourcery/_event_store/backend.py

def with_outbox(self, filterer: OutboxFiltererStrategy = no_filter) -> Self:
    """
    Configure the outbox with a custom filter.
    """
    raise NotImplementedError()