Source code for eventsourcing.application.sqlalchemy

from typing import Any, Optional

from eventsourcing.application.simple import ApplicationWithConcreteInfrastructure
from eventsourcing.infrastructure.factory import InfrastructureFactory
from eventsourcing.infrastructure.sqlalchemy.factory import (
    SQLAlchemyInfrastructureFactory,
)
from eventsourcing.infrastructure.sqlalchemy.records import (
    EntitySnapshotRecord,
    StoredEventRecord,
)


[docs]class SQLAlchemyApplication(ApplicationWithConcreteInfrastructure): infrastructure_factory_class = SQLAlchemyInfrastructureFactory stored_event_record_class = StoredEventRecord snapshot_record_class = EntitySnapshotRecord is_constructed_with_session = True tracking_record_class: Any = None
[docs] def __init__( self, uri: Optional[str] = None, session: Optional[Any] = None, tracking_record_class: Any = None, **kwargs: Any ): self.uri = uri self._session = session self.tracking_record_class = ( tracking_record_class or type(self).tracking_record_class ) super(SQLAlchemyApplication, self).__init__(**kwargs)
@property def session(self) -> Optional[Any]: return self._session
[docs] def construct_infrastructure(self, *args: Any, **kwargs: Any) -> None: super(SQLAlchemyApplication, self).construct_infrastructure( session=self.session, uri=self.uri, tracking_record_class=self.tracking_record_class, *args, **kwargs )
[docs] def construct_infrastructure_factory( self, *args: Any, **kwargs: Any ) -> InfrastructureFactory: return super(SQLAlchemyApplication, self).construct_infrastructure_factory( *args, **kwargs )
[docs] def construct_datastore(self) -> None: super(SQLAlchemyApplication, self).construct_datastore() assert self._datastore assert self._datastore.session if self._session is None: self._session = self._datastore.session