Start using Events to integrate Microservices

Author: heykarimoff

Makefile

@@ -20,7 +20,7 @@ logs:
 test: up test-e2e test-integration test-unit
 
 test-e2e:
-	docker-compose run --rm --no-deps --entrypoint=pytest app /tests/e2e -v -s
+	docker-compose run --rm --no-deps --entrypoint=pytest app /tests/e2e -vv -s
 
 test-integration:
 	docker-compose run --rm --no-deps --entrypoint=pytest app /tests/integration -v -s

docker-compose.yml

@@ -6,10 +6,12 @@ services:
       dockerfile: Dockerfile
     depends_on:
       - postgres
+      - redis
     environment:
       - DB_HOST=postgres
       - DB_PASSWORD=abc123
       - API_HOST=app
+      - REDIS_HOST=redis
       - PYTHONDONTWRITEBYTECODE=1
     volumes:
       - ./src:/src
@@ -24,3 +26,8 @@ services:
       - POSTGRES_PASSWORD=abc123
     ports:
       - "54321:5432"
+
+  redis:
+    image: redis:alpine
+    ports:
+      - "6379:6379"

requirements.txt

@@ -4,6 +4,7 @@ flask==1.1.2
 isort==5.7.0
 psycopg2-binary==2.8.6
 pytest==6.2.2
+redis==3.5.3
 requests==2.25.1
 sqlalchemy==1.3.23
 tenacity==6.3.1

src/allocation/adapters/event_publisher.py

@@ -0,0 +1,14 @@
+import json
+import logging
+from dataclasses import asdict
+
+from allocation import config
+from redis import Redis
+
+logger = logging.getLogger(__name__)
+redis_client = Redis(**config.get_redis_host_and_port())
+
+
+def publish(channel, event):
+    logger.debug(f"Publishing channel: {channel}, event: {event}")
+    redis_client.publish(channel, json.dumps(asdict(event)))

src/allocation/config.py

@@ -13,3 +13,9 @@ def get_api_url():
     host = os.environ.get("API_HOST", "localhost")
     port = 5005 if host == "localhost" else 80
     return f"http://{host}:{port}"
+
+
+def get_redis_host_and_port():
+    host = os.environ.get("REDIS_HOST", "localhost")
+    port = 6379 if host == "localhost" else 6379
+    return dict(host=host, port=port)

src/allocation/domain/events.py

@@ -5,6 +5,14 @@ class Event:
     pass
 
 
+@dataclass
+class Allocated(Event):
+    orderid: str
+    sku: str
+    qty: int
+    batchref: str
+
+
 @dataclass
 class OutOfStock(Event):
     sku: str

src/allocation/domain/model.py

@@ -3,7 +3,7 @@
 from typing import List, NewType, Optional, Set
 
 from allocation.domain.commands import Allocate
-from allocation.domain.events import Event, OutOfStock
+from allocation.domain.events import Allocated, Event, OutOfStock
 
 Reference = NewType("Reference", str)
 Sku = NewType("Sku", str)
@@ -102,6 +102,9 @@ def allocate(self, line: OrderLine):
             return None
 
         batch.allocate(line)
+        self.events.append(
+            Allocated(line.orderid, line.sku, line.qty, batch.reference)
+        )
 
         return batch.reference
 

src/allocation/entrypoints/event_consumer.py

@@ -0,0 +1,26 @@
+import json
+import logging
+
+from allocation import config
+from allocation.adapters import orm
+from allocation.domain import commands
+from allocation.service_layer import messagebus, unit_of_work
+from redis import Redis
+
+logger = logging.getLevelName(__name__)
+redis_client = Redis(**config.get_redis_host_and_port())
+
+
+def main():
+    orm.start_mappers()
+    pubsub = redis_client.pubsub(ignore_subscribe_messages=True)
+    pubsub.subscribe("change_batch_quantity")
+
+    for message in pubsub.listen():
+        logger.debug(f"Received message: {message}")
+
+
+def handle_change_batch_quantity(message):
+    data = json.loads(message["data"])
+    command = commands.ChangeBatchQuantity(data["reference"], data["qty"])
+    messagebus.handle(message=command, uow=unit_of_work.SqlAlchemyUnitOfWork())

src/allocation/service_layer/handlers.py

@@ -1,4 +1,4 @@
-from allocation.adapters import email
+from allocation.adapters import email, event_publisher
 from allocation.domain import commands, events, model
 from allocation.service_layer import unit_of_work
 
@@ -65,6 +65,12 @@ def deallocate(
         uow.commit()
 
 
+def publish_allocated_event(
+    message: events.Allocated, uow: unit_of_work.AbstractUnitOfWork
+):
+    event_publisher.publish("line_allocated", message)
+
+
 def send_out_of_stock_notification(
     message: events.OutOfStock, uow: unit_of_work.AbstractUnitOfWork
 ):

src/allocation/service_layer/messagebus.py

@@ -61,6 +61,7 @@ def handle_command(
 
 
 EVENT_HANDLERS = {
+    events.Allocated: [handlers.publish_allocated_event],
     events.OutOfStock: [handlers.send_out_of_stock_notification],
 }