Quick start

This section shows how to write a very simple event sourced application using classes from the library. It shows the general story, which is elaborated over the following pages.

Please use pip to install the library with the ‘sqlalchemy’ option.

pip install eventsourcing[sqlalchemy]

Domain

Firstly, import the example entity class Example and its factory function create_new_example().

from eventsourcing.example.domainmodel import create_new_example, Example

These classes will be used as the domain model in this example.

Infrastructure

Next, setup an SQLite database in memory, using library classes SQLAlchemyDatastore, with SQLAlchemySettings and IntegerSequencedItemRecord.

from eventsourcing.infrastructure.sqlalchemy.datastore import SQLAlchemySettings, SQLAlchemyDatastore
from eventsourcing.infrastructure.sqlalchemy.activerecords import IntegerSequencedItemRecord

datastore = SQLAlchemyDatastore(
    settings=SQLAlchemySettings(uri='sqlite:///:memory:'),
    tables=(IntegerSequencedItemRecord,),
)

datastore.setup_connection()
datastore.setup_tables()

Application

Finally, define an application object factory, that constructs an application object from library class ApplicationWithPersistencePolicies. The application class happens to take an active record strategy object and a session object.

The active record strategy is an instance of class SQLAlchemyActiveRecordStrategy. The session object is an argument of the application factory, and will be a normal SQLAlchemy session object.

from eventsourcing.application.base import ApplicationWithPersistencePolicies
from eventsourcing.infrastructure.sqlalchemy.activerecords import SQLAlchemyActiveRecordStrategy
from eventsourcing.infrastructure.sequenceditem import SequencedItem
from eventsourcing.infrastructure.eventsourcedrepository import EventSourcedRepository

def construct_application(session):
    app = ApplicationWithPersistencePolicies(
        entity_active_record_strategy=SQLAlchemyActiveRecordStrategy(
            active_record_class=IntegerSequencedItemRecord,
            session=session
        )
    )
    app.example_repository = EventSourcedRepository(
        event_store=app.entity_event_store,
        mutator=Example._mutate,
    )
    return app

An example repository constructed from class EventSourcedRepository, and is assigned to the application object attribute example_repository. It is possible to subclass the library application class, and extend it by constructing entity repositories in the __init__(), we just didn’t do that here.

Run the code

Now, use the application to create, read, update, and delete “example” entities.

with construct_application(datastore.session) as app:

    # Create.
    example = create_new_example(foo='bar')

    # Read.
    assert example.id in app.example_repository
    assert app.example_repository[example.id].foo == 'bar'

    # Update.
    example.foo = 'baz'
    assert app.example_repository[example.id].foo == 'baz'

    # Delete.
    example.discard()
    assert example.id not in app.example_repository