Application 7 - Vertical slices

This example demonstrates how to do event sourcing with the “vertical slice architecture” advocated by the event modelling community.

Add product to shop

class AddProductToShop(Command):
    product_id: UUID
    name: str
    description: str
    price: Decimal

    def handle(self, events: DomainEvents) -> DomainEvents:
        if len(events):
            raise ProductAlreadyInShopError
        return (
            AddedProductToShop(
                originator_id=self.product_id,
                originator_version=1,
                name=self.name,
                description=self.description,
                price=self.price,
            ),
        )

    def execute(self) -> int | None:
        return put_events(self.handle(get_events(self.product_id)))

class TestAddProductToShop(TestCase):
    def test_add_product_to_shop(self) -> None:
        product_id = uuid4()

        product_events: tuple[DomainEvent, ...] = ()

        cmd = AddProductToShop(
            product_id=product_id,
            name="Coffee",
            description="A very nice coffee",
            price=Decimal("5.99"),
        )

        new_events = cmd.handle(product_events)
        self.assertEqual(len(new_events), 1)
        self.assertIsInstance(new_events[0], AddedProductToShop)
        new_event = cast(AddedProductToShop, new_events[0])
        self.assertEqual(new_event.originator_id, product_id)
        self.assertEqual(new_event.originator_version, 1)
        self.assertEqual(new_event.name, "Coffee")
        self.assertEqual(new_event.description, "A very nice coffee")
        self.assertEqual(new_event.price, Decimal("5.99"))

    def test_already_added_product_to_shop(self) -> None:
        product_id = uuid4()

        product_events: tuple[DomainEvent, ...] = (
            AddedProductToShop(
                originator_id=product_id,
                originator_version=1,
                name="Tea",
                description="A very nice tea",
                price=Decimal("5.99"),
            ),
        )

        cmd = AddProductToShop(
            product_id=product_id,
            name="Coffee",
            description="A very nice coffee",
            price=Decimal("3.99"),
        )

        with self.assertRaises(ProductAlreadyInShopError):
            cmd.handle(product_events)

Adjust product inventory

class AdjustProductInventory(Command):
    product_id: UUID
    adjustment: int

    def handle(self, events: DomainEvents) -> DomainEvents:
        if not events:
            raise ProductNotFoundInShopError
        return (
            AdjustedProductInventory(
                originator_id=self.product_id,
                originator_version=len(events) + 1,
                adjustment=self.adjustment,
            ),
        )

    def execute(self) -> int | None:
        return put_events(self.handle(get_events(self.product_id)))

class TestAdjustProductInventory(unittest.TestCase):
    def test_adjust_inventory(self) -> None:
        product_id = uuid4()
        cmd = AdjustProductInventory(
            product_id=product_id,
            adjustment=2,
        )
        product_events = (
            AddedProductToShop(
                originator_id=product_id,
                originator_version=1,
                name="Tea",
                description="A very nice tea",
                price=Decimal("3.99"),
            ),
        )
        new_events = cmd.handle(product_events)
        assert len(new_events) == 1
        self.assertIsInstance(new_events[0], AdjustedProductInventory)
        new_event = cast(AdjustedProductInventory, new_events[0])
        self.assertEqual(new_event.originator_id, product_id)
        self.assertEqual(new_event.originator_version, 2)
        self.assertEqual(new_event.adjustment, 2)

    def test_adjust_inventory_product_not_found(self) -> None:
        product_id = uuid4()
        cmd = AdjustProductInventory(
            product_id=product_id,
            adjustment=2,
        )
        product_events = ()

        with self.assertRaises(ProductNotFoundInShopError):
            cmd.handle(product_events)

List products in shop

class ListProductsInShop(Query):
    @staticmethod
    def projection(events: DomainEvents) -> Sequence[ProductDetails]:
        products: dict[UUID, ProductDetails] = {}
        for event in events:
            if isinstance(event, AddedProductToShop):
                products[event.originator_id] = ProductDetails(
                    id=event.originator_id,
                    name=event.name,
                    description=event.description,
                    price=event.price,
                )
            elif isinstance(event, AdjustedProductInventory):
                product = products[event.originator_id]
                products[event.originator_id] = ProductDetails(
                    id=event.originator_id,
                    name=product.name,
                    description=product.description,
                    price=product.price,
                    inventory=product.inventory + event.adjustment,
                )
        return tuple(products.values())

    def execute(self) -> Sequence[ProductDetails]:
        # TODO: Make this a materialised view.
        return self.projection(
            get_all_events(
                topics=(
                    get_topic(AddedProductToShop),
                    get_topic(AdjustedProductInventory),
                )
            )
        )

class ProductDetails(Immutable):
    id: UUID
    name: str
    description: str
    price: Decimal
    inventory: int = 0

class TestListProductsInShop(unittest.TestCase):
    def test_list_products_one_product(self) -> None:
        products = ListProductsInShop.projection(())
        self.assertEqual(len(products), 0)

        product_id1 = uuid4()
        events: tuple[DomainEvent, ...] = (
            AddedProductToShop(
                originator_id=product_id1,
                originator_version=1,
                name="Coffee",
                description="A very nice coffee",
                price=Decimal("5.99"),
            ),
        )
        products = ListProductsInShop.projection(events)
        self.assertEqual(len(products), 1)
        self.assertEqual(products[0].id, product_id1)
        self.assertEqual(products[0].name, "Coffee")
        self.assertEqual(products[0].description, "A very nice coffee")
        self.assertEqual(products[0].price, Decimal("5.99"))
        self.assertEqual(products[0].inventory, 0)

    def test_list_products_two_products_also_item_added_to_cart(self) -> None:

        product_id1 = uuid4()
        product_id2 = uuid4()
        events = (
            AddedProductToShop(
                originator_id=product_id1,
                originator_version=1,
                name="Coffee",
                description="A very nice coffee",
                price=Decimal("5.99"),
            ),
            AddedItemToCart(
                originator_id=uuid4(),
                originator_version=1,
                product_id=uuid4(),
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            AddedProductToShop(
                originator_id=product_id2,
                originator_version=1,
                name="Tea",
                description="A very nice tea",
                price=Decimal("3.99"),
            ),
        )
        products = ListProductsInShop.projection(events)
        self.assertEqual(len(products), 2)
        self.assertEqual(products[0].id, product_id1)
        self.assertEqual(products[0].name, "Coffee")
        self.assertEqual(products[0].description, "A very nice coffee")
        self.assertEqual(products[0].price, Decimal("5.99"))
        self.assertEqual(products[0].inventory, 0)
        self.assertEqual(products[1].id, product_id2)
        self.assertEqual(products[1].name, "Tea")
        self.assertEqual(products[1].description, "A very nice tea")
        self.assertEqual(products[1].price, Decimal("3.99"))
        self.assertEqual(products[1].inventory, 0)

    def test_list_products_two_products_also_adjusted_inventory(self) -> None:

        product_id1 = uuid4()
        product_id2 = uuid4()
        events = (
            AddedProductToShop(
                originator_id=product_id1,
                originator_version=1,
                name="Coffee",
                description="A very nice coffee",
                price=Decimal("5.99"),
            ),
            AdjustedProductInventory(
                originator_id=product_id1,
                originator_version=2,
                adjustment=2,
            ),
            AddedProductToShop(
                originator_id=product_id2,
                originator_version=1,
                name="Tea",
                description="A very nice tea",
                price=Decimal("3.99"),
            ),
        )
        products = ListProductsInShop.projection(events)
        self.assertEqual(len(products), 2)
        self.assertEqual(products[0].id, product_id1)
        self.assertEqual(products[0].name, "Coffee")
        self.assertEqual(products[0].description, "A very nice coffee")
        self.assertEqual(products[0].price, Decimal("5.99"))
        self.assertEqual(products[0].inventory, 2)
        self.assertEqual(products[1].id, product_id2)
        self.assertEqual(products[1].name, "Tea")
        self.assertEqual(products[1].description, "A very nice tea")
        self.assertEqual(products[1].price, Decimal("3.99"))
        self.assertEqual(products[1].inventory, 0)

Get cart items

class GetCartItems(Query):
    cart_id: UUID

    @staticmethod
    def projection(events: DomainEvents) -> Sequence[CartItem]:
        cart_items: list[CartItem] = []
        for event in events:
            if isinstance(event, AddedItemToCart):
                cart_items.append(
                    CartItem(
                        product_id=event.product_id,
                        name=event.name,
                        description=event.description,
                        price=event.price,
                    )
                )
            elif isinstance(event, RemovedItemFromCart):
                for i, cart_item in enumerate(cart_items):
                    if cart_item.product_id == event.product_id:
                        cart_items.pop(i)
                        break
            elif isinstance(event, ClearedCart):
                cart_items.clear()
        return tuple(cart_items)

    def execute(self) -> Sequence[CartItem]:
        return self.projection(get_events(self.cart_id))

class CartItem(Immutable):
    product_id: UUID
    name: str
    description: str
    price: Decimal

class TestGetCartItems(TestCase):
    def test_cart_empty(self) -> None:
        cart_events: tuple[DomainEvent, ...] = ()
        cart_items = GetCartItems.projection(cart_events)
        self.assertEqual(len(cart_items), 0)

    def test_cart_added_item(self) -> None:
        cart_id: UUID = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="name",
                description="description",
                price=Decimal("1"),
            ),
        )
        cart_items = GetCartItems.projection(cart_events)
        self.assertEqual(len(cart_items), 1)
        self.assertEqual(cart_items[0].product_id, product_id)
        self.assertEqual(cart_items[0].name, "name")
        self.assertEqual(cart_items[0].description, "description")
        self.assertEqual(cart_items[0].price, Decimal("1"))

    def test_cart_added_item_and_removed_item(self) -> None:
        cart_id: UUID = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="name",
                description="description",
                price=Decimal("1"),
            ),
            RemovedItemFromCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
            ),
        )
        cart_items = GetCartItems.projection(cart_events)
        self.assertEqual(len(cart_items), 0)

    def test_cart_added_two_items_and_removed_two_items(self) -> None:
        cart_id: UUID = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="name",
                description="description",
                price=Decimal("1"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
                name="name",
                description="description",
                price=Decimal("1"),
            ),
            RemovedItemFromCart(
                originator_id=cart_id,
                originator_version=3,
                product_id=product_id,
            ),
            RemovedItemFromCart(
                originator_id=cart_id,
                originator_version=4,
                product_id=product_id,
            ),
        )
        cart_items = GetCartItems.projection(cart_events)
        self.assertEqual(len(cart_items), 0)

    def test_cart_added_item_and_cleared_cart(self) -> None:
        cart_id: UUID = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="name",
                description="description",
                price=Decimal("1"),
            ),
            ClearedCart(
                originator_id=cart_id,
                originator_version=2,
            ),
        )
        cart_items = GetCartItems.projection(cart_events)
        self.assertEqual(len(cart_items), 0)

Add item to cart

class AddItemToCart(Command):
    cart_id: UUID
    product_id: UUID
    description: str
    price: Decimal
    name: str

    def handle(self, events: DomainEvents) -> DomainEvents:
        product_ids = []
        is_submitted = False
        for event in events:
            if isinstance(event, AddedItemToCart):
                product_ids.append(event.product_id)
            elif isinstance(event, RemovedItemFromCart):
                product_ids.remove(event.product_id)
            elif isinstance(event, ClearedCart):
                product_ids.clear()
            elif isinstance(event, SubmittedCart):
                is_submitted = True

        if is_submitted:
            raise CartAlreadySubmittedError

        if len(product_ids) >= 3:
            raise CartFullError

        return (
            AddedItemToCart(
                originator_id=self.cart_id,
                originator_version=len(events) + 1,
                product_id=self.product_id,
                name=self.name,
                description=self.description,
                price=self.price,
            ),
        )

    def execute(self) -> int | None:
        return put_events(self.handle(get_events(self.cart_id)))

class TestAddItemToCart(unittest.TestCase):
    def test_add_item_to_empty_cart(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cmd = AddItemToCart(
            cart_id=cart_id,
            product_id=product_id,
            name="Coffee",
            description="A very special coffee",
            price=Decimal("5.99"),
        )
        cart_events: tuple[DomainEvent, ...] = ()
        new_events = cmd.handle(cart_events)
        self.assertEqual(1, len(new_events))
        self.assertIsInstance(new_events[0], AddedItemToCart)
        new_event = cast(AddedItemToCart, new_events[0])
        self.assertEqual(cmd.product_id, new_event.product_id)

    def test_add_item_to_full_cart(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=3,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
        )

        cmd = AddItemToCart(
            cart_id=cart_id,
            product_id=uuid4(),
            name="Coffee",
            description="A very special coffee",
            price=Decimal("5.99"),
        )

        with self.assertRaises(CartFullError):
            cmd.handle(cart_events)

    def test_add_item_to_cart_after_adding_three_and_removing_one(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=3,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            RemovedItemFromCart(
                originator_id=cart_id,
                originator_version=4,
                product_id=product_id,
            ),
        )

        cmd = AddItemToCart(
            cart_id=cart_id,
            product_id=uuid4(),
            name="Coffee",
            description="A very special coffee",
            price=Decimal("5.99"),
        )

        cmd.handle(cart_events)

    def test_add_item_to_cart_after_adding_three_and_clearing_cart(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=3,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("5.99"),
            ),
            ClearedCart(
                originator_id=cart_id,
                originator_version=4,
            ),
        )

        cmd = AddItemToCart(
            cart_id=cart_id,
            product_id=uuid4(),
            name="Coffee",
            description="A very special coffee",
            price=Decimal("5.99"),
        )

        cmd.handle(cart_events)

    def test_add_item_after_submitted_cart(self) -> None:
        cart_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            SubmittedCart(
                originator_id=cart_id,
                originator_version=1,
            ),
        )

        cmd = AddItemToCart(
            cart_id=cart_id,
            product_id=uuid4(),
            name="Coffee",
            description="A very special coffee",
            price=Decimal("5.99"),
        )

        with self.assertRaises(CartAlreadySubmittedError):
            cmd.handle(cart_events)

Remove item from cart

class RemoveItemFromCart(Command):
    cart_id: UUID
    product_id: UUID

    def handle(self, events: DomainEvents) -> DomainEvents:
        product_ids = []
        is_submitted = False

        for event in events:
            if isinstance(event, AddedItemToCart):
                product_ids.append(event.product_id)
            elif isinstance(event, RemovedItemFromCart):
                product_ids.remove(event.product_id)
            elif isinstance(event, ClearedCart):
                product_ids.clear()
            elif isinstance(event, SubmittedCart):
                is_submitted = True

        if is_submitted:
            raise CartAlreadySubmittedError

        if self.product_id not in product_ids:
            raise ProductNotInCartError
        return (
            RemovedItemFromCart(
                originator_id=self.cart_id,
                originator_version=len(events) + 1,
                product_id=self.product_id,
            ),
        )

    def execute(self) -> int | None:
        return put_events(self.handle(get_events(self.cart_id)))

class TestRemoveItemFromCart(unittest.TestCase):
    def test_remove_item_from_empty_cart(self) -> None:
        cart_events = ()
        cmd = RemoveItemFromCart(
            cart_id=uuid4(),
            product_id=uuid4(),
        )
        with self.assertRaises(ProductNotInCartError):
            cmd.handle(cart_events)

    def test_remove_item_from_cart_after_item_added(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("1"),
            ),
        )
        cmd = RemoveItemFromCart(
            cart_id=cart_id,
            product_id=product_id,
        )
        new_events = cmd.handle(cart_events)
        self.assertEqual(len(new_events), 1)
        self.assertIsInstance(new_events[0], RemovedItemFromCart)

    def test_remove_item_from_cart_after_item_removed(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("1"),
            ),
            RemovedItemFromCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
            ),
        )
        cmd = RemoveItemFromCart(
            cart_id=cart_id,
            product_id=product_id,
        )
        with self.assertRaises(ProductNotInCartError):
            cmd.handle(cart_events)

    def test_remove_item_from_cart_after_cart_cleared(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("1"),
            ),
            ClearedCart(
                originator_id=cart_id,
                originator_version=2,
            ),
        )
        cmd = RemoveItemFromCart(
            cart_id=cart_id,
            product_id=product_id,
        )
        with self.assertRaises(ProductNotInCartError):
            cmd.handle(cart_events)

    def test_remove_item_from_cart_after_submitted_cart(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            SubmittedCart(
                originator_id=cart_id,
                originator_version=1,
            ),
        )
        cmd = RemoveItemFromCart(
            cart_id=cart_id,
            product_id=product_id,
        )
        with self.assertRaises(CartAlreadySubmittedError):
            cmd.handle(cart_events)

        with self.assertRaises(CartAlreadySubmittedError):
            cmd.handle(cart_events)

Clear cart

class ClearCart(Command):
    cart_id: UUID

    def handle(self, events: DomainEvents) -> DomainEvents:
        is_submitted = False
        for event in events:
            if isinstance(event, SubmittedCart):
                is_submitted = True

        if is_submitted:
            raise CartAlreadySubmittedError

        return (
            ClearedCart(
                originator_id=self.cart_id,
                originator_version=len(events) + 1,
            ),
        )

    def execute(self) -> int | None:
        return put_events(self.handle(get_events(self.cart_id)))

class TestClearCart(unittest.TestCase):
    def test_clear_cart(self) -> None:
        cart_id = uuid4()
        cmd = ClearCart(
            cart_id=cart_id,
        )
        events: tuple[DomainEvent, ...] = ()
        new_events = cmd.handle(events)
        self.assertEqual(len(new_events), 1)
        self.assertIsInstance(new_events[0], ClearedCart)
        new_event = cast(ClearedCart, new_events[0])
        self.assertEqual(new_event.originator_id, cart_id)
        self.assertEqual(new_event.originator_version, 1)

    def test_clear_cart_after_submitted_cart(self) -> None:
        cart_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            SubmittedCart(
                originator_id=cart_id,
                originator_version=1,
            ),
        )

        cmd = ClearCart(
            cart_id=cart_id,
        )

        with self.assertRaises(CartAlreadySubmittedError):
            cmd.handle(cart_events)

Submit cart

class SubmitCart(Command):
    cart_id: UUID

    def handle(self, events: DomainEvents) -> DomainEvents:
        requested_products: dict[UUID, int] = defaultdict(int)
        is_submitted = False

        for event in events:
            if isinstance(event, AddedItemToCart):
                requested_products[event.product_id] += 1
            elif isinstance(event, RemovedItemFromCart):
                requested_products[event.product_id] -= 1
            elif isinstance(event, ClearedCart):
                requested_products.clear()
            elif isinstance(event, SubmittedCart):
                is_submitted = True

        if is_submitted:
            raise CartAlreadySubmittedError

        # Check inventory.
        for product_id, requested_amount in requested_products.items():
            current_inventory = 0
            for product_event in get_events(product_id):
                if isinstance(product_event, AdjustedProductInventory):
                    current_inventory += product_event.adjustment
            if current_inventory < requested_amount:
                msg = f"Insufficient inventory for product with ID {product_id}"
                raise InsufficientInventoryError(msg)

        return (
            SubmittedCart(
                originator_id=self.cart_id,
                originator_version=len(events) + 1,
            ),
        )

    def execute(self) -> int | None:
        return put_events(self.handle(get_events(self.cart_id)))

class TestSubmitCart(unittest.TestCase):
    def setUp(self) -> None:
        reset_application()

    def test_submit_cart_sufficient_inventory_empty_cart(self) -> None:
        cart_id = uuid4()
        cmd = SubmitCart(
            cart_id=cart_id,
        )
        cart_events: tuple[DomainEvent, ...] = ()
        new_events = cmd.handle(cart_events)
        self.assertEqual(len(new_events), 1)
        self.assertIsInstance(new_events[0], SubmittedCart)
        new_event = cast(SubmittedCart, new_events[0])
        self.assertEqual(new_event.originator_id, cart_id)
        self.assertEqual(new_event.originator_version, 1)

    def test_submit_cart_sufficient_inventory_after_item_removed(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()
        cmd = SubmitCart(
            cart_id=cart_id,
        )
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("1"),
            ),
            RemovedItemFromCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
            ),
        )
        new_events = cmd.handle(cart_events)
        self.assertEqual(len(new_events), 1)
        self.assertIsInstance(new_events[0], SubmittedCart)
        new_event = cast(SubmittedCart, new_events[0])
        self.assertEqual(new_event.originator_id, cart_id)
        self.assertEqual(new_event.originator_version, 3)

    def test_submit_cart_sufficient_inventory_after_cart_cleared(self) -> None:
        cart_id = uuid4()
        cmd = SubmitCart(
            cart_id=cart_id,
        )
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=uuid4(),
                name="",
                description="",
                price=Decimal("1"),
            ),
            ClearedCart(
                originator_id=cart_id,
                originator_version=2,
            ),
        )
        new_events = cmd.handle(cart_events)
        self.assertEqual(len(new_events), 1)
        self.assertIsInstance(new_events[0], SubmittedCart)
        new_event = cast(SubmittedCart, new_events[0])
        self.assertEqual(new_event.originator_id, cart_id)
        self.assertEqual(new_event.originator_version, 3)

    def test_submit_cart_insufficient_inventory_after_item_added(self) -> None:
        cart_id = uuid4()
        cmd = SubmitCart(
            cart_id=cart_id,
        )
        product_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("100"),
            ),
        )
        with self.assertRaises(InsufficientInventoryError):
            cmd.handle(cart_events)

    def test_submit_cart_sufficient_inventory_after_item_added(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()

        AddProductToShop(
            product_id=product_id,
            name="",
            description="",
            price=Decimal("100"),
        ).execute()
        AdjustProductInventory(
            product_id=product_id,
            adjustment=1,
        ).execute()

        cmd = SubmitCart(
            cart_id=cart_id,
        )
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("100"),
            ),
        )
        new_events = cmd.handle(cart_events)
        self.assertEqual(len(new_events), 1)
        self.assertIsInstance(new_events[0], SubmittedCart)
        new_event = cast(SubmittedCart, new_events[0])
        self.assertEqual(new_event.originator_id, cart_id)
        self.assertEqual(new_event.originator_version, 2)

    def test_submit_cart_insufficient_inventory_after_item_added_twice(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()

        AddProductToShop(
            product_id=product_id,
            name="",
            description="",
            price=Decimal("100"),
        ).execute()
        AdjustProductInventory(
            product_id=product_id,
            adjustment=1,
        ).execute()

        cmd = SubmitCart(cart_id=cart_id)

        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("100"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("100"),
            ),
        )
        with self.assertRaises(InsufficientInventoryError):
            cmd.handle(cart_events)

    def test_submit_cart_sufficient_inventory_after_item_added_twice(self) -> None:
        cart_id = uuid4()
        product_id = uuid4()

        AddProductToShop(
            product_id=product_id,
            name="",
            description="",
            price=Decimal("100"),
        ).execute()
        AdjustProductInventory(
            product_id=product_id,
            adjustment=2,
        ).execute()

        cmd = SubmitCart(
            cart_id=cart_id,
        )
        cart_events: tuple[DomainEvent, ...] = (
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=1,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("100"),
            ),
            AddedItemToCart(
                originator_id=cart_id,
                originator_version=2,
                product_id=product_id,
                name="",
                description="",
                price=Decimal("100"),
            ),
        )
        new_events = cmd.handle(cart_events)
        self.assertEqual(len(new_events), 1)
        self.assertIsInstance(new_events[0], SubmittedCart)
        new_event = cast(SubmittedCart, new_events[0])
        self.assertEqual(new_event.originator_id, cart_id)
        self.assertEqual(new_event.originator_version, 3)

    def test_submit_cart_after_submitted_cart(self) -> None:
        cart_id = uuid4()
        cart_events: tuple[DomainEvent, ...] = (
            SubmittedCart(
                originator_id=cart_id,
                originator_version=1,
            ),
        )

        cmd = SubmitCart(
            cart_id=cart_id,
        )

        with self.assertRaises(CartAlreadySubmittedError):
            cmd.handle(cart_events)

Events

from __future__ import annotations

from collections.abc import Sequence
from decimal import Decimal  # noqa: TC003
from typing import TYPE_CHECKING
from uuid import UUID  # noqa: TC003

from examples.aggregate7.immutablemodel import Immutable

if TYPE_CHECKING:
    from typing_extensions import TypeAlias


class DomainEvent(Immutable):
    originator_id: UUID
    originator_version: int


DomainEvents: TypeAlias = Sequence[DomainEvent]


class AddedProductToShop(DomainEvent):
    name: str
    description: str
    price: Decimal


class AdjustedProductInventory(DomainEvent):
    adjustment: int


class AddedItemToCart(DomainEvent):
    product_id: UUID
    name: str
    description: str
    price: Decimal


class RemovedItemFromCart(DomainEvent):
    product_id: UUID


class ClearedCart(DomainEvent):
    pass


class SubmittedCart(DomainEvent):
    pass

Exceptions

class ProductNotFoundInShopError(Exception):
    pass


class ProductAlreadyInShopError(Exception):
    pass


class CartFullError(Exception):
    pass


class ProductNotInCartError(Exception):
    pass


class InsufficientInventoryError(Exception):
    pass


class CartAlreadySubmittedError(Exception):
    pass

Common code

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, cast

from examples.aggregate7.immutablemodel import Immutable
from examples.aggregate7.orjsonpydantic import PydanticApplication
from examples.shopvertical.events import DomainEvents

if TYPE_CHECKING:
    from collections.abc import Sequence
    from uuid import UUID


class Command(Immutable, ABC):
    @abstractmethod
    def handle(self, events: DomainEvents) -> DomainEvents:
        pass  # pragma: no cover

    @abstractmethod
    def execute(self) -> int | None:
        pass  # pragma: no cover


class Query(Immutable, ABC):
    @abstractmethod
    def execute(self) -> Any:
        pass  # pragma: no cover


class _Globals:
    app = PydanticApplication()


def reset_application() -> None:
    _Globals.app = PydanticApplication()


def get_events(originator_id: UUID) -> DomainEvents:
    return cast(DomainEvents, tuple(_Globals.app.events.get(originator_id)))


def put_events(events: DomainEvents) -> int | None:
    recordings = _Globals.app.events.put(events)
    return recordings[-1].notification.id if recordings else None


def get_all_events(topics: Sequence[str] = ()) -> DomainEvents:
    return cast(
        DomainEvents,
        tuple(
            map(
                _Globals.app.mapper.to_domain_event,
                _Globals.app.recorder.select_notifications(
                    start=None,
                    limit=1000000,
                    topics=topics,
                ),
            )
        ),
    )

Integration test

class TestShop(TestCase):
    def setUp(self) -> None:
        reset_application()

    def test(self) -> None:
        product_id1 = uuid4()
        product_id2 = uuid4()
        product_id3 = uuid4()
        product_id4 = uuid4()
        product_id5 = uuid4()

        # Add products to shop.
        AddProductToShop(
            product_id=product_id1,
            name="Coffee",
            description="A very nice coffee",
            price=Decimal("5.99"),
        ).execute()

        with self.assertRaises(ProductAlreadyInShopError):
            AddProductToShop(
                product_id=product_id1,
                name="Coffee",
                description="A very nice coffee",
                price=Decimal("5.99"),
            ).execute()

        AddProductToShop(
            product_id=product_id2,
            name="Tea",
            description="A very nice tea",
            price=Decimal("3.99"),
        ).execute()

        # Adjust product inventory.
        AdjustProductInventory(
            product_id=product_id1,
            adjustment=3,
        ).execute()

        # Product not in shop.
        with self.assertRaises(ProductNotFoundInShopError):
            AdjustProductInventory(
                product_id=product_id3,
                adjustment=1,
            ).execute()

        AddProductToShop(
            product_id=product_id3,
            name="Sugar",
            description="A very nice sugar",
            price=Decimal("2.99"),
        ).execute()

        AddProductToShop(
            product_id=product_id4,
            name="Milk",
            description="A very nice milk",
            price=Decimal("1.99"),
        ).execute()

        # List products.
        products = ListProductsInShop().execute()
        self.assertEqual(len(products), 4)
        self.assertEqual(products[0].id, product_id1)
        self.assertEqual(products[0].name, "Coffee")
        self.assertEqual(products[0].description, "A very nice coffee")
        self.assertEqual(products[0].price, Decimal("5.99"))
        self.assertEqual(products[0].inventory, 3)
        self.assertEqual(products[1].id, product_id2)
        self.assertEqual(products[1].name, "Tea")
        self.assertEqual(products[1].description, "A very nice tea")
        self.assertEqual(products[1].price, Decimal("3.99"))
        self.assertEqual(products[1].inventory, 0)
        self.assertEqual(products[2].id, product_id3)
        self.assertEqual(products[2].name, "Sugar")
        self.assertEqual(products[2].description, "A very nice sugar")
        self.assertEqual(products[2].price, Decimal("2.99"))
        self.assertEqual(products[2].inventory, 0)
        self.assertEqual(products[3].id, product_id4)
        self.assertEqual(products[3].name, "Milk")
        self.assertEqual(products[3].description, "A very nice milk")
        self.assertEqual(products[3].price, Decimal("1.99"))
        self.assertEqual(products[3].inventory, 0)

        # Get cart items - should be 0.
        cart_id = uuid4()
        cart_items = GetCartItems(cart_id=cart_id).execute()
        self.assertEqual(len(cart_items), 0)

        # Add item to cart.
        AddItemToCart(
            cart_id=cart_id,
            product_id=product_id1,
            name="Coffee",
            description="A very nice coffee",
            price=Decimal("5.99"),
        ).execute()

        # Get cart items - should be 1.
        cart_items = GetCartItems(cart_id=cart_id).execute()
        self.assertEqual(len(cart_items), 1)

        # Check everything is getting serialised and deserialised correctly.
        self.assertEqual(cart_items[0].product_id, product_id1)
        self.assertEqual(cart_items[0].name, "Coffee")
        self.assertEqual(cart_items[0].description, "A very nice coffee")
        self.assertEqual(cart_items[0].price, Decimal("5.99"))

        # Clear cart.
        ClearCart(cart_id=cart_id).execute()

        # Get cart items - should be 0.
        cart_items = GetCartItems(cart_id=cart_id).execute()
        self.assertEqual(len(cart_items), 0)

        # Add item to cart.
        AddItemToCart(
            cart_id=cart_id,
            product_id=product_id1,
            name="Coffee",
            description="A very nice coffee",
            price=Decimal("5.99"),
        ).execute()

        # Add item to cart.
        AddItemToCart(
            cart_id=cart_id,
            product_id=product_id2,
            name="Tea",
            description="A very nice tea",
            price=Decimal("3.99"),
        ).execute()

        # Get cart items - should be 2.
        cart_items = GetCartItems(cart_id=cart_id).execute()
        self.assertEqual(len(cart_items), 2)
        self.assertEqual(cart_items[0].product_id, product_id1)
        self.assertEqual(cart_items[1].product_id, product_id2)

        # Add item to cart.
        AddItemToCart(
            cart_id=cart_id,
            product_id=product_id3,
            name="Sugar",
            description="A very nice sugar",
            price=Decimal("2.99"),
        ).execute()

        # Get cart items - should be 3.
        cart_items = GetCartItems(cart_id=cart_id).execute()
        self.assertEqual(len(cart_items), 3)
        self.assertEqual(cart_items[0].product_id, product_id1)
        self.assertEqual(cart_items[1].product_id, product_id2)
        self.assertEqual(cart_items[2].product_id, product_id3)

        # Cart full error.
        with self.assertRaises(CartFullError):
            AddItemToCart(
                cart_id=cart_id,
                product_id=product_id4,
                name="Milk",
                description="A very nice milk",
                price=Decimal("1.99"),
            ).execute()

        # Get cart items - should be 3.
        cart_items = GetCartItems(cart_id=cart_id).execute()
        self.assertEqual(len(cart_items), 3)
        self.assertEqual(cart_items[0].product_id, product_id1)
        self.assertEqual(cart_items[1].product_id, product_id2)
        self.assertEqual(cart_items[2].product_id, product_id3)

        # Remove item from cart.
        RemoveItemFromCart(
            cart_id=cart_id,
            product_id=product_id2,
        ).execute()

        # Get cart items - should be 2.
        cart_items = GetCartItems(cart_id=cart_id).execute()
        self.assertEqual(len(cart_items), 2)
        self.assertEqual(cart_items[0].product_id, product_id1)
        self.assertEqual(cart_items[1].product_id, product_id3)

        # Product not in cart error.
        with self.assertRaises(ProductNotInCartError):
            RemoveItemFromCart(
                cart_id=cart_id,
                product_id=product_id2,
            ).execute()

        # Add item to cart.
        AddItemToCart(
            cart_id=cart_id,
            product_id=product_id5,
            name="Spoon",
            description="A very nice spoon",
            price=Decimal("5.99"),
        ).execute()

        # Get cart items - should be 3.
        cart_items = GetCartItems(cart_id=cart_id).execute()
        self.assertEqual(len(cart_items), 3)
        self.assertEqual(cart_items[0].product_id, product_id1)
        self.assertEqual(cart_items[1].product_id, product_id3)
        self.assertEqual(cart_items[2].product_id, product_id5)

        # Insufficient inventory error.
        with self.assertRaises(InsufficientInventoryError):
            SubmitCart(cart_id=cart_id).execute()

        # Adjust product inventory.
        AdjustProductInventory(
            product_id=product_id3,
            adjustment=3,
        ).execute()

        # Insufficient inventory error.
        with self.assertRaises(InsufficientInventoryError):
            SubmitCart(cart_id=cart_id).execute()

        # Add item to shop.
        AddProductToShop(
            product_id=product_id5,
            name="Spoon",
            description="A very nice spoon",
            price=Decimal("0.99"),
        ).execute()

        # Adjust product inventory.
        AdjustProductInventory(
            product_id=product_id5,
            adjustment=3,
        ).execute()

        # Submit cart.
        SubmitCart(cart_id=cart_id).execute()

        # Cart already submitted.
        with self.assertRaises(CartAlreadySubmittedError):
            AddItemToCart(
                cart_id=cart_id,
                product_id=product_id4,
                name="Milk",
                description="A very nice milk",
                price=Decimal("1.99"),
            ).execute()

        with self.assertRaises(CartAlreadySubmittedError):
            RemoveItemFromCart(
                cart_id=cart_id,
                product_id=product_id1,
            ).execute()

        with self.assertRaises(CartAlreadySubmittedError):
            ClearCart(
                cart_id=cart_id,
            ).execute()

        with self.assertRaises(CartAlreadySubmittedError):
            SubmitCart(
                cart_id=cart_id,
            ).execute()

Code reference

class examples.shopvertical.slices.add_product_to_shop.cmd.AddProductToShop(*, product_id: UUID, name: str, description: str, price: Decimal)

Bases: Command

product_id: UUID

name: str

description: str

price: Decimal

handle(events: Sequence[DomainEvent]) → Sequence[DomainEvent]

execute() → int | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'description': FieldInfo(annotation=str, required=True), 'name': FieldInfo(annotation=str, required=True), 'price': FieldInfo(annotation=Decimal, required=True), 'product_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.adjust_product_inventory.cmd.AdjustProductInventory(*, product_id: UUID, adjustment: int)

Bases: Command

product_id: UUID

adjustment: int

handle(events: Sequence[DomainEvent]) → Sequence[DomainEvent]

execute() → int | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'adjustment': FieldInfo(annotation=int, required=True), 'product_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.list_products_in_shop.query.ProductDetails(*, id: UUID, name: str, description: str, price: Decimal, inventory: int = 0)

Bases: Immutable

id: UUID

name: str

description: str

price: Decimal

inventory: int

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'description': FieldInfo(annotation=str, required=True), 'id': FieldInfo(annotation=UUID, required=True), 'inventory': FieldInfo(annotation=int, required=False, default=0), 'name': FieldInfo(annotation=str, required=True), 'price': FieldInfo(annotation=Decimal, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.list_products_in_shop.query.ListProductsInShop

Bases: Query

static projection(events: Sequence[DomainEvent]) → Sequence[ProductDetails]

execute() → Sequence[ProductDetails]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.get_cart_items.query.CartItem(*, product_id: UUID, name: str, description: str, price: Decimal)

Bases: Immutable

product_id: UUID

name: str

description: str

price: Decimal

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'description': FieldInfo(annotation=str, required=True), 'name': FieldInfo(annotation=str, required=True), 'price': FieldInfo(annotation=Decimal, required=True), 'product_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.get_cart_items.query.GetCartItems(*, cart_id: UUID)

Bases: Query

cart_id: UUID

static projection(events: Sequence[DomainEvent]) → Sequence[CartItem]

execute() → Sequence[CartItem]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'cart_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.add_item_to_cart.cmd.AddItemToCart(*, cart_id: UUID, product_id: UUID, description: str, price: Decimal, name: str)

Bases: Command

cart_id: UUID

product_id: UUID

description: str

price: Decimal

name: str

handle(events: Sequence[DomainEvent]) → Sequence[DomainEvent]

execute() → int | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'cart_id': FieldInfo(annotation=UUID, required=True), 'description': FieldInfo(annotation=str, required=True), 'name': FieldInfo(annotation=str, required=True), 'price': FieldInfo(annotation=Decimal, required=True), 'product_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.remove_item_from_cart.cmd.RemoveItemFromCart(*, cart_id: UUID, product_id: UUID)

Bases: Command

cart_id: UUID

product_id: UUID

handle(events: Sequence[DomainEvent]) → Sequence[DomainEvent]

execute() → int | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'cart_id': FieldInfo(annotation=UUID, required=True), 'product_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.clear_cart.cmd.ClearCart(*, cart_id: UUID)

Bases: Command

cart_id: UUID

handle(events: Sequence[DomainEvent]) → Sequence[DomainEvent]

execute() → int | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'cart_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.slices.submit_cart.cmd.SubmitCart(*, cart_id: UUID)

Bases: Command

cart_id: UUID

handle(events: Sequence[DomainEvent]) → Sequence[DomainEvent]

execute() → int | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'cart_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.events.DomainEvent(*, originator_id: UUID, originator_version: int)

Bases: Immutable

originator_id: UUID

originator_version: int

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'originator_id': FieldInfo(annotation=UUID, required=True), 'originator_version': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.events.AddedProductToShop(*, originator_id: UUID, originator_version: int, name: str, description: str, price: Decimal)

Bases: DomainEvent

name: str

description: str

price: Decimal

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'description': FieldInfo(annotation=str, required=True), 'name': FieldInfo(annotation=str, required=True), 'originator_id': FieldInfo(annotation=UUID, required=True), 'originator_version': FieldInfo(annotation=int, required=True), 'price': FieldInfo(annotation=Decimal, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.events.AdjustedProductInventory(*, originator_id: UUID, originator_version: int, adjustment: int)

Bases: DomainEvent

adjustment: int

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'adjustment': FieldInfo(annotation=int, required=True), 'originator_id': FieldInfo(annotation=UUID, required=True), 'originator_version': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.events.AddedItemToCart(*, originator_id: UUID, originator_version: int, product_id: UUID, name: str, description: str, price: Decimal)

Bases: DomainEvent

product_id: UUID

name: str

description: str

price: Decimal

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'description': FieldInfo(annotation=str, required=True), 'name': FieldInfo(annotation=str, required=True), 'originator_id': FieldInfo(annotation=UUID, required=True), 'originator_version': FieldInfo(annotation=int, required=True), 'price': FieldInfo(annotation=Decimal, required=True), 'product_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.events.RemovedItemFromCart(*, originator_id: UUID, originator_version: int, product_id: UUID)

Bases: DomainEvent

product_id: UUID

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'originator_id': FieldInfo(annotation=UUID, required=True), 'originator_version': FieldInfo(annotation=int, required=True), 'product_id': FieldInfo(annotation=UUID, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.events.ClearedCart(*, originator_id: UUID, originator_version: int)

Bases: DomainEvent

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'originator_id': FieldInfo(annotation=UUID, required=True), 'originator_version': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

originator_id: UUID

originator_version: int

class examples.shopvertical.events.SubmittedCart(*, originator_id: UUID, originator_version: int)

Bases: DomainEvent

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'originator_id': FieldInfo(annotation=UUID, required=True), 'originator_version': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

originator_id: UUID

originator_version: int

exception examples.shopvertical.exceptions.ProductNotFoundInShopError

Bases: Exception

exception examples.shopvertical.exceptions.ProductAlreadyInShopError

Bases: Exception

exception examples.shopvertical.exceptions.CartFullError

Bases: Exception

exception examples.shopvertical.exceptions.ProductNotInCartError

Bases: Exception

exception examples.shopvertical.exceptions.InsufficientInventoryError

Bases: Exception

exception examples.shopvertical.exceptions.CartAlreadySubmittedError

Bases: Exception

class examples.shopvertical.common.Command

Bases: Immutable, ABC

abstractmethod handle(events: Sequence[DomainEvent]) → Sequence[DomainEvent]

abstractmethod execute() → int | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class examples.shopvertical.common.Query

Bases: Immutable, ABC

abstractmethod execute() → Any

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

examples.shopvertical.common.reset_application() → None

examples.shopvertical.common.get_events(originator_id: UUID) → DomainEvents

examples.shopvertical.common.put_events(events: Sequence[DomainEvent]) → int | None

examples.shopvertical.common.get_all_events(topics: Sequence[str] = ()) → DomainEvents