Split into Events & Commands

Author: heykarimoff

src/allocation/domain/commands.py

@@ -0,0 +1,35 @@
+from dataclasses import dataclass
+from datetime import date
+from typing import Optional
+
+
+class Command:
+    pass
+
+
+@dataclass
+class CreateBatch(Command):
+    reference: str
+    sku: str
+    qty: int
+    eta: Optional[date] = None
+
+
+@dataclass
+class ChangeBatchQuantity(Command):
+    reference: str
+    qty: int
+
+
+@dataclass
+class Allocate(Command):
+    orderid: str
+    sku: str
+    qty: int
+
+
+@dataclass
+class Deallocate(Command):
+    orderid: str
+    sku: str
+    qty: int

src/allocation/domain/events.py

@@ -10,31 +10,3 @@ class Event:
 @dataclass
 class OutOfStock(Event):
     sku: str
-
-
-@dataclass
-class BatchCreated(Event):
-    reference: str
-    sku: str
-    qty: int
-    eta: Optional[date] = None
-
-
-@dataclass
-class BatchQuantityChanged(Event):
-    reference: str
-    qty: int
-
-
-@dataclass
-class AllocationRequired(Event):
-    orderid: str
-    sku: str
-    qty: int
-
-
-@dataclass
-class DeallocationRequired(Event):
-    orderid: str
-    sku: str
-    qty: int

src/allocation/domain/model.py

@@ -2,7 +2,8 @@
 from datetime import date
 from typing import List, NewType, Optional, Set
 
-from allocation.domain.events import AllocationRequired, Event, OutOfStock
+from allocation.domain.commands import Allocate
+from allocation.domain.events import Event, OutOfStock
 
 Reference = NewType("Reference", str)
 Sku = NewType("Sku", str)
@@ -113,6 +114,4 @@ def change_batch_quantity(self, reference: Reference, qty: Quantity):
         batch._purchased_quantity = qty
         while batch.allocated_quaitity > qty:
             line = batch.deallocate_one()
-            self.events.append(
-                AllocationRequired(line.orderid, line.sku, line.qty)
-            )
+            self.events.append(Allocate(line.orderid, line.sku, line.qty))

src/allocation/entrypoints/flask_app.py

@@ -1,8 +1,8 @@
 from datetime import datetime
 
 from allocation.adapters import orm
-from allocation.domain import events
-from allocation.service_layer import handlers, unit_of_work, messagebus
+from allocation.domain import commands, events
+from allocation.service_layer import handlers, messagebus, unit_of_work
 from flask import Flask, jsonify, request
 
 orm.start_mappers()
@@ -17,14 +17,14 @@ def add_batch_endpoint():
     if eta is not None:
         eta = datetime.fromisoformat(eta).date()
 
-    event = events.BatchCreated(
+    message = commands.CreateBatch(
         request.json["reference"],
         request.json["sku"],
         request.json["qty"],
         eta=eta,
     )
 
-    messagebus.handle(event, uow)
+    messagebus.handle(message, uow)
 
     return "OK", 201
 
@@ -34,10 +34,10 @@ def allocate_endpoint():
     uow = unit_of_work.SqlAlchemyUnitOfWork()
 
     try:
-        event = events.AllocationRequired(
+        message = commands.Allocate(
             request.json["orderid"], request.json["sku"], request.json["qty"]
         )
-        results = messagebus.handle(event, uow)
+        results = messagebus.handle(message, uow)
         batchref = results.pop(0)
     except handlers.InvalidSku as e:
         return {"message": str(e)}, 400
@@ -52,10 +52,10 @@ def allocate_endpoint():
 def deallocate_endpoint():
     uow = unit_of_work.SqlAlchemyUnitOfWork()
     try:
-        event = events.DeallocationRequired(
+        message = commands.Deallocate(
             request.json["orderid"], request.json["sku"], request.json["qty"]
         )
-        results = messagebus.handle(event, uow)
+        results = messagebus.handle(message, uow)
     except handlers.InvalidSku as e:
         return {"message": str(e)}, 400
 

src/allocation/service_layer/handlers.py

@@ -2,17 +2,21 @@
 from typing import Optional
 
 from allocation.adapters import email
-from allocation.domain import events, model
+from allocation.domain import commands, events, model
 from allocation.service_layer import unit_of_work
 
 
 class InvalidSku(Exception):
     pass
 
 
-def add_batch(event: events.BatchCreated, uow: unit_of_work.AbstractUnitOfWork):
-    sku = event.sku
-    batch = model.Batch(event.reference, event.sku, event.qty, event.eta)
+def add_batch(
+    message: commands.CreateBatch, uow: unit_of_work.AbstractUnitOfWork
+):
+    sku = message.sku
+    batch = model.Batch(
+        message.reference, message.sku, message.qty, message.eta
+    )
     with uow:
         product = uow.products.get(sku=sku)
         if product is None:
@@ -22,20 +26,24 @@ def add_batch(event: events.BatchCreated, uow: unit_of_work.AbstractUnitOfWork):
         uow.commit()
 
 
-def change_batch_quantity(event: events.BatchQuantityChanged, uow):
+def change_batch_quantity(message: commands.ChangeBatchQuantity, uow):
     with uow:
-        product = uow.products.get_by_batch_reference(reference=event.reference)
-        product.change_batch_quantity(reference=event.reference, qty=event.qty)
+        product = uow.products.get_by_batch_reference(
+            reference=message.reference
+        )
+        product.change_batch_quantity(
+            reference=message.reference, qty=message.qty
+        )
         uow.commit()
 
 
 def allocate(
-    event: events.AllocationRequired, uow: unit_of_work.AbstractUnitOfWork
+    message: commands.Allocate, uow: unit_of_work.AbstractUnitOfWork
 ) -> str:
-    line = model.OrderLine(event.orderid, event.sku, event.qty)
+    line = model.OrderLine(message.orderid, message.sku, message.qty)
 
     with uow:
-        product = uow.products.get(sku=event.sku)
+        product = uow.products.get(sku=message.sku)
 
         if product is None:
             raise InvalidSku(f"Invalid sku {line.sku}")
@@ -47,11 +55,11 @@ def allocate(
 
 
 def deallocate(
-    event: events.DeallocationRequired, uow: unit_of_work.AbstractUnitOfWork
+    message: commands.Deallocate, uow: unit_of_work.AbstractUnitOfWork
 ) -> None:
-    line = model.OrderLine(event.orderid, event.sku, event.qty)
+    line = model.OrderLine(message.orderid, message.sku, message.qty)
     with uow:
-        product = uow.products.get(sku=event.sku)
+        product = uow.products.get(sku=message.sku)
 
         if product is None:
             raise InvalidSku(f"Invalid sku {line.sku}")
@@ -61,6 +69,6 @@ def deallocate(
 
 
 def send_out_of_stock_notification(
-    event: events.OutOfStock, uow: unit_of_work.AbstractUnitOfWork
+    message: events.OutOfStock, uow: unit_of_work.AbstractUnitOfWork
 ):
-    email.send_mail("stock-admin@made.com", f"Out of stock: {event.sku}")
+    email.send_mail("stock-admin@made.com", f"Out of stock: {message.sku}")

src/allocation/service_layer/messagebus.py

@@ -1,24 +1,38 @@
-from allocation.domain import events
-from allocation.service_layer import handlers
-from allocation.service_layer import unit_of_work
+from typing import Union
 
+from allocation.domain import commands, events
+from allocation.service_layer import handlers, unit_of_work
 
-def handle(event: events.Event, uow: unit_of_work.AbstractUnitOfWork):
+Message = Union[commands.Command, events.Event]
+
+
+def handle(message: Message, uow: unit_of_work.AbstractUnitOfWork):
     results = []
-    queue = [event]
+    queue = [message]
     while queue:
-        event = queue.pop(0)
-        for handler in HANDLERS[type(event)]:
-            results.append(handler(event, uow))
+        message = queue.pop(0)
+        if isinstance(message, events.Event):
+            for handler in EVENT_HANDLERS[type(message)]:
+                results.append(handler(message, uow))
+                queue.extend(uow.collect_new_events())
+        elif isinstance(message, commands.Command):
+            handler = COMMAND_HANDLERS[type(message)]
+            results.append(handler(message, uow))
             queue.extend(uow.collect_new_events())
+        else:
+            raise TypeError(f"Unknown message type {type(message)}")
 
     return results
 
 
-HANDLERS = {
-    events.BatchCreated: [handlers.add_batch],
-    events.BatchQuantityChanged: [handlers.change_batch_quantity],
-    events.AllocationRequired: [handlers.allocate],
-    events.DeallocationRequired: [handlers.deallocate],
+EVENT_HANDLERS = {
     events.OutOfStock: [handlers.send_out_of_stock_notification],
 }
+
+
+COMMAND_HANDLERS = {
+    commands.CreateBatch: handlers.add_batch,
+    commands.ChangeBatchQuantity: handlers.change_batch_quantity,
+    commands.Allocate: handlers.allocate,
+    commands.Deallocate: handlers.deallocate,
+}