Revert "view now users redis, tweak tests+app. [...]" [chapter_12_cqrs_ends]

This reverts commit 6dd38db04ea7c637f2d2510e7f587ac2009dd2ac.

Revert "new helpers to update read model ..."

This reverts commit dbaebdf249d952f2c3915526fd4d1bc6fe3cd18f.

Revert "handlers talk to redis ..."

This reverts commit 00658e2181de3e118579fa0a0da9b7ebb9e081f9.

Author: hjwp

src/allocation/adapters/redis_eventpublisher.py

@@ -14,11 +14,3 @@
 def publish(channel, event: events.Event):
     logging.debug('publishing: channel=%s, event=%s', channel, event)
     r.publish(channel, json.dumps(asdict(event)))
-
-
-def update_readmodel(orderid, sku, batchref):
-    r.hset(orderid, sku, batchref)
-
-
-def get_readmodel(orderid):
-    return r.hgetall(orderid)

src/allocation/entrypoints/flask_app.py

@@ -40,7 +40,8 @@ def allocate_endpoint():
 
 @app.route("/allocations/<orderid>", methods=['GET'])
 def allocations_view_endpoint(orderid):
-    result = views.allocations(orderid)
+    uow = unit_of_work.SqlAlchemyUnitOfWork()
+    result = views.allocations(orderid, uow)
     if not result:
         return 'not found', 404
     return jsonify(result), 200

src/allocation/service_layer/handlers.py

@@ -72,8 +72,24 @@ def publish_allocated_event(
     redis_eventpublisher.publish('line_allocated', event)
 
 
-def add_allocation_to_read_model(event: events.Allocated, _):
-    redis_eventpublisher.update_readmodel(event.orderid, event.sku, event.batchref)
+def add_allocation_to_read_model(
+        event: events.Allocated, uow: unit_of_work.SqlAlchemyUnitOfWork,
+):
+    with uow:
+        uow.session.execute(
+            'INSERT INTO allocations_view (orderid, sku, batchref)'
+            ' VALUES (:orderid, :sku, :batchref)',
+            dict(orderid=event.orderid, sku=event.sku, batchref=event.batchref)
+        )
+        uow.commit()
 
-def remove_allocation_from_read_model(event: events.Deallocated, _):
-    redis_eventpublisher.update_readmodel(event.orderid, event.sku, None)
+def remove_allocation_from_read_model(
+        event: events.Deallocated, uow: unit_of_work.SqlAlchemyUnitOfWork,
+):
+    with uow:
+        uow.session.execute(
+            'DELETE FROM allocations_view '
+            ' WHERE orderid = :orderid AND sku = :sku',
+            dict(orderid=event.orderid, sku=event.sku)
+        )
+        uow.commit()

src/allocation/views.py

@@ -1,9 +1,9 @@
-from allocation.adapters import redis_eventpublisher
 from allocation.service_layer import unit_of_work
 
-def allocations(orderid):
-    batches = redis_eventpublisher.get_readmodel(orderid)
-    return [
-        {'batchref': b.decode(), 'sku': s.decode()}
-        for s, b in batches.items()
-    ]
+def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork):
+    with uow:
+        results = list(uow.session.execute(
+            'SELECT sku, batchref FROM allocations_view WHERE orderid = :orderid',
+            dict(orderid=orderid)
+        ))
+    return [dict(r) for r in results]

tests/integration/test_views.py

@@ -1,20 +1,8 @@
 from datetime import date
-import pytest
-import redis
-from allocation import config, views
+from allocation import views
 from allocation.domain import commands
 from allocation.service_layer import messagebus, unit_of_work
 
-@pytest.fixture
-def cleanup_redis():
-    r = redis.Redis(**config.get_redis_host_and_port())
-    yield
-    for k in r.keys():
-        print('cleaning up redis key', k)
-        r.delete(k)
-
-pytestmark = pytest.mark.usefixtures('cleanup_redis')
-
 
 def test_allocations_view(sqlite_session_factory):
     uow = unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory)
@@ -27,7 +15,7 @@ def test_allocations_view(sqlite_session_factory):
     messagebus.handle(commands.Allocate('otherorder', 'sku1', 30), uow)
     messagebus.handle(commands.Allocate('otherorder', 'sku2', 10), uow)
 
-    assert views.allocations('order1') == [
+    assert views.allocations('order1', uow) == [
         {'sku': 'sku1', 'batchref': 'sku1batch'},
         {'sku': 'sku2', 'batchref': 'sku2batch'},
     ]
@@ -40,6 +28,6 @@ def test_deallocation(sqlite_session_factory):
     messagebus.handle(commands.Allocate('o1', 'sku1', 40), uow)
     messagebus.handle(commands.ChangeBatchQuantity('b1', 10), uow)
 
-    assert views.allocations('o1') == [
+    assert views.allocations('o1', uow) == [
         {'sku': 'sku1', 'batchref': 'b2'},
     ]