Domain Driven Design:
Project description
Inspired by Udi Dahan’s fabulous Domain Events implementation in particular, and by Domain Driven Design best practices in general.
- The DDD Domain Events package makes it easy to:
Register to Domain Events from the Application Layer
Raise Domain Events from the Domain Layer so they can be handled in the Application Layer
The Domain Events are local to the execution thread (via Python’s threading.local) and hence are thread specific.
Installing ddd-domain-events
pip install ddd-domain-events
Using DDD Domain Events
Using Domain Events in an existing Application Service can easily be done by using Python’s with-statement.
For example:
Somewhere in the Application Layer…
with DomainEvents() as domain_events: high_price_volume_callback = DomainEventCallable(OrderEvent.HIGH_VOLUME_PRICE, cls.notify_top_management), domain_events.register_event(callback) order = Order() order.add_order_items(order_items)
** Somewhere in the Domain Layer…**
DomainEvents.raise_event(OrderEvent.HIGH_VOLUME_PRICE, order=self)
How it works
Bellow is a simplified example that should help you understand how and when you might choose to use Domain Events.
Step 1: Define a Domain Event Type in your Domain Layer
class OrderEvent(Enum): """Domain Event raised for special order use cases""" HIGH_QUANTITY = 'HIGH_QUANTITY' HIGH_VOLUME_PRICE = 'HIGH_VOLUME_PRICE'
Step 2: Define a Domain Entity that raises Domain Events
class OrderItem: """OrderItem value object that contains order details for a single item""" def __init__(self, product_id: str, price: float, quantity: int): self.product_id = product_id self.price = price self.quantity = quantity class Order: """Order entity that contains order items""" HIGH_VOLUME_PRICE = 1_000_000 HIGH_QUANTITY = 10_000 def __init__(self): self._order_items = [] @property def order_items(self): for order_item in self._order_items: yield order_item def add_order_items(self, order_items: List[OrderItem]) -> None: total_price = 0 total_quantity = 0 for order_item in order_items: total_price += (order_item.price * order_item.quantity) total_quantity += order_item.quantity # Process the actual business logic related to this method, # which is add OrderItem value objects to this Order Entity self._order_items.append(order_item) # Notify whoever might be interested about high price volume orders if total_price >= self.HIGH_VOLUME_PRICE: DomainEvents.raise_event(OrderEvent.HIGH_VOLUME_PRICE, order=self) # Notify whoever might be interested about high quantity volume orders if total_quantity >= self.HIGH_QUANTITY: DomainEvents.raise_event(OrderEvent.HIGH_QUANTITY, order=self)
Step 3: Define an Application Service that registers to Domain Events
class OrderService: """Application Service for handling Order related operations""" @classmethod def create_order(cls, order_items: List[OrderItem]) -> Order: with DomainEvents() as domain_events: # Create callbacks for 'side effects' that are related to domain logic, # and which should be handled by the Application Layer callbacks = [ DomainEventCallable(OrderEvent.HIGH_VOLUME_PRICE, cls.notify_top_management), DomainEventCallable(OrderEvent.HIGH_VOLUME_PRICE, cls.notify_sales_team), DomainEventCallable(OrderEvent.HIGH_QUANTITY, cls.notify_inventory_team) ] # Register for these domain events for callback in callbacks: domain_events.register_event(callback) order = Order() order.add_order_items(order_items) return order @staticmethod def notify_sales_team(order: Order) -> None: """A callback for notifying the sales team about the important order""" @staticmethod def notify_top_management(order: Order) -> None: """A callback for notifying the top management about the important order""" @staticmethod def notify_inventory_team(order: Order) -> None: """A callback for notifying the inventory team required quantities"""
Project details
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.