Add signalling with event_received, event_pre_save. Fixes #10

Author: geeknam

esser/events.py

@@ -1,6 +1,6 @@
 from esser.validators import EsserValidator
-from esser.constants import AGGREGATE_KEY_DELIMITER
 from esser.exceptions import EventValidationException
+from esser.signals import event_pre_save
 
 
 class BaseEvent(object):
@@ -31,9 +31,19 @@ def attach_entity(self, entity):
         setattr(self.validator, 'aggregate', entity)
 
     def persist(self, attrs):
+        event_version = self.get_version()
+        # dispatch the signal before persisting
+        event_pre_save.send(
+            sender=self.__class__,
+            aggregate_name=self.entity.aggregate_name,
+            aggregate_id=self.entity.aggregate_id,
+            event_name=self.event_name,
+            version=event_version,
+            payload=attrs,
+        )
         return self.entity.repository.persist(
             aggregate_id=self.entity.aggregate_id,
-            version=self.get_version(),
+            version=event_version,
             event_type=self.event_name,
             attrs=attrs
         )

esser/handlers/__init__.py

@@ -1,5 +1,6 @@
 import importlib
 from collections import defaultdict
+from esser.signals import event_received
 from esser.registry import registry
 
 
@@ -14,27 +15,40 @@ def get_aggregate(aggregate_name, aggregate_id):
     return aggregate
 
 
-def handle_event(event, context):
-    event_name = event['EventName']
-    aggregate_id = event.get('AggregateId', None)
-    aggregate = get_aggregate(event['AggregateName'], aggregate_id)
-    event_class_attr = None
-    for event_key, cls in aggregate.__class__.__dict__.items():
-        if cls.__class__.__name__ == event_name:
-            event_class_attr = event_key
-    aggregate_event = getattr(aggregate, event_class_attr, None)
-    if aggregate_event:
-        return aggregate_event.save(attrs=event['Payload'])
-    return
-
-
-def handle_stream(event, context):
-    aggregates = defaultdict(dict)
-    for record in event['Records']:
-        keys = record['dynamodb']['Keys']
-        aggregate_name = keys['aggregate_name']['S']
-        aggregate_key = keys['aggregate_key']['S']
-        aggregate_id = aggregate_key.split(':')[0]
-        aggregate = get_aggregate(aggregate_name, aggregate_id)
-        aggregates[aggregate_name][aggregate_id] = aggregate.current_state
-    return aggregates
+class LambdaHandler(object):
+
+    def handle_event(self, event, context):
+        event_name = event['EventName']
+        aggregate_id = event.get('AggregateId', None)
+        aggregate = get_aggregate(event['AggregateName'], aggregate_id)
+        event_received.send(
+            sender=self.__class__,
+            aggregate_name=event['AggregateName'],
+            aggregate_id=aggregate_id,
+            payload=event['Payload']
+        )
+        event_class_attr = None
+        for event_key, cls in aggregate.__class__.__dict__.items():
+            if cls.__class__.__name__ == event_name:
+                event_class_attr = event_key
+        aggregate_event = getattr(aggregate, event_class_attr, None)
+        if aggregate_event:
+            return aggregate_event.save(attrs=event['Payload'])
+        return
+
+    def handle_stream(self, event, context):
+        aggregates = defaultdict(dict)
+        for record in event['Records']:
+            keys = record['dynamodb']['Keys']
+            aggregate_name = keys['aggregate_name']['S']
+            aggregate_key = keys['aggregate_key']['S']
+            aggregate_id = aggregate_key.split(':')[0]
+            aggregate = get_aggregate(aggregate_name, aggregate_id)
+            aggregates[aggregate_name][aggregate_id] = aggregate.current_state
+        return aggregates
+
+
+handler = LambdaHandler()
+
+handle_event = handler.handle_event
+handle_stream = handler.handle_stream

esser/signals/__init__.py

@@ -0,0 +1,19 @@
+import dispatch
+
+event_received = dispatch.Signal(
+    providing_args=['aggregate_name', 'aggregate_id', 'payload']
+)
+
+event_pre_save = dispatch.Signal(
+    providing_args=[
+        'aggregate_name', 'aggregate_id', 'payload',
+        'event_name', 'version',
+    ]
+)
+
+event_post_save = dispatch.Signal(
+    providing_args=[
+        'aggregate_name', 'aggregate_id', 'payload',
+        'event_name', 'version',
+    ]
+)

esser/signals/decorators.py

@@ -0,0 +1,21 @@
+
+
+def receiver(signal, **kwargs):
+    """
+    A decorator for connecting receivers to signals. Used by passing in the
+    signal (or list of signals) and keyword arguments to connect::
+        @receiver(post_save, sender=Sender)
+        def signal_receiver(sender, **kwargs):
+            ...
+        @receiver([post_save, post_delete], sender=Sender)
+        def signals_receiver(sender, **kwargs):
+            ...
+    """
+    def _decorator(func):
+        if isinstance(signal, (list, tuple)):
+            for s in signal:
+                s.connect(func, **kwargs)
+        else:
+            signal.connect(func, **kwargs)
+        return func
+    return _decorator

examples/items/aggregate.py

@@ -2,6 +2,7 @@
 from esser.reducer import BaseReducer
 from esser.registry import register
 from items import events
+from items import receivers
 
 
 class ItemReducer(BaseReducer):

examples/items/receivers.py

@@ -0,0 +1,16 @@
+from esser.signals.decorators import receiver
+from esser.signals import event_pre_save, event_received
+from esser.handlers import LambdaHandler
+from items.events import PriceUpdated
+
+
+@receiver(event_pre_save, sender=PriceUpdated)
+def check_price_update(sender, **kwargs):
+    print sender
+    print kwargs
+
+
+@receiver(event_received, sender=LambdaHandler)
+def do_something(sender, **kwargs):
+    print sender
+    print kwargs

requirements/base.txt

@@ -1,2 +1,3 @@
 pynamodb
-cerberus
+cerberus
+dispatcher