From 3c3bdc4db80c4a6f77f5e37b8693127b5f671750 Mon Sep 17 00:00:00 2001 From: Stanis Trendelenburg Date: Wed, 27 May 2026 20:38:11 +0200 Subject: [PATCH 1/4] Update python docker image to 3.13 --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 1a3c97655..976b0c8ec 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.9-slim-buster +FROM python:3.13-slim-trixie # RUN apt install gcc libpq (no longer needed bc we use psycopg2-binary) From 4befb6fb761ea34687173b2a7a77d2afa01c4dad Mon Sep 17 00:00:00 2001 From: Stanis Trendelenburg Date: Wed, 27 May 2026 20:44:21 +0200 Subject: [PATCH 2/4] Update sqlalchemy to 2.x Just making sure the tests pass for now --- requirements.txt | 2 +- src/allocation/adapters/orm.py | 5 +++- src/allocation/service_layer/handlers.py | 21 ++++++++------ src/allocation/views.py | 11 +++++--- tests/integration/test_uow.py | 35 ++++++++++++++---------- 5 files changed, 46 insertions(+), 28 deletions(-) diff --git a/requirements.txt b/requirements.txt index 789a24fea..882cb3524 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,5 @@ # app -sqlalchemy<2 +sqlalchemy flask psycopg2-binary redis diff --git a/src/allocation/adapters/orm.py b/src/allocation/adapters/orm.py index 81d704ca0..76acb194b 100644 --- a/src/allocation/adapters/orm.py +++ b/src/allocation/adapters/orm.py @@ -9,7 +9,7 @@ ForeignKey, event, ) -from sqlalchemy.orm import mapper, relationship +from sqlalchemy.orm import registry, relationship from allocation.domain import model @@ -17,6 +17,8 @@ metadata = MetaData() +mapper_registry = registry() + order_lines = Table( "order_lines", metadata, @@ -62,6 +64,7 @@ def start_mappers(): logger.info("Starting mappers") + mapper = mapper_registry.map_imperatively lines_mapper = mapper(model.OrderLine, order_lines) batches_mapper = mapper( model.Batch, diff --git a/src/allocation/service_layer/handlers.py b/src/allocation/service_layer/handlers.py index 2d7aa8d45..1fa16f907 100644 --- a/src/allocation/service_layer/handlers.py +++ b/src/allocation/service_layer/handlers.py @@ -2,6 +2,7 @@ from __future__ import annotations from dataclasses import asdict from typing import List, Dict, Callable, Type, TYPE_CHECKING +from sqlalchemy import text from allocation.domain import commands, events, model from allocation.domain.model import OrderLine @@ -83,10 +84,12 @@ def add_allocation_to_read_model( ): with uow: uow.session.execute( - """ - INSERT INTO allocations_view (orderid, sku, batchref) - VALUES (:orderid, :sku, :batchref) - """, + text( + """ + INSERT INTO allocations_view (orderid, sku, batchref) + VALUES (:orderid, :sku, :batchref) + """ + ), dict(orderid=event.orderid, sku=event.sku, batchref=event.batchref), ) uow.commit() @@ -98,10 +101,12 @@ def remove_allocation_from_read_model( ): with uow: uow.session.execute( - """ - DELETE FROM allocations_view - WHERE orderid = :orderid AND sku = :sku - """, + text( + """ + DELETE FROM allocations_view + WHERE orderid = :orderid AND sku = :sku + """ + ), dict(orderid=event.orderid, sku=event.sku), ) uow.commit() diff --git a/src/allocation/views.py b/src/allocation/views.py index a952887f4..79b90fc01 100644 --- a/src/allocation/views.py +++ b/src/allocation/views.py @@ -1,12 +1,15 @@ +from sqlalchemy import text from allocation.service_layer import unit_of_work def allocations(orderid: str, uow: unit_of_work.SqlAlchemyUnitOfWork): with uow: results = uow.session.execute( - """ - SELECT sku, batchref FROM allocations_view WHERE orderid = :orderid - """, + text( + """ + SELECT sku, batchref FROM allocations_view WHERE orderid = :orderid + """ + ), dict(orderid=orderid), - ) + ).mappings() return [dict(r) for r in results] diff --git a/tests/integration/test_uow.py b/tests/integration/test_uow.py index 61f3aae55..fdf58fea8 100644 --- a/tests/integration/test_uow.py +++ b/tests/integration/test_uow.py @@ -5,6 +5,7 @@ from typing import List from unittest.mock import Mock import pytest +from sqlalchemy import text from allocation.domain import model from allocation.service_layer import unit_of_work from ..random_refs import random_sku, random_batchref, random_orderid @@ -14,24 +15,28 @@ def insert_batch(session, ref, sku, qty, eta, product_version=1): session.execute( - "INSERT INTO products (sku, version_number) VALUES (:sku, :version)", + text("INSERT INTO products (sku, version_number) VALUES (:sku, :version)"), dict(sku=sku, version=product_version), ) session.execute( - "INSERT INTO batches (reference, sku, _purchased_quantity, eta)" - " VALUES (:ref, :sku, :qty, :eta)", + text( + "INSERT INTO batches (reference, sku, _purchased_quantity, eta)" + " VALUES (:ref, :sku, :qty, :eta)" + ), dict(ref=ref, sku=sku, qty=qty, eta=eta), ) def get_allocated_batch_ref(session, orderid, sku): [[orderlineid]] = session.execute( - "SELECT id FROM order_lines WHERE orderid=:orderid AND sku=:sku", + text("SELECT id FROM order_lines WHERE orderid=:orderid AND sku=:sku"), dict(orderid=orderid, sku=sku), ) [[batchref]] = session.execute( - "SELECT b.reference FROM allocations JOIN batches AS b ON batch_id = b.id" - " WHERE orderline_id=:orderlineid", + text( + "SELECT b.reference FROM allocations JOIN batches AS b ON batch_id = b.id" + " WHERE orderline_id=:orderlineid" + ), dict(orderlineid=orderlineid), ) return batchref @@ -59,7 +64,7 @@ def test_rolls_back_uncommitted_work_by_default(sqlite_session_factory): insert_batch(uow.session, "batch1", "MEDIUM-PLINTH", 100, None) new_session = sqlite_session_factory() - rows = list(new_session.execute('SELECT * FROM "batches"')) + rows = list(new_session.execute(text('SELECT * FROM "batches"'))) assert rows == [] @@ -74,7 +79,7 @@ class MyException(Exception): raise MyException() new_session = sqlite_session_factory() - rows = list(new_session.execute('SELECT * FROM "batches"')) + rows = list(new_session.execute(text('SELECT * FROM "batches"'))) assert rows == [] @@ -113,7 +118,7 @@ def test_concurrent_updates_to_version_are_not_allowed(postgres_session_factory) thread2.join() [[version]] = session.execute( - "SELECT version_number FROM products WHERE sku=:sku", + text("SELECT version_number FROM products WHERE sku=:sku"), dict(sku=sku), ) assert version == 2 @@ -121,12 +126,14 @@ def test_concurrent_updates_to_version_are_not_allowed(postgres_session_factory) assert "could not serialize access due to concurrent update" in str(exception) orders = session.execute( - "SELECT orderid FROM allocations" - " JOIN batches ON allocations.batch_id = batches.id" - " JOIN order_lines ON allocations.orderline_id = order_lines.id" - " WHERE order_lines.sku=:sku", + text( + "SELECT orderid FROM allocations" + " JOIN batches ON allocations.batch_id = batches.id" + " JOIN order_lines ON allocations.orderline_id = order_lines.id" + " WHERE order_lines.sku=:sku" + ), dict(sku=sku), ) assert orders.rowcount == 1 with unit_of_work.SqlAlchemyUnitOfWork(postgres_session_factory) as uow: - uow.session.execute("select 1") + uow.session.execute(text("select 1")) From bbf1ecf5e6bee1e960c3c4179a0123b0e6f383de Mon Sep 17 00:00:00 2001 From: Stanis Trendelenburg Date: Wed, 27 May 2026 22:39:18 +0200 Subject: [PATCH 3/4] Separate ORM from domain model classes - Drop SQL schema, classical/imperative mapping - Translate between domain and ORM objects explicitly - Implement value objects using immutable (frozen) dataclasses --- src/allocation/adapters/orm.py | 154 ++++++++++++----------- src/allocation/adapters/repository.py | 42 +++---- src/allocation/bootstrap.py | 6 +- src/allocation/domain/model.py | 16 ++- src/allocation/service_layer/handlers.py | 4 +- tests/conftest.py | 11 +- tests/integration/test_email.py | 3 - tests/integration/test_repository.py | 3 - tests/integration/test_uow.py | 20 ++- tests/integration/test_views.py | 3 - tests/unit/test_handlers.py | 2 - 11 files changed, 129 insertions(+), 135 deletions(-) diff --git a/src/allocation/adapters/orm.py b/src/allocation/adapters/orm.py index 76acb194b..b4d0b506a 100644 --- a/src/allocation/adapters/orm.py +++ b/src/allocation/adapters/orm.py @@ -1,57 +1,94 @@ -import logging -from sqlalchemy import ( - Table, - MetaData, - Column, - Integer, - String, - Date, - ForeignKey, - event, -) -from sqlalchemy.orm import registry, relationship +from datetime import date +from typing import Self +from sqlalchemy import Column, ForeignKey, String, Table +from sqlalchemy.orm import registry as create_registry, relationship, Mapped, mapped_column from allocation.domain import model -logger = logging.getLogger(__name__) +registry = create_registry() -metadata = MetaData() +metadata = registry.metadata -mapper_registry = registry() -order_lines = Table( - "order_lines", - metadata, - Column("id", Integer, primary_key=True, autoincrement=True), - Column("sku", String(255)), - Column("qty", Integer, nullable=False), - Column("orderid", String(255)), -) +@registry.mapped_as_dataclass +class OrderLine: + __tablename__ = "order_lines" -products = Table( - "products", - metadata, - Column("sku", String(255), primary_key=True), - Column("version_number", Integer, nullable=False, server_default="0"), -) + batch_ref: Mapped[str] = mapped_column(ForeignKey("batches.reference"), primary_key=True) + _idx: Mapped[int] = mapped_column(primary_key=True) + sku: Mapped[str] = mapped_column(String(255)) + qty: Mapped[int] + orderid: Mapped[str] = mapped_column(String(255)) -batches = Table( - "batches", - metadata, - Column("id", Integer, primary_key=True, autoincrement=True), - Column("reference", String(255)), - Column("sku", ForeignKey("products.sku")), - Column("_purchased_quantity", Integer, nullable=False), - Column("eta", Date, nullable=True), -) -allocations = Table( - "allocations", - metadata, - Column("id", Integer, primary_key=True, autoincrement=True), - Column("orderline_id", ForeignKey("order_lines.id")), - Column("batch_id", ForeignKey("batches.id")), -) +@registry.mapped_as_dataclass +class Product: + __tablename__ = "products" + + sku: Mapped[str] = mapped_column(String(255), primary_key=True) + version_number: Mapped[int] = mapped_column(server_default="0") + + batches: Mapped[list["Batch"]] = relationship(cascade="all,delete-orphan") + + @classmethod + def from_domain(cls, product: model.Product) -> Self: + return cls( + sku=product.sku, + version_number=product.version_number, + batches=[Batch.from_domain(b) for b in product.batches], + ) + + def to_domain(self) -> model.Product: + return model.Product( + sku=self.sku, + batches=[b.to_domain() for b in self.batches], + version_number=self.version_number, + ) + + +@registry.mapped_as_dataclass +class Batch: + __tablename__ = "batches" + + reference: Mapped[str] = mapped_column(String(255), primary_key=True) + sku: Mapped[str] = mapped_column(ForeignKey("products.sku")) + purchased_quantity: Mapped[int] + eta: Mapped[date | None] + + allocations: Mapped[list[OrderLine]] = relationship(cascade="all,delete-orphan") + + @classmethod + def from_domain(cls, batch: model.Batch) -> Self: + return cls( + reference=batch.reference, + sku=batch.sku, + eta=batch.eta, + purchased_quantity=batch._purchased_quantity, + allocations=[ + OrderLine( + batch_ref=batch.reference, + _idx=idx, + sku=ol.sku, + qty=ol.qty, + orderid=ol.orderid, + ) + for idx, ol in enumerate(sorted(batch._allocations)) + ], + ) + + def to_domain(self) -> model.Batch: + batch = model.Batch( + ref=self.reference, + sku=self.sku, + eta=self.eta, + qty=self.purchased_quantity, + ) + batch._allocations = set( + model.OrderLine(orderid=ol.orderid, sku=ol.sku, qty=ol.qty) + for ol in self.allocations + ) + return batch + allocations_view = Table( "allocations_view", @@ -60,30 +97,3 @@ Column("sku", String(255)), Column("batchref", String(255)), ) - - -def start_mappers(): - logger.info("Starting mappers") - mapper = mapper_registry.map_imperatively - lines_mapper = mapper(model.OrderLine, order_lines) - batches_mapper = mapper( - model.Batch, - batches, - properties={ - "_allocations": relationship( - lines_mapper, - secondary=allocations, - collection_class=set, - ) - }, - ) - mapper( - model.Product, - products, - properties={"batches": relationship(batches_mapper)}, - ) - - -@event.listens_for(model.Product, "load") -def receive_load(product, _): - product.events = [] diff --git a/src/allocation/adapters/repository.py b/src/allocation/adapters/repository.py index 15a6b0ae3..62de88029 100644 --- a/src/allocation/adapters/repository.py +++ b/src/allocation/adapters/repository.py @@ -1,5 +1,7 @@ import abc from typing import Set +from sqlalchemy import select +from sqlalchemy.orm import Session from allocation.adapters import orm from allocation.domain import model @@ -8,52 +10,48 @@ class AbstractRepository(abc.ABC): def __init__(self): self.seen = set() # type: Set[model.Product] - def add(self, product: model.Product): + def add(self, product: model.Product) -> None: self._add(product) self.seen.add(product) - def get(self, sku) -> model.Product: + def get(self, sku) -> model.Product | None: product = self._get(sku) if product: self.seen.add(product) return product - def get_by_batchref(self, batchref) -> model.Product: + def get_by_batchref(self, batchref: str) -> model.Product | None: product = self._get_by_batchref(batchref) if product: self.seen.add(product) return product @abc.abstractmethod - def _add(self, product: model.Product): + def _add(self, product: model.Product) -> None: raise NotImplementedError @abc.abstractmethod - def _get(self, sku) -> model.Product: + def _get(self, sku) -> model.Product | None: raise NotImplementedError @abc.abstractmethod - def _get_by_batchref(self, batchref) -> model.Product: + def _get_by_batchref(self, batchref: str) -> model.Product | None: raise NotImplementedError class SqlAlchemyRepository(AbstractRepository): - def __init__(self, session): + def __init__(self, session: Session): super().__init__() self.session = session - def _add(self, product): - self.session.add(product) - - def _get(self, sku): - return self.session.query(model.Product).filter_by(sku=sku).first() - - def _get_by_batchref(self, batchref): - return ( - self.session.query(model.Product) - .join(model.Batch) - .filter( - orm.batches.c.reference == batchref, - ) - .first() - ) + def _add(self, product: model.Product) -> None: + self.session.merge(orm.Product.from_domain(product)) + + def _get(self, sku) -> model.Product | None: + if p := self.session.get(orm.Product, sku): + return p.to_domain() + + def _get_by_batchref(self, batchref: str) -> model.Product | None: + query = select(orm.Product).join(orm.Batch).where(orm.Batch.reference == batchref) + if p := self.session.scalars(query).first(): + return p.to_domain() diff --git a/src/allocation/bootstrap.py b/src/allocation/bootstrap.py index 22112a064..11766f6c6 100644 --- a/src/allocation/bootstrap.py +++ b/src/allocation/bootstrap.py @@ -1,6 +1,6 @@ import inspect from typing import Callable -from allocation.adapters import orm, redis_eventpublisher +from allocation.adapters import redis_eventpublisher from allocation.adapters.notifications import ( AbstractNotifications, EmailNotifications, @@ -9,7 +9,6 @@ def bootstrap( - start_orm: bool = True, uow: unit_of_work.AbstractUnitOfWork = unit_of_work.SqlAlchemyUnitOfWork(), notifications: AbstractNotifications = None, publish: Callable = redis_eventpublisher.publish, @@ -18,9 +17,6 @@ def bootstrap( if notifications is None: notifications = EmailNotifications() - if start_orm: - orm.start_mappers() - dependencies = {"uow": uow, "notifications": notifications, "publish": publish} injected_event_handlers = { event_type: [ diff --git a/src/allocation/domain/model.py b/src/allocation/domain/model.py index dd4ac7823..a24a43fb5 100644 --- a/src/allocation/domain/model.py +++ b/src/allocation/domain/model.py @@ -7,11 +7,15 @@ class Product: def __init__(self, sku: str, batches: List[Batch], version_number: int = 0): - self.sku = sku + self._sku = sku self.batches = batches self.version_number = version_number self.events = [] # type: List[events.Event] + @property + def sku(self): + return self._sku + def allocate(self, line: OrderLine) -> str: try: batch = next(b for b in sorted(self.batches) if b.can_allocate(line)) @@ -37,8 +41,16 @@ def change_batch_quantity(self, ref: str, qty: int): line = batch.deallocate_one() self.events.append(events.Deallocated(line.orderid, line.sku, line.qty)) + def __eq__(self, other: object) -> bool: + if not isinstance(other, Product): + return NotImplemented + return self.sku == other.sku + + def __hash__(self) -> int: + return hash(self.sku) + -@dataclass(unsafe_hash=True) +@dataclass(frozen=True, order=True) class OrderLine: orderid: str sku: str diff --git a/src/allocation/service_layer/handlers.py b/src/allocation/service_layer/handlers.py index 1fa16f907..ec51aa4f3 100644 --- a/src/allocation/service_layer/handlers.py +++ b/src/allocation/service_layer/handlers.py @@ -23,8 +23,8 @@ def add_batch( product = uow.products.get(sku=cmd.sku) if product is None: product = model.Product(cmd.sku, batches=[]) - uow.products.add(product) product.batches.append(model.Batch(cmd.ref, cmd.sku, cmd.qty, cmd.eta)) + uow.products.add(product) uow.commit() @@ -38,6 +38,7 @@ def allocate( if product is None: raise InvalidSku(f"Invalid sku {line.sku}") product.allocate(line) + uow.products.add(product) uow.commit() @@ -55,6 +56,7 @@ def change_batch_quantity( with uow: product = uow.products.get_by_batchref(batchref=cmd.ref) product.change_batch_quantity(ref=cmd.ref, qty=cmd.qty) + uow.products.add(product) uow.commit() diff --git a/tests/conftest.py b/tests/conftest.py index f91f93a07..3af283694 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,10 +8,10 @@ import redis import requests from sqlalchemy import create_engine -from sqlalchemy.orm import sessionmaker, clear_mappers +from sqlalchemy.orm import sessionmaker from tenacity import retry, stop_after_delay -from allocation.adapters.orm import metadata, start_mappers +from allocation.adapters.orm import metadata from allocation import config pytest.register_assert_rewrite("tests.e2e.api_client") @@ -29,13 +29,6 @@ def sqlite_session_factory(in_memory_sqlite_db): yield sessionmaker(bind=in_memory_sqlite_db) -@pytest.fixture -def mappers(): - start_mappers() - yield - clear_mappers() - - @retry(stop=stop_after_delay(10)) def wait_for_postgres_to_come_up(engine): return engine.connect() diff --git a/tests/integration/test_email.py b/tests/integration/test_email.py index 4aade37b6..324c75a57 100644 --- a/tests/integration/test_email.py +++ b/tests/integration/test_email.py @@ -1,7 +1,6 @@ # pylint: disable=redefined-outer-name import pytest import requests -from sqlalchemy.orm import clear_mappers from allocation import bootstrap, config from allocation.domain import commands from allocation.adapters import notifications @@ -12,13 +11,11 @@ @pytest.fixture def bus(sqlite_session_factory): bus = bootstrap.bootstrap( - start_orm=True, uow=unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory), notifications=notifications.EmailNotifications(), publish=lambda *args: None, ) yield bus - clear_mappers() def get_email_from_mailhog(sku): diff --git a/tests/integration/test_repository.py b/tests/integration/test_repository.py index 7961be2a3..245b26f23 100644 --- a/tests/integration/test_repository.py +++ b/tests/integration/test_repository.py @@ -1,9 +1,6 @@ -import pytest from allocation.adapters import repository from allocation.domain import model -pytestmark = pytest.mark.usefixtures("mappers") - def test_get_by_batchref(sqlite_session_factory): session = sqlite_session_factory() diff --git a/tests/integration/test_uow.py b/tests/integration/test_uow.py index fdf58fea8..76e4f7e1d 100644 --- a/tests/integration/test_uow.py +++ b/tests/integration/test_uow.py @@ -10,8 +10,6 @@ from allocation.service_layer import unit_of_work from ..random_refs import random_sku, random_batchref, random_orderid -pytestmark = pytest.mark.usefixtures("mappers") - def insert_batch(session, ref, sku, qty, eta, product_version=1): session.execute( @@ -20,7 +18,7 @@ def insert_batch(session, ref, sku, qty, eta, product_version=1): ) session.execute( text( - "INSERT INTO batches (reference, sku, _purchased_quantity, eta)" + "INSERT INTO batches (reference, sku, purchased_quantity, eta)" " VALUES (:ref, :sku, :qty, :eta)" ), dict(ref=ref, sku=sku, qty=qty, eta=eta), @@ -28,16 +26,12 @@ def insert_batch(session, ref, sku, qty, eta, product_version=1): def get_allocated_batch_ref(session, orderid, sku): - [[orderlineid]] = session.execute( - text("SELECT id FROM order_lines WHERE orderid=:orderid AND sku=:sku"), - dict(orderid=orderid, sku=sku), - ) [[batchref]] = session.execute( text( - "SELECT b.reference FROM allocations JOIN batches AS b ON batch_id = b.id" - " WHERE orderline_id=:orderlineid" + "SELECT batch_ref FROM order_lines" + " WHERE orderid=:orderid AND sku=:sku" ), - dict(orderlineid=orderlineid), + dict(orderid=orderid, sku=sku), ) return batchref @@ -52,6 +46,7 @@ def test_uow_can_retrieve_a_batch_and_allocate_to_it(sqlite_session_factory): product = uow.products.get(sku="HIPSTER-WORKBENCH") line = model.OrderLine("o1", "HIPSTER-WORKBENCH", 10) product.allocate(line) + uow.products.add(product) uow.commit() batchref = get_allocated_batch_ref(session, "o1", "HIPSTER-WORKBENCH") @@ -89,6 +84,7 @@ def try_to_allocate(orderid, sku, exceptions, session_factory): with unit_of_work.SqlAlchemyUnitOfWork(session_factory) as uow: product = uow.products.get(sku=sku) product.allocate(line) + uow.products.add(product) time.sleep(0.2) uow.commit() except Exception as e: # pylint: disable=broad-except @@ -127,9 +123,7 @@ def test_concurrent_updates_to_version_are_not_allowed(postgres_session_factory) orders = session.execute( text( - "SELECT orderid FROM allocations" - " JOIN batches ON allocations.batch_id = batches.id" - " JOIN order_lines ON allocations.orderline_id = order_lines.id" + "SELECT orderid FROM order_lines" " WHERE order_lines.sku=:sku" ), dict(sku=sku), diff --git a/tests/integration/test_views.py b/tests/integration/test_views.py index ccd5d542e..f9b253c71 100644 --- a/tests/integration/test_views.py +++ b/tests/integration/test_views.py @@ -1,6 +1,5 @@ # pylint: disable=redefined-outer-name from datetime import date -from sqlalchemy.orm import clear_mappers from unittest import mock import pytest from allocation import bootstrap, views @@ -13,13 +12,11 @@ @pytest.fixture def sqlite_bus(sqlite_session_factory): bus = bootstrap.bootstrap( - start_orm=True, uow=unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory), notifications=mock.Mock(), publish=lambda *args: None, ) yield bus - clear_mappers() def test_allocations_view(sqlite_bus): diff --git a/tests/unit/test_handlers.py b/tests/unit/test_handlers.py index f12185403..af778a86b 100644 --- a/tests/unit/test_handlers.py +++ b/tests/unit/test_handlers.py @@ -51,7 +51,6 @@ def send(self, destination, message): def bootstrap_test_app(): return bootstrap.bootstrap( - start_orm=False, uow=FakeUnitOfWork(), notifications=FakeNotifications(), publish=lambda *args: None, @@ -98,7 +97,6 @@ def test_commits(self): def test_sends_email_on_out_of_stock_error(self): fake_notifs = FakeNotifications() bus = bootstrap.bootstrap( - start_orm=False, uow=FakeUnitOfWork(), notifications=fake_notifs, publish=lambda *args: None, From ea2c62918be7e82dc5f0ee17ab19eb67a73f9caa Mon Sep 17 00:00:00 2001 From: Stanis Trendelenburg Date: Thu, 28 May 2026 18:12:38 +0200 Subject: [PATCH 4/4] Automatically persist changes to aggregates --- src/allocation/adapters/repository.py | 7 +++++++ src/allocation/service_layer/handlers.py | 4 +--- src/allocation/service_layer/unit_of_work.py | 1 + tests/integration/test_uow.py | 2 -- 4 files changed, 9 insertions(+), 5 deletions(-) diff --git a/src/allocation/adapters/repository.py b/src/allocation/adapters/repository.py index 62de88029..3c4b78b56 100644 --- a/src/allocation/adapters/repository.py +++ b/src/allocation/adapters/repository.py @@ -26,6 +26,9 @@ def get_by_batchref(self, batchref: str) -> model.Product | None: self.seen.add(product) return product + def commit(self) -> None: + pass + @abc.abstractmethod def _add(self, product: model.Product) -> None: raise NotImplementedError @@ -55,3 +58,7 @@ def _get_by_batchref(self, batchref: str) -> model.Product | None: query = select(orm.Product).join(orm.Batch).where(orm.Batch.reference == batchref) if p := self.session.scalars(query).first(): return p.to_domain() + + def commit(self) -> None: + for product in self.seen: + self._add(product) diff --git a/src/allocation/service_layer/handlers.py b/src/allocation/service_layer/handlers.py index ec51aa4f3..1fa16f907 100644 --- a/src/allocation/service_layer/handlers.py +++ b/src/allocation/service_layer/handlers.py @@ -23,8 +23,8 @@ def add_batch( product = uow.products.get(sku=cmd.sku) if product is None: product = model.Product(cmd.sku, batches=[]) + uow.products.add(product) product.batches.append(model.Batch(cmd.ref, cmd.sku, cmd.qty, cmd.eta)) - uow.products.add(product) uow.commit() @@ -38,7 +38,6 @@ def allocate( if product is None: raise InvalidSku(f"Invalid sku {line.sku}") product.allocate(line) - uow.products.add(product) uow.commit() @@ -56,7 +55,6 @@ def change_batch_quantity( with uow: product = uow.products.get_by_batchref(batchref=cmd.ref) product.change_batch_quantity(ref=cmd.ref, qty=cmd.qty) - uow.products.add(product) uow.commit() diff --git a/src/allocation/service_layer/unit_of_work.py b/src/allocation/service_layer/unit_of_work.py index 586ae027e..20585e123 100644 --- a/src/allocation/service_layer/unit_of_work.py +++ b/src/allocation/service_layer/unit_of_work.py @@ -20,6 +20,7 @@ def __exit__(self, *args): self.rollback() def commit(self): + self.products.commit() self._commit() def collect_new_events(self): diff --git a/tests/integration/test_uow.py b/tests/integration/test_uow.py index 76e4f7e1d..41fb0c7e3 100644 --- a/tests/integration/test_uow.py +++ b/tests/integration/test_uow.py @@ -46,7 +46,6 @@ def test_uow_can_retrieve_a_batch_and_allocate_to_it(sqlite_session_factory): product = uow.products.get(sku="HIPSTER-WORKBENCH") line = model.OrderLine("o1", "HIPSTER-WORKBENCH", 10) product.allocate(line) - uow.products.add(product) uow.commit() batchref = get_allocated_batch_ref(session, "o1", "HIPSTER-WORKBENCH") @@ -84,7 +83,6 @@ def try_to_allocate(orderid, sku, exceptions, session_factory): with unit_of_work.SqlAlchemyUnitOfWork(session_factory) as uow: product = uow.products.get(sku=sku) product.allocate(line) - uow.products.add(product) time.sleep(0.2) uow.commit() except Exception as e: # pylint: disable=broad-except