Views

Views, a.k.a Read models, are representations of data optimized for querying and reading purposes. It is designed to provide data in a format that makes it easy and efficient to read, often tailored to the specific needs of a particular view or user interface.

Views are typically populated in response to Domain Events raised in the domain model.

Defining a View

Views are defined with the Domain.view decorator.

@domain.view
class ProductInventory:
    product_id = Identifier(identifier=True, required=True)
    name = String(max_length=100, required=True)
    description = Text(required=True)
    price = Float(required=True)
    stock_quantity = Integer(default=0)

Workflow

ManageInventory Command Handler handles AdjustStock command, loads the product and updates it, and then persists the product, generating domain events.

sequenceDiagram
  autonumber
  App->>Manage Inventory: AdjustStock object
  Manage Inventory->>Manage Inventory: Extract data and load product
  Manage Inventory->>product: adjust stock
  product->>product: Mutate
  product-->>Manage Inventory: 
  Manage Inventory->>Repository: Persist product
  Repository->>Broker: Publish events

The events are then consumend by the event handler that loads the view record and updates it.

sequenceDiagram
  autonumber
  Broker-->>Sync Inventory: Pull events
  Sync Inventory->>Sync Inventory: Extract data and load inventory record
  Sync Inventory->>inventory: update
  inventory->>inventory: Mutate
  inventory-->>Sync Inventory: 
  Sync Inventory->>Repository: Persist inventory record

Example

Below is a full-blown example of a view ProductInventory synced with the Product aggregate with the help of ProductAdded and StockAdjusted domain events.

from protean import Domain, handle
from protean.fields import Identifier, Integer, String

domain = Domain(__file__, load_toml=False)
domain.config["event_processing"] = "sync"


@domain.event(part_of="Order")
class OrderShipped:
    order_id = Identifier(required=True)
    book_id = Identifier(required=True)
    quantity = Integer(required=True)
    total_amount = Integer(required=True)


@domain.aggregate
class Order:
    book_id = Identifier(required=True)
    quantity = Integer(required=True)
    total_amount = Integer(required=True)
    status = String(choices=["PENDING", "SHIPPED", "DELIVERED"], default="PENDING")

    def ship_order(self):
        self.status = "SHIPPED"

        self.raise_(  # (1)
            OrderShipped(
                order_id=self.id,
                book_id=self.book_id,
                quantity=self.quantity,
                total_amount=self.total_amount,
            )
        )


@domain.aggregate
class Inventory:
    book_id = Identifier(required=True)
    in_stock = Integer(required=True)


@domain.event_handler(part_of=Inventory, stream_category="order")
class ManageInventory:
    @handle(OrderShipped)
    def reduce_stock_level(self, event: OrderShipped):
        repo = domain.repository_for(Inventory)
        inventory = repo._dao.find_by(book_id=event.book_id)

        inventory.in_stock -= event.quantity  # (2)

        repo.add(inventory)


domain.init()
with domain.domain_context():
    # Persist Order
    order = Order(book_id=1, quantity=10, total_amount=100)
    domain.repository_for(Order).add(order)

    # Persist Inventory
    inventory = Inventory(book_id=1, in_stock=100)
    domain.repository_for(Inventory).add(inventory)

    # Ship Order
    order.ship_order()
    domain.repository_for(Order).add(order)

    # Verify that Inventory Level has been reduced
    stock = domain.repository_for(Inventory).get(inventory.id)
    print(stock.to_dict())
    assert stock.in_stock == 90