Application 6 - Shopping cart¶
This example suggests how a shopping cart might be implemented.
Application¶
class Shop(PydanticApplication):
def add_product_to_shop(
self, product_id: UUID, name: str, description: str, price: Decimal
) -> None:
try:
self.save(Product(product_id, name, description, price))
except IntegrityError:
raise ProductAlreadyInShopError from None
def adjust_product_inventory(self, product_id: UUID, adjustment: int) -> None:
try:
product: Product = self.repository.get(product_id)
except AggregateNotFoundError:
raise ProductNotFoundInShopError from None
else:
product.adjust_inventory(adjustment)
self.save(product)
def list_products_in_shop(self) -> Sequence[ProductDetails]:
# TODO: Make this a materialised view.
return tuple(
ProductDetails(
id=product.id,
name=product.name,
description=product.description,
price=product.price,
inventory=product.inventory,
)
for n in self.recorder.select_notifications(
start=None,
limit=1000000,
topics=[get_topic(Product.Created)],
)
if (
product := cast(
Product, self.repository.get(cast(UUID, n.originator_id))
)
)
)
def get_cart_items(self, cart_id: UUID) -> Sequence[CartItem]:
return tuple(self._get_cart(cart_id).items)
def add_item_to_cart(
self,
cart_id: UUID,
product_id: UUID,
name: str,
description: str,
price: Decimal,
) -> None:
cart = self._get_cart(cart_id)
cart.add_item(product_id, name, description, price)
self.save(cart)
def remove_item_from_cart(self, cart_id: UUID, product_id: UUID) -> None:
cart = self._get_cart(cart_id)
cart.remove_item(product_id)
self.save(cart)
def clear_cart(self, cart_id: UUID) -> None:
cart = self._get_cart(cart_id)
cart.clear()
self.save(cart)
def submit_cart(self, cart_id: UUID) -> None:
cart = self._get_cart(cart_id)
# Check inventory.
requested_products = Counter(i.product_id for i in cart.items)
for product_id, requested_amount in requested_products.items():
try:
product: Product = self.repository.get(product_id)
except AggregateNotFoundError:
current_inventory = 0
else:
current_inventory = product.inventory
if current_inventory < requested_amount:
msg = f"Insufficient inventory for product with ID {product_id}"
raise InsufficientInventoryError(msg)
cart.submit()
self.save(cart)
def _get_cart(self, cart_id: UUID) -> Cart:
try:
return self.repository.get(cart_id)
except AggregateNotFoundError:
return Cart(id=cart_id)
class ProductDetails(Immutable):
id: UUID
name: str
description: str
price: Decimal
inventory: int
Domain model¶
class Product(Aggregate):
def __init__(self, id: UUID, name: str, description: str, price: Decimal):
self._id = id
self.name = name
self.description = description
self.price = price
self.inventory = 0
class InventoryAdjusted(Aggregate.Event):
adjustment: int
@event(InventoryAdjusted)
def adjust_inventory(self, adjustment: int) -> None:
self.inventory += adjustment
class Cart(Aggregate):
def __init__(self, id: UUID):
self._id = id
self.items: list[CartItem] = []
self.is_submitted = False
class ItemAdded(Aggregate.Event):
product_id: UUID
name: str
description: str
price: Decimal
class ItemRemoved(Aggregate.Event):
product_id: UUID
class Cleared(Aggregate.Event):
pass
class Submitted(Aggregate.Event):
pass
@event(ItemAdded)
def add_item(
self, product_id: UUID, name: str, description: str, price: Decimal
) -> None:
if self.is_submitted:
raise CartAlreadySubmittedError
if len(self.items) >= 3:
raise CartFullError
self.items.append(
CartItem(
product_id=product_id,
name=name,
description=description,
price=price,
)
)
@event(ItemRemoved)
def remove_item(self, product_id: UUID) -> None:
if self.is_submitted:
raise CartAlreadySubmittedError
for i, item in enumerate(self.items):
if item.product_id == product_id:
self.items.pop(i)
break
else:
raise ProductNotInCartError
@event(Cleared)
def clear(self) -> None:
if self.is_submitted:
raise CartAlreadySubmittedError
self.items = []
@event(Submitted)
def submit(self) -> None:
if self.is_submitted:
raise CartAlreadySubmittedError
self.is_submitted = True
class CartItem(Immutable):
product_id: UUID
name: str
description: str
price: Decimal
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
Test¶
class TestShop(TestCase):
def test(self) -> None:
app = Shop()
product_id1: UUID = uuid4()
product_id2: UUID = uuid4()
product_id3: UUID = uuid4()
product_id4: UUID = uuid4()
product_id5: UUID = uuid4()
# Add products to shop.
app.add_product_to_shop(
product_id=product_id1,
name="Coffee",
description="A very nice coffee",
price=Decimal("5.99"),
)
with self.assertRaises(ProductAlreadyInShopError):
app.add_product_to_shop(
product_id=product_id1,
name="Coffee",
description="A very nice coffee",
price=Decimal("5.99"),
)
app.add_product_to_shop(
product_id=product_id2,
name="Tea",
description="A very nice tea",
price=Decimal("3.99"),
)
# Adjust product inventory.
app.adjust_product_inventory(
product_id=product_id1,
adjustment=3,
)
# Product not in shop.
with self.assertRaises(ProductNotFoundInShopError):
app.adjust_product_inventory(
product_id=product_id3,
adjustment=1,
)
app.add_product_to_shop(
product_id=product_id3,
name="Sugar",
description="A very nice sugar",
price=Decimal("2.99"),
)
app.add_product_to_shop(
product_id=product_id4,
name="Milk",
description="A very nice milk",
price=Decimal("1.99"),
)
# List products.
products = app.list_products_in_shop()
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 = app.get_cart_items(cart_id)
self.assertEqual(len(cart_items), 0)
# Add item to cart.
app.add_item_to_cart(
cart_id=cart_id,
product_id=product_id1,
name="Coffee",
description="A very nice coffee",
price=Decimal("5.99"),
)
# Get cart items - should be 1.
cart_items = app.get_cart_items(cart_id)
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.
app.clear_cart(cart_id)
# Get cart items - should be 0.
cart_items = app.get_cart_items(cart_id)
self.assertEqual(len(cart_items), 0)
# Add item to cart.
app.add_item_to_cart(
cart_id=cart_id,
product_id=product_id1,
name="Coffee",
description="A very nice coffee",
price=Decimal("5.99"),
)
# Add item to cart.
app.add_item_to_cart(
cart_id=cart_id,
product_id=product_id2,
name="Tea",
description="A very nice tea",
price=Decimal("3.99"),
)
# Get cart items - should be 2.
cart_items = app.get_cart_items(cart_id)
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.
app.add_item_to_cart(
cart_id=cart_id,
product_id=product_id3,
name="Sugar",
description="A very nice sugar",
price=Decimal("2.99"),
)
# Get cart items - should be 3.
cart_items = app.get_cart_items(cart_id)
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):
app.add_item_to_cart(
cart_id=cart_id,
product_id=product_id4,
name="Milk",
description="A very nice milk",
price=Decimal("1.99"),
)
# Get cart items - should be 3.
cart_items = app.get_cart_items(cart_id)
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.
app.remove_item_from_cart(
cart_id=cart_id,
product_id=product_id2,
)
# Get cart items - should be 2.
cart_items = app.get_cart_items(cart_id)
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):
app.remove_item_from_cart(
cart_id=cart_id,
product_id=product_id2,
)
# Add item to cart.
app.add_item_to_cart(
cart_id=cart_id,
product_id=product_id5,
name="Spoon",
description="A very nice spoon",
price=Decimal("0.99"),
)
# Get cart items - should be 3.
cart_items = app.get_cart_items(cart_id)
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):
app.submit_cart(cart_id)
# Adjust product inventory.
app.adjust_product_inventory(
product_id=product_id3,
adjustment=3,
)
# Insufficient inventory error.
with self.assertRaises(InsufficientInventoryError):
app.submit_cart(cart_id)
# Add item to shop.
app.add_product_to_shop(
product_id=product_id5,
name="Spoon",
description="A very nice spoon",
price=Decimal("0.99"),
)
# Adjust product inventory.
app.adjust_product_inventory(
product_id=product_id5,
adjustment=3,
)
# Submit cart.
app.submit_cart(cart_id)
# Cart already submitted.
with self.assertRaises(CartAlreadySubmittedError):
app.add_item_to_cart(
cart_id=cart_id,
product_id=product_id4,
name="Milk",
description="A very nice milk",
price=Decimal("1.99"),
)
with self.assertRaises(CartAlreadySubmittedError):
app.remove_item_from_cart(
cart_id=cart_id,
product_id=product_id1,
)
with self.assertRaises(CartAlreadySubmittedError):
app.clear_cart(
cart_id=cart_id,
)
with self.assertRaises(CartAlreadySubmittedError):
app.submit_cart(
cart_id=cart_id,
)
Code reference¶
- class examples.shopstandard.application.Shop(env: Mapping[str, str] | None = None)[source]¶
Bases:
PydanticApplication
- list_products_in_shop() Sequence[ProductDetails] [source]¶
- add_item_to_cart(cart_id: UUID, product_id: UUID, name: str, description: str, price: Decimal) None [source]¶
- name = 'Shop'¶
- class examples.shopstandard.domain.ProductDetails(*, id: UUID, name: str, description: str, price: Decimal, inventory: int)[source]¶
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=True), '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.shopstandard.domain.Product(*args: Any, **kwargs: Any)[source]¶
Bases:
Aggregate
- class InventoryAdjusted(*, originator_id: UUID, originator_version: int, timestamp: datetime, adjustment: int)[source]¶
Bases:
DecoratorEvent
,InventoryAdjusted
- 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), 'timestamp': FieldInfo(annotation=datetime, 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 Created(*, originator_id: UUID, originator_version: int, timestamp: datetime, originator_topic: str, 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_topic': FieldInfo(annotation=str, required=True), 'originator_version': FieldInfo(annotation=int, required=True), 'price': FieldInfo(annotation=Decimal, required=True), 'timestamp': FieldInfo(annotation=datetime, 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_type¶
alias of
UUID
- name: str¶
- description: str¶
- price: Decimal¶
- class Event(*, originator_id: UUID, originator_version: int, timestamp: datetime)¶
Bases:
Event
- 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), 'timestamp': FieldInfo(annotation=datetime, 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_type¶
alias of
UUID
- class examples.shopstandard.domain.CartItem(*, product_id: UUID, name: str, description: str, price: Decimal)[source]¶
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.shopstandard.domain.Cart(*args: Any, **kwargs: Any)[source]¶
Bases:
Aggregate
- class ItemAdded(*, originator_id: UUID, originator_version: int, timestamp: datetime, product_id: UUID, name: str, description: str, price: Decimal)[source]¶
Bases:
DecoratorEvent
,ItemAdded
- 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), 'timestamp': FieldInfo(annotation=datetime, 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 ItemRemoved(*, originator_id: UUID, originator_version: int, timestamp: datetime, product_id: UUID)[source]¶
Bases:
DecoratorEvent
,ItemRemoved
- 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), 'timestamp': FieldInfo(annotation=datetime, 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 Cleared(*, originator_id: UUID, originator_version: int, timestamp: datetime)[source]¶
Bases:
DecoratorEvent
,Cleared
- 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), 'timestamp': FieldInfo(annotation=datetime, 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 Submitted(*, originator_id: UUID, originator_version: int, timestamp: datetime)[source]¶
Bases:
DecoratorEvent
,Submitted
- 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), 'timestamp': FieldInfo(annotation=datetime, 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 Created(*, originator_id: UUID, originator_version: int, timestamp: datetime, originator_topic: str)¶
-
- 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_topic': FieldInfo(annotation=str, required=True), 'originator_version': FieldInfo(annotation=int, required=True), 'timestamp': FieldInfo(annotation=datetime, 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_type¶
alias of
UUID
- class Event(*, originator_id: UUID, originator_version: int, timestamp: datetime)¶
Bases:
Event
- 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), 'timestamp': FieldInfo(annotation=datetime, 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_type¶
alias of
UUID