pAvYwxJP

1# -*- coding: utf-8 -*-
2
3from abc import ABCMeta, abstractmethod
4from datetime import *
5from . import Immutable, Identifiable, Aggregate, Repository, snake_case, 
6    ConventionBasedMessageHandler, ConventionBasedMessageBus, Serializer
7
8
9class EventSource(Identifiable):
10
11    def __init__(self):
12        Identifiable.__init__(self)
13        self.version = 0
14        self._recorded_events = []
15
16    def get_version(self):
17        return self.version
18
19    def get_recorded_events(self):
20        return self._recorded_events
21
22    def clear_recorded_events(self):
23        self._recorded_events = []
24
25    def _record(self, event):
26        self._apply(event)
27        self._recorded_events.append(event)
28
29    _record_that = _record_event = _record # sugar alias
30
31    def _apply(self, event):
32        method_name = '_apply_' + snake_case(event.__class__.__name__)
33        method = getattr(self, method_name, None)
34        if not method or not callable(method):
35            raise Exception('no %s method defined in class' % (method_name))
36        method(event)
37
38    def load_from_history(self, events):
39        """reconstitutes state of this object based on its history of events"""
40        for event in events:
41            self._apply(event)
42        self.version = len(events)
43
44    reconstitute_from_history = load_from_history # sugar alias
45
46
47class AggregateWithEventSourcing(Aggregate, EventSource):
48
49    def __init__(self):
50        Aggregate.__init__(self)
51        EventSource.__init__(self)
52
53
54
55class WorkFlow(ConventionBasedMessageHandler):
56    """
57    WorkFlows listen for events and instruct other parts of the system to
58    perform tasks based upon the events.
59    This is juxtaposed to aggregates which are told to do something and then
60    alert the world that they performed some action.
61    This could be generalized into the following: WorkFlows listen to events and
62    dispatch commands while aggregates receive commands and publish events.
63
64    Workflows are commonly confused or mixed with Sagas but are different.
65
66    A workflow is built on top of a state machine and the main difference between a
67    state machine and activity diagram (i.e. workflow) is that the focus is on
68    actions instead of states and the transitions occur when an action is
69    completed, instead of when an event occurs.
70    """
71    __metaclass__ = ABCMeta
72
73    @abstractmethod
74    def get_undispatched_messages(self):
75        pass
76
77    @abstractmethod
78    def clear_undispatched_messages(self):
79        pass
80
81class WorkFlowWithEventSourcing(Aggregate, EventSource):
82
83    def __init__(self):
84        Aggregate.__init__(self)
85        EventSource.__init__(self)
86        self.dispatches = []
87
88    def dispatch(self, command):
89        self.dispatches.append(command)
90
91    def get_undispatched_messages(self):
92        return self.dispatches
93
94    def clear_dispatched_messages(self):
95        self.dispatches = []
96
97    def clear_recorded_events(self):
98        self.clear_recorded_events()
99        self.clear_dispatched_messages()
100
101
102class EventStore:
103
104    def __init__(self, db, mediator, serializer):
105        if not isinstance(mediator, ConventionBasedMessageBus):
106            raise ValueError('mediator should be of type ConventionBasedMessageBus')
107        if not isinstance(serializer, Serializer):
108            raise ValueError('serializer should be of type Serializer')
109        self._mediator = mediator
110        self._serializer = serializer
111        self._db = db
112
113    def initialize(self):
114        sql = """
115            -- The 20 in BIGINT(20) doesn't mean size..it only affects the zero-fill
116            -- http://stackoverflow.com/questions/3135804/types-in-mysql-bigint20-vs-int20/3135854
117            -- https://www.percona.com/blog/2014/12/19/store-uuid-optimized-way/
118            CREATE TABLE IF NOT EXISTS events (
119                stream_id VARBINARY(16) NOT NULL,
120                stream_id_text VARCHAR(36) GENERATED ALWAYS AS (
121                    LOWER(CONCAT(
122                        SUBSTRING(HEX(stream_id), 9, 8), '-',
123                        SUBSTRING(HEX(stream_id), 5, 1),
124                        SUBSTRING(HEX(stream_id), 6, 3), '-',
125                        SUBSTRING(HEX(stream_id), 1, 1),
126                        SUBSTRING(HEX(stream_id), 2, 3), '-',
127                        SUBSTRING(HEX(stream_id), 17, 4), '-',
128                        SUBSTRING(HEX(stream_id), 21, 12)
129                    ))
130                ) VIRTUAL,
131                stream_version BIGINT(20) UNSIGNED NOT NULL,
132                stream_type VARCHAR(191) NOT NULL, -- 191 because of utf8mb4 using 4 bytes per character while limit of index is 767 bytes
133                event_id BINARY(16) NOT NULL,
134                event_id_text VARCHAR(36) GENERATED ALWAYS AS (
135                    LOWER(CONCAT(
136                        SUBSTRING(HEX(event_id), 9, 8), '-',
137                        SUBSTRING(HEX(event_id), 5, 1),
138                        SUBSTRING(HEX(event_id), 6, 3), '-',
139                        SUBSTRING(HEX(event_id), 1, 1),
140                        SUBSTRING(HEX(event_id), 2, 3), '-',
141                        SUBSTRING(HEX(event_id), 17, 4), '-',
142                        SUBSTRING(HEX(event_id), 21, 12)
143                    ))
144                ) VIRTUAL,
145                event_type VARCHAR(255) NOT NULL,
146                event_data TEXT NOT NULL,
147                event_date DATETIME NOT NULL, -- utc
148                -- correlation_id VARBINARY(16),
149                -- causation_id VARBINARY(16),
150                -- causation_event_ordinal BIGINT(20) UNSIGNED,
151                UNIQUE (stream_id, stream_version, stream_type),
152                UNIQUE KEY (event_id)
153            ) ENGINE = MyISAM DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
154        """
155        self._db.get_cursor().execute(sql)
156        self._db.commit()
157
158    def destroy(self):
159        sql = 'DROP TABLE IF EXISTS events'
160        self._db.get_cursor().execute(sql)
161        self._db.commit()
162
163    def get_version(self, event_source_id):
164        sql = "SELECT MAX(stream_version) as version FROM events WHERE stream_id = uuid_to_bin('%s')" % (str(event_source_id))
165        cursor = self._db.get_cursor()
166        cursor.execute(sql)
167        rv = cursor.fetchone()
168        cursor.close()
169        return rv['version']
170
171    def get_events(self, event_source_id):
172        sql = "SELECT * FROM events WHERE stream_id = uuid_to_bin('%s') ORDER BY stream_version ASC" % (event_source_id)
173        cursor = self._db.get_cursor()
174        cursor.execute(sql)
175        descriptors = cursor.fetchall()
176        events = []
177        cursor.close()
178        for descriptor in descriptors:
179            events.append(self._serializer.deserialize(descriptor['event_data']))
180        return events
181
182    def save_events_for_aggregate(self, aggregate_id, aggregate_type, events, expected_version):
183        if expected_version < 0:
184            raise ValueError('"expected_version" cannot be negative')
185        self._save_events(aggregate_id, aggregate_type, expected_version, events)
186        self._publish_events(events)
187
188    def save_events_for_workflow(self, workflow_id, workflow_type, events, expected_version, dispatches):
189        self._save_events(workflow_id, workflow_type, expected_version, events)
190        self._dispatch_commands(dispatches)
191
192    def _save_events(self, event_source_id, event_source_type, expected_version, events):
193        if len(events) == 0: return
194
195        found_version = self.get_version(event_source_id)
196
197        if expected_version < 0:
198            raise ValueError('"expected_version" cannot be negative')
199
200        if found_version is None and not (expected_version == 0 or expected_version is None):
201            raise ValueError('the initial "expected_version" should be "0" or "None"')
202
203        if found_version and found_version != expected_version:
204            raise ConcurrencyException()
205
206        event_descriptors = []
207
208        i = expected_version;
209
210        for event in events:
211            i = i + 1
212            event_descriptors.append({
213                "stream_id": event_source_id,
214                "stream_type": event_source_type,
215                "stream_version": i,
216                "event_id": event.id,
217                "event_type": event.__class__.__name__,
218                "event_data": self._serializer.serialize(event)
219            })
220
221        sql = "INSERT INTO events(stream_id, stream_type, stream_version, event_id, event_type, event_date, event_data) VALUES "
222        now = datetime.utcnow()
223        place_holders = []
224        values = []
225
226        for descriptor in event_descriptors:
227            values.extend([descriptor['stream_id'], descriptor['stream_type'], descriptor['stream_version'], descriptor['event_id'], descriptor['event_type'], now, descriptor['event_data']])
228            place_holders.append("(uuid_to_bin('%s'), '%s', %d, uuid_to_bin('%s'), '%s', '%s', '%s')")
229
230        sql = sql + ', '.join(place_holders)
231        sql = sql % tuple(values)
232        cursor = self._db.get_cursor()
233        cursor.execute(sql)
234        self._db.commit()
235        cursor.close()
236
237    def _publish_events(self, events):
238        for event in events:
239            self._mediator.publish(event)
240
241    def _dispatch_commands(self, commands):
242        for command in commands:
243            self._mediator.send(command)
244
245
246class EventSourcingRepository(Repository):
247
248    def __init__(self, store):
249        self._store = store
250
251    def save(self, event_source):
252        if not event_source: return
253
254        if not isinstance(event_source, EventSource):
255            raise ValueError('"event_source" must be a child of EventSource')
256
257        if not event_source.id: return
258
259        event_source_type = event_source.__class__.__name__
260
261        if isinstance(event_source, WorkFlow):
262            self._store.save_events_for_workflow(
263                event_source.id,
264                event_source_type,
265                event_source.get_recorded_events(),
266                event_source.version,
267                event_source.dispatches
268            )
269        else:
270            self._store.save_events_for_aggregate(
271                event_source.id,
272                event_source_type,
273                event_source.get_recorded_events(),
274                event_source.version
275            )
276
277        event_source.clear_recorded_events()
278
279    def get(self, cls, event_source_id):
280        if not issubclass(cls, EventSource):
281            raise ValueError('cls must inherit from "EventSource"')
282
283        history = self._store.get_events(event_source_id)
284        obj = cls()
285        obj.id = event_source_id
286        obj.load_from_history(history)
287        return obj
288
289
290# Errors
291# ---
292
293class ValidationError(ValueError):
294    pass
295
296class AggregateVersionException(Exception):
297    pass
298
299class AggregateDeletedException(Exception):
300    pass
301
302class AggregateNotFoundException(Exception):
303    pass
304
305class ConcurrencyException(Exception):
306    """Occurs when version on the aggregate does not match the version
307    supplied by the client"""
308    pass
309	
310class SampleEvent(Event):
311    def get_event_source_id(self):
312        pass
313
314class EventStoreTestCase(unittest.TestCase):
315
316    def setUp(self):
317        mediator = Mediator()
318        serializer = PickleSerializer()
319        self.es = es = EventStore(db, mediator, serializer)
320        es.initialize()
321
322    def tearDown(self):
323        self.es.destroy()
324
325    def test_should_increse_version_by_one(self):
326        aggr_id = uuid.uuid1()
327        self.es.save_events_for_aggregate(aggr_id, 'demo', [SampleEvent()], 0)
328        self.assertEquals(1, self.es.get_version(aggr_id))
329        self.es.save_events_for_aggregate(aggr_id, 'demo', [SampleEvent()], 1)
330        self.assertEquals(2, self.es.get_version(aggr_id))
331
332    def test_should_throw_concurrency_exception_when_appending_events_with_version_lower_than_exising_version(self):
333        with self.assertRaises(ConcurrencyException):
334            aggr_id = uuid.uuid1()
335            self.es.save_events_for_aggregate(aggr_id, 'demo', [SampleEvent()], 0)
336            self.es.save_events_for_aggregate(aggr_id, 'demo', [SampleEvent()], 0)
337
338    def test_should_append_events(self):
339        events = [SampleEvent(), SampleEvent()]
340        self.es.save_events_for_aggregate(uuid.uuid1(), 'demo', events, 0)
341
342    def test_event_store_should_restore_events_in_fifo_order(self):
343        aggr_id = uuid.uuid1()
344        events = [SampleEvent(), SampleEvent()]
345        self.es.save_events_for_aggregate(aggr_id, 'demo', events, 0)
346        store_events = self.es.get_events(aggr_id)
347        self.assertEquals(len(events), len(store_events))
348        for event in store_events:
349            self.assertTrue(isinstance(event, SampleEvent))
350
351        for pair in list(zip(events, store_events)):
352            self.assertEquals(pair[0].id, pair[1].id)
353
354    def eventStore_should_initialize_db(self):
355        sql = "SHOW TABLES LIKE es_events"
356
357class SampleAggregate(AggregateWithEventSourcing):
358
359    def __init__(self):
360        AggregateWithEventSourcing.__init__(self)
361        self.id=uuid.uuid1()
362
363    def do_something(self):
364        event = SampleEvent()
365        self._record_that(event)
366
367    def _apply_sample_event(self, evt):
368        self.sample_event_applied = True
369
370class EventSourcingRepositoryTestCase(unittest.TestCase):
371
372    def setUp(self):
373        mediator = Mediator()
374        serializer = PickleSerializer()
375        self.es = es = EventStore(db, mediator, serializer)
376        es.initialize()
377        self.repository = EventSourcingRepository(es)
378
379    def tearDown(self):
380        self.es.destroy()
381
382    def test_repository_saves_and_gets_aggregate(self):
383        aggr = SampleAggregate()
384        aggr.do_something()
385        self.repository.save(aggr)
386        db_aggr = self.repository.get(SampleAggregate, aggr.id)
387        self.assertIsNotNone(db_aggr)
388        self.assertTrue(isinstance(db_aggr, SampleAggregate))
389        self.assertEquals(aggr.id, db_aggr.id)
390        self.assertTrue(aggr.equals(db_aggr))
391        self.assertTrue(aggr==db_aggr)
392        self.assertTrue(db_aggr.sample_event_applied)