4
\$\begingroup\$

I've been studying the event sourcing pattern and trying to come up with a set of classes or minimal library I can share across projects.

Below you can find the implementation of this main parts of this pattern including the EventStore and Repository. I'd be thankful on getting a review on whether this is what the pattern is about and advice on where I could improve.

Note that I excluded classes Immutable, Identifiable and Aggregate for brevity but I hope their nouns are self-describing enough to provide the necessary context.

# -*- coding: utf-8 -*-

from abc import ABCMeta, abstractmethod
from datetime import *
from . import Immutable, Identifiable, Aggregate, Repository, snake_case, \
    ConventionBasedMessageHandler, ConventionBasedMessageBus, Serializer


class EventSource(Identifiable):

    def __init__(self):
        Identifiable.__init__(self)
        self.version = 0
        self._recorded_events = []

    def get_version(self):
        return self.version

    def get_recorded_events(self):
        return self._recorded_events

    def clear_recorded_events(self):
        self._recorded_events = []

    def _record(self, event):
        self._apply(event)
        self._recorded_events.append(event)

    _record_that = _record_event = _record # sugar alias

    def _apply(self, event):
        method_name = '_apply_' + snake_case(event.__class__.__name__)
        method = getattr(self, method_name, None)
        if not method or not callable(method):
            raise Exception('no %s method defined in class' % (method_name))
        method(event)

    def load_from_history(self, events):
        """reconstitutes state of this object based on its history of events"""
        for event in events:
            self._apply(event)
        self.version = len(events)

    reconstitute_from_history = load_from_history # sugar alias


class AggregateWithEventSourcing(Aggregate, EventSource):

    def __init__(self):
        Aggregate.__init__(self)
        EventSource.__init__(self)



class WorkFlow(ConventionBasedMessageHandler):
    """
    WorkFlows listen for events and instruct other parts of the system to
    perform tasks based upon the events.
    This is juxtaposed to aggregates which are told to do something and then
    alert the world that they performed some action.
    This could be generalized into the following: WorkFlows listen to events and
    dispatch commands while aggregates receive commands and publish events.

    Workflows are commonly confused or mixed with Sagas but are different.

    A workflow is built on top of a state machine and the main difference between a
    state machine and activity diagram (i.e. workflow) is that the focus is on
    actions instead of states and the transitions occur when an action is
    completed, instead of when an event occurs.
    """
    __metaclass__ = ABCMeta

    @abstractmethod
    def get_undispatched_messages(self):
        pass

    @abstractmethod
    def clear_undispatched_messages(self):
        pass

class WorkFlowWithEventSourcing(Aggregate, EventSource):

    def __init__(self):
        Aggregate.__init__(self)
        EventSource.__init__(self)
        self.dispatches = []

    def dispatch(self, command):
        self.dispatches.append(command)

    def get_undispatched_messages(self):
        return self.dispatches

    def clear_dispatched_messages(self):
        self.dispatches = []

    def clear_recorded_events(self):
        self.clear_recorded_events()
        self.clear_dispatched_messages()


class EventStore:

    def __init__(self, db, mediator, serializer):
        if not isinstance(mediator, ConventionBasedMessageBus):
            raise ValueError('mediator should be of type ConventionBasedMessageBus')
        if not isinstance(serializer, Serializer):
            raise ValueError('serializer should be of type Serializer')
        self._mediator = mediator
        self._serializer = serializer
        self._db = db

    def initialize(self):
        sql = """
            -- The 20 in BIGINT(20) doesn't mean size..it only affects the zero-fill
            -- http://stackoverflow.com/questions/3135804/types-in-mysql-bigint20-vs-int20/3135854
            -- https://www.percona.com/blog/2014/12/19/store-uuid-optimized-way/
            CREATE TABLE IF NOT EXISTS events (
                stream_id VARBINARY(16) NOT NULL,
                stream_id_text VARCHAR(36) GENERATED ALWAYS AS (
                    LOWER(CONCAT(
                        SUBSTRING(HEX(stream_id), 9, 8), '-',
                        SUBSTRING(HEX(stream_id), 5, 1),
                        SUBSTRING(HEX(stream_id), 6, 3), '-',
                        SUBSTRING(HEX(stream_id), 1, 1),
                        SUBSTRING(HEX(stream_id), 2, 3), '-',
                        SUBSTRING(HEX(stream_id), 17, 4), '-',
                        SUBSTRING(HEX(stream_id), 21, 12)
                    ))
                ) VIRTUAL,
                stream_version BIGINT(20) UNSIGNED NOT NULL,
                stream_type VARCHAR(191) NOT NULL, -- 191 because of utf8mb4 using 4 bytes per character while limit of index is 767 bytes
                event_id BINARY(16) NOT NULL,
                event_id_text VARCHAR(36) GENERATED ALWAYS AS (
                    LOWER(CONCAT(
                        SUBSTRING(HEX(event_id), 9, 8), '-',
                        SUBSTRING(HEX(event_id), 5, 1),
                        SUBSTRING(HEX(event_id), 6, 3), '-',
                        SUBSTRING(HEX(event_id), 1, 1),
                        SUBSTRING(HEX(event_id), 2, 3), '-',
                        SUBSTRING(HEX(event_id), 17, 4), '-',
                        SUBSTRING(HEX(event_id), 21, 12)
                    ))
                ) VIRTUAL,
                event_type VARCHAR(255) NOT NULL,
                event_data TEXT NOT NULL,
                event_date DATETIME NOT NULL, -- utc
                -- correlation_id VARBINARY(16),
                -- causation_id VARBINARY(16),
                -- causation_event_ordinal BIGINT(20) UNSIGNED,
                UNIQUE (stream_id, stream_version, stream_type),
                UNIQUE KEY (event_id)
            ) ENGINE = MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
        """
        self._db.get_cursor().execute(sql)
        self._db.commit()

    def destroy(self):
        sql = 'DROP TABLE IF EXISTS events'
        self._db.get_cursor().execute(sql)
        self._db.commit()

    def get_version(self, event_source_id):
        sql = "SELECT MAX(stream_version) as version FROM events WHERE stream_id = uuid_to_bin('%s')" % (str(event_source_id))
        cursor = self._db.get_cursor()
        cursor.execute(sql)
        rv = cursor.fetchone()
        cursor.close()
        return rv['version']

    def get_events(self, event_source_id):
        sql = "SELECT * FROM events WHERE stream_id = uuid_to_bin('%s') ORDER BY stream_version ASC" % (event_source_id)
        cursor = self._db.get_cursor()
        cursor.execute(sql)
        descriptors = cursor.fetchall()
        events = []
        cursor.close()
        for descriptor in descriptors:
            events.append(self._serializer.deserialize(descriptor['event_data']))
        return events

    def save_events_for_aggregate(self, aggregate_id, aggregate_type, events, expected_version):
        if expected_version < 0:
            raise ValueError('"expected_version" cannot be negative')
        self._save_events(aggregate_id, aggregate_type, expected_version, events)
        self._publish_events(events)

    def save_events_for_workflow(self, workflow_id, workflow_type, events, expected_version, dispatches):
        self._save_events(workflow_id, workflow_type, expected_version, events)
        self._dispatch_commands(dispatches)

    def _save_events(self, event_source_id, event_source_type, expected_version, events):
        if len(events) == 0: return

        found_version = self.get_version(event_source_id)

        if expected_version < 0:
            raise ValueError('"expected_version" cannot be negative')

        if found_version is None and not (expected_version == 0 or expected_version is None):
            raise ValueError('the initial "expected_version" should be "0" or "None"')

        if found_version and found_version != expected_version:
            raise ConcurrencyException()

        event_descriptors = []

        i = expected_version;

        for event in events:
            i = i + 1
            event_descriptors.append({
                "stream_id": event_source_id,
                "stream_type": event_source_type,
                "stream_version": i,
                "event_id": event.id,
                "event_type": event.__class__.__name__,
                "event_data": self._serializer.serialize(event)
            })

        sql = "INSERT INTO events(stream_id, stream_type, stream_version, event_id, event_type, event_date, event_data) VALUES "
        now = datetime.utcnow()
        place_holders = []
        values = []

        for descriptor in event_descriptors:
            values.extend([descriptor['stream_id'], descriptor['stream_type'], descriptor['stream_version'], descriptor['event_id'], descriptor['event_type'], now, descriptor['event_data']])
            place_holders.append("(uuid_to_bin('%s'), '%s', %d, uuid_to_bin('%s'), '%s', '%s', '%s')")

        sql = sql + ', '.join(place_holders)
        sql = sql % tuple(values)
        cursor = self._db.get_cursor()
        cursor.execute(sql)
        self._db.commit()
        cursor.close()

    def _publish_events(self, events):
        for event in events:
            self._mediator.publish(event)

    def _dispatch_commands(self, commands):
        for command in commands:
            self._mediator.send(command)


class EventSourcingRepository(Repository):

    def __init__(self, store):
        self._store = store

    def save(self, event_source):
        if not event_source: return

        if not isinstance(event_source, EventSource):
            raise ValueError('"event_source" must be a child of EventSource')

        if not event_source.id: return

        event_source_type = event_source.__class__.__name__

        if isinstance(event_source, WorkFlow):
            self._store.save_events_for_workflow(
                event_source.id,
                event_source_type,
                event_source.get_recorded_events(),
                event_source.version,
                event_source.dispatches
            )
        else:
            self._store.save_events_for_aggregate(
                event_source.id,
                event_source_type,
                event_source.get_recorded_events(),
                event_source.version
            )

        event_source.clear_recorded_events()

    def get(self, cls, event_source_id):
        if not issubclass(cls, EventSource):
            raise ValueError('cls must inherit from "EventSource"')

        history = self._store.get_events(event_source_id)
        obj = cls()
        obj.id = event_source_id
        obj.load_from_history(history)
        return obj


# Errors
# ---

class ValidationError(ValueError):
    pass

class AggregateVersionException(Exception):
    pass

class AggregateDeletedException(Exception):
    pass

class AggregateNotFoundException(Exception):
    pass

class ConcurrencyException(Exception):
    """Occurs when version on the aggregate does not match the version
    supplied by the client"""
    pass

Unit tests for this code are:

class SampleEvent(Event):
    def get_event_source_id(self):
        pass

class EventStoreTestCase(unittest.TestCase):

    def setUp(self):
        mediator = Mediator()
        serializer = PickleSerializer()
        self.es = es = EventStore(db, mediator, serializer)
        es.initialize()

    def tearDown(self):
        self.es.destroy()

    def test_should_increse_version_by_one(self):
        aggr_id = uuid.uuid1()
        self.es.save_events_for_aggregate(aggr_id, 'demo', [SampleEvent()], 0)
        self.assertEquals(1, self.es.get_version(aggr_id))
        self.es.save_events_for_aggregate(aggr_id, 'demo', [SampleEvent()], 1)
        self.assertEquals(2, self.es.get_version(aggr_id))

    def test_should_throw_concurrency_exception_when_appending_events_with_version_lower_than_exising_version(self):
        with self.assertRaises(ConcurrencyException):
            aggr_id = uuid.uuid1()
            self.es.save_events_for_aggregate(aggr_id, 'demo', [SampleEvent()], 0)
            self.es.save_events_for_aggregate(aggr_id, 'demo', [SampleEvent()], 0)

    def test_should_append_events(self):
        events = [SampleEvent(), SampleEvent()]
        self.es.save_events_for_aggregate(uuid.uuid1(), 'demo', events, 0)

    def test_event_store_should_restore_events_in_fifo_order(self):
        aggr_id = uuid.uuid1()
        events = [SampleEvent(), SampleEvent()]
        self.es.save_events_for_aggregate(aggr_id, 'demo', events, 0)
        store_events = self.es.get_events(aggr_id)
        self.assertEquals(len(events), len(store_events))
        for event in store_events:
            self.assertTrue(isinstance(event, SampleEvent))

        for pair in list(zip(events, store_events)):
            self.assertEquals(pair[0].id, pair[1].id)

    def eventStore_should_initialize_db(self):
        sql = "SHOW TABLES LIKE es_events"

class SampleAggregate(AggregateWithEventSourcing):

    def __init__(self):
        AggregateWithEventSourcing.__init__(self)
        self.id=uuid.uuid1()

    def do_something(self):
        event = SampleEvent()
        self._record_that(event)

    def _apply_sample_event(self, evt):
        self.sample_event_applied = True

class EventSourcingRepositoryTestCase(unittest.TestCase):

    def setUp(self):
        mediator = Mediator()
        serializer = PickleSerializer()
        self.es = es = EventStore(db, mediator, serializer)
        es.initialize()
        self.repository = EventSourcingRepository(es)

    def tearDown(self):
        self.es.destroy()

    def test_repository_saves_and_gets_aggregate(self):
        aggr = SampleAggregate()
        aggr.do_something()
        self.repository.save(aggr)
        db_aggr = self.repository.get(SampleAggregate, aggr.id)
        self.assertIsNotNone(db_aggr)
        self.assertTrue(isinstance(db_aggr, SampleAggregate))
        self.assertEquals(aggr.id, db_aggr.id)
        self.assertTrue(aggr.equals(db_aggr))
        self.assertTrue(aggr==db_aggr)
        self.assertTrue(db_aggr.sample_event_applied)
\$\endgroup\$
  • 1
    \$\begingroup\$ If a parameter is of wrong type, you ought to raise TypeError (or a subclass) instead of ValueError. For instance: raise TypeError(repr(type(mediator))). \$\endgroup\$ – Laurent LAPORTE Nov 14 '16 at 23:03
  • \$\begingroup\$ Your code looks complex with Abstract classes. Are they really useful? \$\endgroup\$ – Laurent LAPORTE Nov 14 '16 at 23:06
  • \$\begingroup\$ Consider using an ORM like SqlAlchemy for your database instead of implementing all the boilerplate yourself in EventStore. \$\endgroup\$ – Laurent LAPORTE Nov 14 '16 at 23:09
  • 1
    \$\begingroup\$ It is a best practice to have a root exception, for instance EventSourcingError, and then inherit from it to create a (small) exceptions hierarchy. That way, users of your API can easily implement exception handling. \$\endgroup\$ – Laurent LAPORTE Nov 14 '16 at 23:13
  • 1
    \$\begingroup\$ Warning: you have an infinite recursion call with clear_recorded_events. \$\endgroup\$ – Laurent LAPORTE Nov 14 '16 at 23:33
3
\$\begingroup\$

I started to review your code. Look # review: comments bellow:

# Is `Identifiable` base class useful? Does it add something signifiant?
class EventSource(object):
    def __init__(self):
        # review: use `super` call here: that way you can change your base class
        super(EventSource, self).__init__()
        # review: add the `self.id` field here.
        self.id = None
        self.version = 0
        self._recorded_events = []

    # review: `get_version()` is useless since `self.version` is public
    # def get_version(self):
    #     return self.version

    def get_recorded_events(self):
        return self._recorded_events

    def clear_recorded_events(self):
        # review: if you want to "clear" a list, do it as follow.
        # That way, any reference to `self._recorded_events` won't be broken.
        self._recorded_events[:] = []

    def _record(self, event):
        self._apply(event)
        self._recorded_events.append(event)

    # review: No need to add "sugar alias" for "protected" methods.
    # _record_that = _record_event = _record  # sugar alias

    # review: It's more natural to raise `AttributeError` instead of `Exception`.
    def _apply(self, event):
        method_name = '_apply_' + snake_case(event.__class__.__name__)
        method = getattr(self, method_name)  # review: raise `AttributeError` if missing.
        method(event)  # review: raise `TypeError` if not callable

    def load_from_history(self, events):
        """reconstitutes state of this object based on its history of events"""
        for event in events:
            self._apply(event)
        self.version = len(events)

    # review: Is it really usefull
    reconstitute_from_history = load_from_history  # sugar alias

Additional remaks:

  • get_recorded_events() method returns self._recorded_events, which is a mutable object (a Python list), So, the end-user can modify it. If it can, drop this method and turn self._recorded_events to a "public" attribute: self.recorded_events.
  • self.id is not used in this class.
  • What is the role of self.version?

TL;TR.

In summary:

  • Try to be more Pythonic,
  • ABC/metaclasses are not useful,
  • There are too many one-line methods: consider refactoring,
  • The class hierarchy is unnecessary too complex,
  • You can use functions (really).

I think you design your application to be object oriented (probably from UML or Java). It's good, but in Python you can also use functional programing.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy