In the intricate landscape of modern software development, Domain-Driven Design (DDD) stands as a powerful methodology for tackling complex business domains. It emphasizes a deep understanding of the core business, fostering a shared language between domain experts and developers, and structuring software to reflect the domain model. However, the initial phases of DDD, particularly the elicitation of the Ubiquitous Language and the identification of Bounded Contexts, can be time-consuming and require significant human interaction and expertise. This article explores how Large Language Model (LLM) chatbots can serve as a revolutionary tool to assist and accelerate the DDD process, whether powered by local or remote LLMs. We will delve into the benefits and liabilities of this approach, broadly and deeply cover all levels of DDD, and illustrate concepts with practical examples.
Introduction to Domain-Driven Design and LLMs
Domain-Driven Design, pioneered by Eric Evans, is an approach to software development that focuses on modeling software to match a domain according to input from domain experts. Its core tenets include building a Ubiquitous Language, defining Bounded Contexts, and crafting rich domain models. The goal is to create software that is aligned with the business, adaptable to change, and maintainable over time.
Large Language Models, on the other hand, are advanced AI systems capable of understanding, generating, and processing human language. Their ability to synthesize information, engage in conversational dialogue, and even generate code makes them a compelling candidate for augmenting human intelligence in complex tasks. By combining the structured thinking of DDD with the conversational power of LLMs, we can create intelligent assistants that guide developers and domain experts through the DDD journey, making it more accessible and efficient.
Benefits of an LLM Chatbot for Domain-Driven Design
Integrating an LLM chatbot into the DDD workflow offers several compelling advantages, streamlining processes and enhancing collaboration.
An LLM chatbot can actively participate in conversations with domain experts and developers, identifying key terms, concepts, and synonyms, thereby accelerating Ubiquitous Language Development. It can then propose a consistent Ubiquitous Language, flagging ambiguities and suggesting clear definitions, which significantly speeds up the consensus-building phase. The chatbot acts as a tireless scribe and an intelligent thesaurus, ensuring that all stakeholders speak the same domain-specific language. By analyzing discussions, user stories, and existing documentation, an LLM can help identify natural boundaries within the domain, leading to Enhanced Bounded Context Identification. It can suggest potential Bounded Contexts by grouping related concepts and functionalities, and even propose relationships between these contexts, aiding in the creation of initial Context Maps, which helps teams to avoid monolithic designs and promotes a modular architecture.
The chatbot acts as an intelligent intermediary, translating complex business jargon into technical terms and vice versa, which results in Improved Communication Between Domain Experts and Developers. It can clarify misunderstandings, elaborate on concepts, and ensure that both parties have a shared understanding of the domain model, fostering a more productive and less error-prone communication channel. Based on conversational input and defined Ubiquitous Language, an LLM can generate initial drafts of DDD artifacts, providing Automated Generation of Initial DDD Artifacts. This includes basic class definitions for Entities, Value Objects, Aggregates, and even skeletal Domain Services, offering a concrete starting point for development teams, reducing boilerplate code, and allowing developers to focus on refining the domain logic.
When faced with design dilemmas, the chatbot can propose various DDD patterns and architectural choices, explaining their pros and cons in the context of the specific domain problem, thus providing Support for Exploring Design Alternatives. It can simulate potential impacts of different design decisions, helping teams make informed choices without extensive manual analysis. For teams new to DDD or those needing a quick refresher, the chatbot can serve as an always-available DDD expert, offering On-Demand DDD Expertise. It can explain complex patterns, provide examples, and answer questions about best practices, effectively democratizing access to DDD knowledge and reducing the learning curve.
Liabilities of an LLM Chatbot for Domain-Driven Design
While the benefits are substantial, it is crucial to acknowledge the potential drawbacks and challenges associated with relying on LLM chatbots for DDD.
LLMs can sometimes generate plausible but incorrect information, a phenomenon known as hallucination, posing a Risk of Hallucination and Inaccuracies. If not carefully validated by human experts, these inaccuracies could lead to flawed domain models and ultimately, incorrect software implementations, undermining the very purpose of DDD. The effectiveness of the chatbot is directly tied to the quality of the underlying LLM and its training data, indicating a Dependence on LLM Quality and Training Data. If the model lacks sufficient understanding of software engineering principles, DDD patterns, or the specific business domain, its suggestions may be suboptimal or irrelevant.
When using remote LLMs (e.g., cloud-based APIs), sensitive domain information must be transmitted to external servers, raising Data Privacy and Security Concerns (Especially with Remote LLMs). This is particularly critical for proprietary business logic or regulated industries, requiring careful consideration of data governance and compliance. An excessive dependence on the chatbot might reduce the critical thinking and problem-solving skills of developers and domain experts, potentially Stifling Human Creativity and Critical Thinking. The human element of deep domain exploration, nuanced understanding, and innovative design should always remain central, with the LLM acting as an assistant, not a replacement.
Integrating an LLM chatbot into existing development workflows can be complex, representing an Integration Complexity (Local vs. Remote). Choosing between local and remote LLMs involves trade-offs in terms of infrastructure, maintenance, and API management, each presenting its own set of technical challenges. Remote LLMs typically operate on a pay-per-use model, which can accumulate significant costs, especially during intensive DDD phases, leading to Cost Implications. Local LLMs, while avoiding per-query costs, require substantial upfront investment in powerful hardware and ongoing maintenance, making cost a critical factor in decision-making.
Core Concepts of Domain-Driven Design (DDD) and LLM Integration
Let us now explore how LLM chatbots can specifically assist with the fundamental building blocks of DDD.
Ubiquitous Language: This is the shared language developed by domain experts and developers, used consistently in all communication and within the code itself. An LLM chatbot can facilitate its creation by listening to conversations, extracting key nouns and verbs, and proposing a glossary of terms. For example, if a domain expert describes "a customer placing an order," the LLM can suggest 'Customer' and 'Order' as core entities and 'PlaceOrder' as a potential domain service operation. It can also identify synonyms (e.g., "client" vs. "customer") and help standardize terminology.
Bounded Contexts: A Bounded Context defines a logical boundary within which a specific domain model is applicable, and terms within that context have a precise, unambiguous meaning. The LLM can analyze functional requirements and discussions to identify natural groupings of responsibilities. For instance, in an e-commerce system, the LLM might suggest distinct Bounded Contexts like 'Sales', 'Inventory', 'Shipping', and 'Billing', based on how different aspects of the business are discussed and managed. It can then help define the interfaces and communication patterns between these contexts.
Entities: An Entity is an object defined by its identity, rather than its attributes. It has a life cycle and typically represents a distinct thing in the domain that needs to be tracked. An LLM, given a description of a business process, can identify potential entities. For an "Order Management System," the chatbot might suggest 'Order' and 'Customer' as entities. It can then propose initial attributes and methods based on common patterns.
--------------------------------------------------------------------
# Example: Initial Entity suggestion for an 'Order'
# Chatbot output based on user input: "Describe an order in our system."
class Order:
def __init__(self, order_id: str, customer_id: str, order_date: datetime):
self.order_id = order_id
self.customer_id = customer_id
self.order_date = order_date
self.line_items = [] # List of LineItem objects
self.status = "PENDING" # Initial status
def add_line_item(self, line_item):
# Logic to add a line item to the order
self.line_items.append(line_item)
def calculate_total(self) -> float:
# Logic to calculate the total price of the order
return sum(item.calculate_subtotal() for item in self.line_items)
def place_order(self):
# Logic to change order status to 'PLACED' and trigger events
if self.status == "PENDING":
self.status = "PLACED"
print(f"Order {self.order_id} has been placed.")
# Potentially publish an OrderPlaced event here
else:
print(f"Order {self.order_id} cannot be placed from status {self.status}.")
# This is a basic structure. An LLM can generate this and suggest further refinements.
--------------------------------------------------------------------
Value Objects: A Value Object is an object that describes a characteristic or attribute but has no conceptual identity. It is immutable and compared by its attributes. Examples include 'Money', 'Address', or 'Quantity'. An LLM can help identify these by looking for concepts that are typically exchanged or measured, and whose identity is not important, only their value.
--------------------------------------------------------------------
# Example: Value Object suggestion for 'Money'
# Chatbot output based on user input: "How do we handle currency and amounts?"
import decimal
class Money:
def __init__(self, amount: decimal.Decimal, currency: str):
if not isinstance(amount, decimal.Decimal):
raise TypeError("Amount must be a Decimal.")
if not isinstance(currency, str) or len(currency) != 3:
raise ValueError("Currency must be a 3-letter string (e.g., 'USD').")
self._amount = amount
self._currency = currency.upper()
@property
def amount(self) -> decimal.Decimal:
return self._amount
@property
def currency(self) -> str:
return self._currency
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("Cannot add Money with different currencies.")
return Money(self.amount + other.amount, self.currency)
def subtract(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("Cannot subtract Money with different currencies.")
return Money(self.amount - other.amount, self.currency)
def __eq__(self, other):
if not isinstance(other, Money):
return NotImplemented
return self.amount == other.amount and self.currency == other.currency
def __hash__(self):
return hash((self.amount, self.currency))
def __str__(self):
return f"{self.currency} {self.amount:.2f}"
def __repr__(self):
return f"Money(amount=Decimal('{self.amount}'), currency='{self.currency}')"
# This immutable class ensures monetary calculations are handled consistently.
--------------------------------------------------------------------
Aggregates: An Aggregate is a cluster of associated objects treated as a unit for data changes. It has an Aggregate Root, which is an Entity that controls access to other objects within the aggregate and ensures the aggregate's consistency. An LLM can assist in defining these consistency boundaries. For an 'Order', the LLM might suggest that 'Line Items' belong within the 'Order' aggregate, and that 'Customer' is a separate aggregate, merely referenced by 'Order' via a 'CustomerId'. It can also help define invariants that must always hold true for the aggregate.
Domain Services: A Domain Service is a stateless operation that performs significant domain logic that does not naturally fit within an Entity or Value Object. An LLM can identify complex business operations that involve multiple aggregates or external systems and suggest them as candidates for Domain Services. For example, 'PlaceOrderService' might be suggested as a service responsible for orchestrating the creation of an order, checking inventory, and publishing an 'OrderPlaced' event.
Domain Events: A Domain Event represents something that happened in the domain that domain experts care about. They are immutable facts about past occurrences. The LLM can help identify these by analyzing phrases like "when X happens," "after Y occurs," or "upon Z completion." For an 'Order' aggregate, 'OrderPlaced', 'OrderShipped', or 'PaymentReceived' could be identified as crucial domain events. The chatbot can also suggest the data that should be included in each event.
Repositories: A Repository provides a mechanism for encapsulating storage, retrieval, and search behavior that mimics a collection of objects. It abstracts the persistence layer from the domain model. An LLM can suggest repository interfaces for each aggregate root, outlining methods like 'get_by_id', 'save', and 'delete'.
Modules: Modules are used to organize the domain model into logical groupings, often aligning with Bounded Contexts or subdomains. The LLM can propose module structures based on the identified Bounded Contexts and the relationships between them, helping to maintain a clean and understandable codebase.
DDD Patterns and Diagrams with LLM Assistance
LLM chatbots can extend their utility beyond core concepts to assist with both strategic and tactical DDD patterns, and even help in generating representations like diagrams.
Strategic Design: This level of DDD focuses on understanding the overall business domain and dividing it into Bounded Contexts.
* Context Mapping: This pattern visually represents the relationships and integrations between different Bounded Contexts. An LLM can analyze descriptions of business processes and system interactions to suggest a textual representation of a Context Map. For example, it might describe a relationship where the 'Sales Context' acts as a Customer to the 'Inventory Context' which acts as a Supplier. This implies that Sales depends on Inventory for product availability. It might also note an 'Anti-Corruption Layer' protecting the Sales Context from the Inventory Context's model. The LLM can help define the nature of the relationship (e.g., Shared Kernel, Conformist, Anti-Corruption Layer) and suggest the necessary integration mechanisms.
* Subdomains: The LLM can help categorize subdomains within a larger business domain as 'Core', 'Supporting', or 'Generic', based on their strategic importance and uniqueness to the business. For an e-commerce system, 'Order Fulfillment' might be identified as a core subdomain, while 'User Authentication' would be a generic subdomain.
Tactical Design: This level focuses on the detailed implementation of the domain model within a Bounded Context, using patterns like Aggregates, Entities, Value Objects, Domain Services, Repositories, and Factories. An LLM can generate initial code structures for these patterns, complete with comments and basic documentation, based on the Ubiquitous Language and domain descriptions provided.
Diagrams: While LLMs are primarily text-based, they can assist in creating or interpreting diagrammatic representations.
* Event Storming: This collaborative workshop technique identifies domain events, commands, and aggregates. An LLM can "simulate" an Event Storming session by asking guiding questions and summarizing identified events and commands, or by analyzing transcripts of actual sessions to extract key elements.
* UML (Unified Modeling Language): An LLM can generate basic UML class diagrams or sequence diagrams in a textual format (e.g., PlantUML syntax, or a simplified ASCII representation) from a textual description of entities, their relationships, and interactions. This provides a structured visual aid for understanding the domain model.
Constituents of an LLM Chatbot for DDD
Building an effective LLM chatbot for DDD involves several key components working in concert.
The User Interface (UI) is the primary interaction point for developers and domain experts, which can be a web-based chat application, a desktop client, or a command-line interface (CLI). The UI needs to be intuitive, allowing users to ask questions, provide domain descriptions, and receive structured responses. The LLM Integration Layer handles the communication with the underlying Large Language Model. For remote LLMs, it involves making API calls (e.g., HTTP requests) to services like OpenAI GPT, Anthropic Claude, or Google Gemini, while for local LLMs, it entails loading the model into memory and interacting with it via local libraries such as 'ollama', 'transformers', or 'llama.cpp'. This layer manages authentication, request formatting, and response parsing.
The Prompt Engineering Module is a critical component responsible for crafting effective prompts that guide the LLM to produce relevant and accurate DDD-specific outputs. It translates user queries into detailed instructions for the LLM, often including context, desired output format, and specific DDD principles to adhere to. For example, if a user asks "Define the Order aggregate," the prompt engineering module might prepend instructions like "Act as a DDD expert. Define the Order aggregate, including its root, invariants, and contained entities/value objects. Provide Python code." To maintain a coherent and useful conversation, the chatbot needs to remember past interactions and the evolving DDD context, which is managed by the Context Management/Memory module. This module stores conversation history, identified Ubiquitous Language terms, Bounded Context definitions, and other relevant domain knowledge, allowing the LLM to build upon previous statements and provide contextually aware responses, avoiding repetitive information or contradictory suggestions.
To enhance accuracy and reduce hallucinations, the chatbot can be augmented with a knowledge base containing authoritative DDD resources, project-specific documentation, or examples, leveraging a Knowledge Base/Retrieval Augmented Generation (RAG) module. The RAG module retrieves relevant information from this knowledge base based on the user's query and injects it into the LLM's prompt, ensuring that the LLM's responses are grounded in factual and specific information, rather than solely relying on its general training data. LLMs often produce free-form text, so the Output Parser/Formatter module is responsible for parsing the LLM's response and formatting it into a structured, usable output. This could involve extracting code blocks, identifying key terms for the Ubiquitous Language glossary, or converting textual descriptions into a format suitable for diagram generation tools. For instance, it might extract a Python class definition from the LLM's response and present it cleanly.
An essential component for continuous improvement is the Feedback Loop, which allows users to correct inaccuracies, refine suggestions, or rate the usefulness of the chatbot's responses. This feedback can then be used to fine-tune prompts, update the knowledge base, or even retrain the LLM (if using a local model), leading to increasingly accurate and helpful interactions over time.
LLM Choices: Local vs. Remote
The decision to use a local or remote LLM significantly impacts the chatbot's architecture, performance, cost, and data handling.
Remote LLMs are cloud-hosted models provided by vendors like OpenAI (GPT series), Anthropic (Claude), or Google (Gemini). Their advantages include High Performance and Scale, as they are typically hosted on powerful infrastructure, offering high performance and the ability to handle a large volume of requests without requiring local hardware investment. They also provide access to Large Models, as vendors often provide access to their largest and most capable models, which generally exhibit superior understanding and generation capabilities. Additionally, they offer Easy Access and Maintenance, as integration is usually straightforward via well-documented APIs, and the vendor handles all infrastructure, updates, and maintenance. However, there are drawbacks: Cost, as usage is typically billed per token, which can become expensive, especially for extensive use or large context windows. Data Privacy and Security are concerns, as sending sensitive domain data to a third-party cloud service raises privacy and security issues, especially for proprietary business logic or regulated industries, requiring careful review of vendor policies and compliance certifications. Internet Dependency is another factor, as a stable internet connection is mandatory for operation, and Vendor Lock-in can occur, as relying heavily on a single vendor's API can create dependency and make switching providers challenging. Integration typically involves HTTP requests to a RESTful API, authenticated with API keys.
Local LLMs are models that run directly on an organization's own hardware, such as Llama 2, Mistral, Falcon, or Phi-3. Their advantages include Data Privacy and Security, as sensitive data remains within the organization's control, addressing privacy concerns. There is No Internet Dependency, as once downloaded, the models can operate entirely offline. They can be Cost-Effective (after initial setup), as while requiring upfront hardware investment, there are no per-token costs, making them potentially more economical for high-volume internal use. Organizations also have Full Control and Customization over the model, allowing for fine-tuning with specific domain data and custom modifications. However, there are disadvantages: Hardware Requirements, as running powerful LLMs locally demands significant computational resources (GPUs, ample RAM), which can be a substantial upfront investment. Performance Variability can be an issue, as performance can vary greatly depending on the local hardware and the chosen model's size, with smaller models potentially having reduced capabilities. Model Management falls on the organization, as they are responsible for managing model deployment, updates, and infrastructure maintenance. Finally, Context Window Limitations mean that many local models, especially smaller ones, have more restrictive context windows compared to their remote counterparts, limiting the amount of conversation history they can process at once. Integration involves using libraries like 'ollama', 'transformers', or 'llama.cpp' to load and interact with the model programmatically.
Building the Chatbot: A Conceptual Walkthrough
Let us outline the steps to construct an LLM chatbot for DDD, using our 'Order Management System' as a running example.
Step 1: Define the Goal. Clearly articulate what specific DDD tasks the chatbot will assist with. For our Order Management System, the goal is to help define the Ubiquitous Language, identify Bounded Contexts, and generate initial code for core domain entities like 'Order' and 'LineItem'.
Step 2: Choose LLM Strategy. Decide between a local or remote LLM. For a proof-of-concept or if data privacy is not a primary concern initially, a remote LLM like GPT-4 might be chosen for its superior capabilities. If the system needs to handle highly sensitive order data, a local LLM running on internal servers would be preferred. Let us assume we start with a remote LLM for ease of initial development.
Step 3: Design the Interaction Flow. Determine how users will interact with the chatbot. An example flow could be: User: "Help me define the core concepts for an Order Management System." Chatbot: "Certainly! Let's start with the Ubiquitous Language. What are the key nouns and verbs related to orders?" User: "Customers place orders, orders contain line items, products are in line items, orders are shipped." Chatbot: "Excellent. I suggest 'Customer', 'Order', 'LineItem', 'Product', and 'Shipment' as core terms. How do we define an 'Order'?" This flow guides the user through DDD concepts.
Step 4: Implement Core Components.
* UI: A simple web interface using a framework like Flask or FastAPI for the backend and a basic HTML/JavaScript frontend for chat.
* LLM Integration: Use the chosen LLM's Python SDK (e.g., 'openai' library) to send prompts and receive responses.
* Prompt Engineering: Develop a set of initial system prompts and user prompt templates.
--------------------------------------------------------------------
# Example: Basic prompt for defining an entity
def generate_entity_prompt(entity_name: str, domain_description: str) -> str:
return f"""
You are an expert in Domain-Driven Design (DDD).
Your task is to help define the core domain model for an Order Management System.
Based on the following domain description:
"{domain_description}"
Please define the DDD Entity '{entity_name}'.
Provide its key attributes, identify its unique identity, and suggest initial methods.
Explain why it is an Entity (identity-based) rather than a Value Object.
Provide a Python class definition for this entity, including type hints and docstrings.
Ensure the code follows clean code principles.
"""
--------------------------------------------------------------------
Step 5: Incorporate DDD Knowledge (RAG). Create a small knowledge base of common DDD patterns, definitions, and best practices. When the user asks about "Aggregates," the RAG system can retrieve the definition and principles of aggregates and inject them into the LLM's prompt, ensuring the LLM's explanation is accurate and comprehensive.
Step 6: Iterate and Refine. Continuously test the chatbot with various DDD scenarios. Gather feedback from developers and domain experts. If the LLM hallucinates or provides unhelpful suggestions, refine the prompts, update the RAG knowledge base, or consider fine-tuning the model (if local) or exploring different LLM parameters. For example, if the LLM consistently forgets to include an 'order_date' in the 'Order' entity, the prompt can be explicitly updated to "ensure 'order_date' is included."
Conclusion
The integration of LLM chatbots into the Domain-Driven Design process presents a significant opportunity to enhance efficiency, foster collaboration, and democratize access to DDD expertise. From accelerating the development of a Ubiquitous Language and identifying Bounded Contexts to generating initial code artifacts, LLMs can act as powerful assistants, guiding teams through the complexities of domain modeling. However, it is imperative to approach this technology with a clear understanding of its liabilities, particularly concerning potential inaccuracies, data privacy, and the risk of over-reliance. The human element of critical thinking, deep domain understanding, and creative problem-solving must always remain at the core of DDD. By strategically leveraging LLM chatbots as intelligent tools, not replacements, organizations can unlock new levels of productivity and build more robust, business-aligned software systems.
Addendum: Full Running Example Code for Order Management System
This addendum provides a more complete, albeit simplified, Python implementation of the core domain model for an Order Management System, demonstrating the DDD concepts discussed in the article. This code is designed to be runnable and illustrates the structure an LLM chatbot could help generate and refine.
--------------------------------------------------------------------
# order_management_system/domain/value_objects.py
import decimal
from typing import NamedTuple
# A Value Object representing monetary amounts.
# It is immutable and compared by its value (amount and currency).
class Money:
def __init__(self, amount: decimal.Decimal, currency: str):
if not isinstance(amount, decimal.Decimal):
raise TypeError("Amount must be a Decimal.")
if not isinstance(currency, str) or len(currency) != 3:
raise ValueError("Currency must be a 3-letter string (e.g., 'USD').")
# Store as private attributes to enforce immutability
self._amount = amount
self._currency = currency.upper()
@property
def amount(self) -> decimal.Decimal:
return self._amount
@property
def currency(self) -> str:
return self._currency
def add(self, other: 'Money') -> 'Money':
"""Adds another Money object to this one, returning a new Money object."""
if self.currency != other.currency:
raise ValueError("Cannot add Money with different currencies.")
return Money(self.amount + other.amount, self.currency)
def multiply(self, factor: decimal.Decimal) -> 'Money':
"""Multiplies the Money amount by a factor, returning a new Money object."""
if not isinstance(factor, decimal.Decimal):
raise TypeError("Factor must be a Decimal.")
return Money(self.amount * factor, self.currency)
def __eq__(self, other):
"""Compares two Money objects for equality based on amount and currency."""
if not isinstance(other, Money):
return NotImplemented
return self.amount == other.amount and self.currency == other.currency
def __hash__(self):
"""Generates a hash for the Money object, allowing it to be used in sets/dicts."""
return hash((self.amount, self.currency))
def __str__(self):
"""Returns a user-friendly string representation."""
return f"{self.currency} {self.amount:.2f}"
def __repr__(self):
"""Returns a developer-friendly string representation."""
return f"Money(amount=Decimal('{self.amount}'), currency='{self.currency}')"
# A Value Object representing a quantity of items.
# Also immutable and compared by its value.
class Quantity:
def __init__(self, value: int):
if not isinstance(value, int) or value < 0:
raise ValueError("Quantity must be a non-negative integer.")
self._value = value
@property
def value(self) -> int:
return self._value
def add(self, other: 'Quantity') -> 'Quantity':
"""Adds another Quantity object, returning a new Quantity."""
return Quantity(self.value + other.value)
def __eq__(self, other):
"""Compares two Quantity objects for equality."""
if not isinstance(other, Quantity):
return NotImplemented
return self.value == other.value
def __hash__(self):
"""Generates a hash for the Quantity object."""
return hash(self.value)
def __str__(self):
"""Returns a user-friendly string representation."""
return str(self.value)
def __repr__(self):
"""Returns a developer-friendly string representation."""
return f"Quantity(value={self.value})"
# A Value Object representing a product reference within an order line item.
# It's immutable and identified by its product ID and name.
class ProductInfo(NamedTuple):
product_id: str
name: str
unit_price: Money
def __str__(self):
return f"{self.name} ({self.product_id}) @ {self.unit_price}"
# A Value Object representing a single line item within an order.
# It contains product information, quantity, and calculates its subtotal.
class OrderLineItem:
def __init__(self, product_info: ProductInfo, quantity: Quantity):
if not isinstance(product_info, ProductInfo):
raise TypeError("product_info must be a ProductInfo Value Object.")
if not isinstance(quantity, Quantity):
raise TypeError("quantity must be a Quantity Value Object.")
self._product_info = product_info
self._quantity = quantity
@property
def product_info(self) -> ProductInfo:
return self._product_info
@property
def quantity(self) -> Quantity:
return self._quantity
def calculate_subtotal(self) -> Money:
"""Calculates the subtotal for this line item."""
return self.product_info.unit_price.multiply(decimal.Decimal(self.quantity.value))
def __eq__(self, other):
"""Compares two OrderLineItem objects for equality."""
if not isinstance(other, OrderLineItem):
return NotImplemented
return self.product_info == other.product_info and self.quantity == other.quantity
def __hash__(self):
"""Generates a hash for the OrderLineItem object."""
return hash((self.product_info, self.quantity))
def __str__(self):
return (f"{self.quantity.value} x {self.product_info.name} "
f"({self.product_info.product_id}) - {self.calculate_subtotal()}")
def __repr__(self):
return (f"OrderLineItem(product_info={self.product_info!r}, "
f"quantity={self.quantity!r})")
# order_management_system/domain/entities.py
import uuid
from datetime import datetime
from typing import List, Optional
# Assuming Value Objects are imported from the same domain package
# from .value_objects import Money, Quantity, ProductInfo, OrderLineItem
# An Entity representing a Customer.
# It has a unique identity (customer_id) and a lifecycle.
class Customer:
def __init__(self, customer_id: str, name: str, email: str):
if not customer_id:
raise ValueError("Customer ID cannot be empty.")
if not name:
raise ValueError("Customer name cannot be empty.")
if not email:
raise ValueError("Customer email cannot be empty.")
self._customer_id = customer_id
self._name = name
self._email = email
@property
def customer_id(self) -> str:
return self._customer_id
@property
def name(self) -> str:
return self._name
@property
def email(self) -> str:
return self._email
def update_email(self, new_email: str):
"""Updates the customer's email address."""
if not new_email:
raise ValueError("New email cannot be empty.")
self._email = new_email
print(f"Customer {self.customer_id} email updated to {new_email}.")
# Potentially publish a CustomerEmailUpdated event here
def __eq__(self, other):
"""Compares two Customer objects based on their identity."""
if not isinstance(other, Customer):
return NotImplemented
return self.customer_id == other.customer_id
def __hash__(self):
"""Generates a hash for the Customer object based on its identity."""
return hash(self.customer_id)
def __str__(self):
return f"Customer(ID: {self.customer_id}, Name: {self.name}, Email: {self.email})"
def __repr__(self):
return (f"Customer(customer_id='{self.customer_id}', name='{self.name}', "
f"email='{self.email}')")
# order_management_system/domain/aggregates.py
# Assuming Value Objects and Entities are imported
# from .value_objects import Money, Quantity, ProductInfo, OrderLineItem
# from .entities import Customer # Customer is a separate aggregate, referenced by ID
# An Aggregate Root representing an Order.
# It encapsulates OrderLineItems and ensures the consistency of the order.
class Order:
class Status:
PENDING = "PENDING"
PLACED = "PLACED"
PAID = "PAID"
SHIPPED = "SHIPPED"
CANCELLED = "CANCELLED"
def __init__(self, order_id: str, customer_id: str, order_date: datetime):
if not order_id:
raise ValueError("Order ID cannot be empty.")
if not customer_id:
raise ValueError("Customer ID cannot be empty.")
if not isinstance(order_date, datetime):
raise TypeError("Order date must be a datetime object.")
self._order_id = order_id
self._customer_id = customer_id
self._order_date = order_date
self._line_items: List[OrderLineItem] = []
self._status = self.Status.PENDING
self._total_price: Optional[Money] = None # Calculated on demand or when placed
@property
def order_id(self) -> str:
return self._order_id
@property
def customer_id(self) -> str:
return self._customer_id
@property
def order_date(self) -> datetime:
return self._order_date
@property
def line_items(self) -> List[OrderLineItem]:
# Return a copy to prevent external modification of the internal list
return list(self._line_items)
@property
def status(self) -> str:
return self._status
@property
def total_price(self) -> Optional[Money]:
"""Calculates the total price of the order from its line items."""
if not self._line_items:
return None
# Assuming all line items have the same currency as the first one
# In a real system, more robust currency handling would be needed.
currency = self._line_items[0].product_info.unit_price.currency
total = Money(decimal.Decimal('0.00'), currency)
for item in self._line_items:
total = total.add(item.calculate_subtotal())
self._total_price = total # Cache the total price
return self._total_price
def add_line_item(self, line_item: OrderLineItem):
"""Adds a new line item to the order."""
if self.status != self.Status.PENDING:
raise ValueError(f"Cannot add items to an order with status '{self.status}'.")
self._line_items.append(line_item)
# Invalidate cached total price
self._total_price = None
print(f"Added line item to order {self.order_id}: {line_item}")
def remove_line_item(self, product_id: str):
"""Removes a line item from the order by product ID."""
if self.status != self.Status.PENDING:
raise ValueError(f"Cannot remove items from an order with status '{self.status}'.")
initial_count = len(self._line_items)
self._line_items = [
item for item in self._line_items if item.product_info.product_id != product_id
]
if len(self._line_items) < initial_count:
# Invalidate cached total price
self._total_price = None
print(f"Removed line item for product {product_id} from order {self.order_id}.")
else:
print(f"Product {product_id} not found in order {self.order_id}.")
def place_order(self):
"""Transitions the order status to PLACED, if currently PENDING."""
if not self._line_items:
raise ValueError("Cannot place an empty order.")
if self._status == self.Status.PENDING:
self._status = self.Status.PLACED
print(f"Order {self.order_id} has been placed.")
# Here, an OrderPlaced domain event would typically be published.
else:
raise ValueError(f"Order {self.order_id} cannot be placed from status '{self.status}'.")
def mark_as_paid(self):
"""Transitions the order status to PAID, if currently PLACED."""
if self._status == self.Status.PLACED:
self._status = self.Status.PAID
print(f"Order {self.order_id} has been marked as paid.")
# Here, a PaymentReceived domain event would typically be published.
else:
raise ValueError(f"Order {self.order_id} cannot be marked paid from status '{self.status}'.")
def mark_as_shipped(self):
"""Transitions the order status to SHIPPED, if currently PAID."""
if self._status == self.Status.PAID:
self._status = self.Status.SHIPPED
print(f"Order {self.order_id} has been marked as shipped.")
# Here, an OrderShipped domain event would typically be published.
else:
raise ValueError(f"Order {self.order_id} cannot be marked shipped from status '{self.status}'.")
def cancel_order(self):
"""Cancels the order, if not already shipped or cancelled."""
if self._status in [self.Status.PENDING, self.Status.PLACED, self.Status.PAID]:
self._status = self.Status.CANCELLED
print(f"Order {self.order_id} has been cancelled.")
# Here, an OrderCancelled domain event would typically be published.
else:
raise ValueError(f"Order {self.order_id} cannot be cancelled from status '{self.status}'.")
def __eq__(self, other):
"""Compares two Order objects based on their identity."""
if not isinstance(other, Order):
return NotImplemented
return self.order_id == other.order_id
def __hash__(self):
"""Generates a hash for the Order object based on its identity."""
return hash(self.order_id)
def __str__(self):
items_str = "\n ".join(str(item) for item in self._line_items)
return (f"Order ID: {self.order_id}\n"
f" Customer ID: {self.customer_id}\n"
f" Date: {self.order_date.isoformat()}\n"
f" Status: {self.status}\n"
f" Total: {self.total_price if self.total_price else 'N/A'}\n"
f" Line Items:\n {items_str if items_str else 'No items'}")
def __repr__(self):
return (f"Order(order_id='{self.order_id}', customer_id='{self.customer_id}', "
f"order_date={self.order_date!r}, status='{self.status}', "
f"line_items={self._line_items!r})")
# order_management_system/domain/services.py
from abc import ABC, abstractmethod
# Assuming domain models are imported
# from .aggregates import Order
# from .entities import Customer
# from .value_objects import ProductInfo, Quantity, Money
# Abstract base class for a generic Repository.
# Repositories abstract the persistence layer for aggregates.
class OrderRepository(ABC):
@abstractmethod
def get_by_id(self, order_id: str) -> Optional[Order]:
"""Retrieves an Order aggregate by its ID."""
pass
@abstractmethod
def save(self, order: Order):
"""Saves or updates an Order aggregate."""
pass
@abstractmethod
def delete(self, order_id: str):
"""Deletes an Order aggregate by its ID."""
pass
class CustomerRepository(ABC):
@abstractmethod
def get_by_id(self, customer_id: str) -> Optional[Customer]:
"""Retrieves a Customer entity by its ID."""
pass
@abstractmethod
def save(self, customer: Customer):
"""Saves or updates a Customer entity."""
pass
# A Domain Service that orchestrates the process of placing an order.
# It involves multiple aggregates (Order, potentially Inventory) and is stateless.
class OrderPlacementService:
def __init__(self, order_repo: OrderRepository, customer_repo: CustomerRepository):
self._order_repo = order_repo
self._customer_repo = customer_repo
# In a real system, an InventoryService or ProductRepository might also be injected
# to check product availability.
def place_new_order(self, customer_id: str, product_data: List[dict]) -> Order:
"""
Orchestrates the creation and placement of a new order.
Args:
customer_id (str): The ID of the customer placing the order.
product_data (List[dict]): A list of dictionaries, each containing
'product_id', 'name', 'unit_price', 'quantity'.
Returns:
Order: The newly placed Order aggregate.
Raises:
ValueError: If customer not found or product data is invalid.
"""
customer = self._customer_repo.get_by_id(customer_id)
if not customer:
raise ValueError(f"Customer with ID {customer_id} not found.")
new_order_id = str(uuid.uuid4())
order = Order(order_id=new_order_id, customer_id=customer_id, order_date=datetime.now())
for item_data in product_data:
product_id = item_data.get('product_id')
product_name = item_data.get('name')
unit_price_amount = decimal.Decimal(str(item_data.get('unit_price')))
unit_price_currency = item_data.get('currency', 'USD')
quantity_value = item_data.get('quantity')
if not all([product_id, product_name, unit_price_amount, quantity_value is not None]):
raise ValueError("Invalid product data provided for order line item.")
product_info = ProductInfo(
product_id=product_id,
name=product_name,
unit_price=Money(unit_price_amount, unit_price_currency)
)
quantity = Quantity(quantity_value)
line_item = OrderLineItem(product_info=product_info, quantity=quantity)
order.add_line_item(line_item)
order.place_order() # This changes the order status and applies domain logic
self._order_repo.save(order) # Persist the new order
print(f"Order {order.order_id} successfully placed for customer {customer.name}.")
return order
# order_management_system/infrastructure/repositories.py
# This layer handles actual persistence, e.g., to a database.
# For simplicity, we'll use in-memory repositories.
# Assuming domain models are imported
# from ..domain.aggregates import Order
# from ..domain.entities import Customer
# from ..domain.services import OrderRepository, CustomerRepository
class InMemoryOrderRepository(OrderRepository):
def __init__(self):
self._orders: dict[str, Order] = {}
def get_by_id(self, order_id: str) -> Optional[Order]:
"""Retrieves an Order from memory."""
print(f"Retrieving order {order_id} from in-memory repository.")
# Return a deep copy to prevent external modification of the stored object
# In a real system, ORM would handle this.
return self._orders.get(order_id) # Simplified: returning direct reference
def save(self, order: Order):
"""Saves or updates an Order in memory."""
print(f"Saving order {order.order_id} to in-memory repository.")
self._orders[order.order_id] = order # Simplified: storing direct reference
def delete(self, order_id: str):
"""Deletes an Order from memory."""
print(f"Deleting order {order_id} from in-memory repository.")
if order_id in self._orders:
del self._orders[order_id]
class InMemoryCustomerRepository(CustomerRepository):
def __init__(self):
self._customers: dict[str, Customer] = {}
def get_by_id(self, customer_id: str) -> Optional[Customer]:
"""Retrieves a Customer from memory."""
print(f"Retrieving customer {customer_id} from in-memory repository.")
return self._customers.get(customer_id)
def save(self, customer: Customer):
"""Saves or updates a Customer in memory."""
print(f"Saving customer {customer.customer_id} to in-memory repository.")
self._customers[customer.customer_id] = customer
# order_management_system/application/main.py
# This is the application layer, orchestrating domain and infrastructure.
# Import all necessary components
# from order_management_system.domain.value_objects import Money, Quantity, ProductInfo, OrderLineItem
# from order_management_system.domain.entities import Customer
# from order_management_system.domain.aggregates import Order
# from order_management_system.domain.services import OrderPlacementService
# from order_management_system.infrastructure.repositories import InMemoryOrderRepository, InMemoryCustomerRepository
# To make the example runnable as a single file, we'll define everything here.
# In a real project, these would be in their respective files as commented above.
import decimal
import uuid
from datetime import datetime
from typing import List, Optional, NamedTuple
from abc import ABC, abstractmethod
# --- Domain Layer: Value Objects ---
class Money:
def __init__(self, amount: decimal.Decimal, currency: str):
if not isinstance(amount, decimal.Decimal):
raise TypeError("Amount must be a Decimal.")
if not isinstance(currency, str) or len(currency) != 3:
raise ValueError("Currency must be a 3-letter string (e.g., 'USD').")
self._amount = amount
self._currency = currency.upper()
@property
def amount(self) -> decimal.Decimal: return self._amount
@property
def currency(self) -> str: return self._currency
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency: raise ValueError("Cannot add Money with different currencies.")
return Money(self.amount + other.amount, self.currency)
def multiply(self, factor: decimal.Decimal) -> 'Money':
if not isinstance(factor, decimal.Decimal): raise TypeError("Factor must be a Decimal.")
return Money(self.amount * factor, self.currency)
def __eq__(self, other):
if not isinstance(other, Money): return NotImplemented
return self.amount == other.amount and self.currency == other.currency
def __hash__(self): return hash((self.amount, self.currency))
def __str__(self): return f"{self.currency} {self.amount:.2f}"
def __repr__(self): return f"Money(amount=Decimal('{self.amount}'), currency='{self.currency}')"
class Quantity:
def __init__(self, value: int):
if not isinstance(value, int) or value < 0: raise ValueError("Quantity must be a non-negative integer.")
self._value = value
@property
def value(self) -> int: return self._value
def add(self, other: 'Quantity') -> 'Quantity': return Quantity(self.value + other.value)
def __eq__(self, other):
if not isinstance(other, Quantity): return NotImplemented
return self.value == other.value
def __hash__(self): return hash(self.value)
def __str__(self): return str(self.value)
def __repr__(self): return f"Quantity(value={self.value})"
class ProductInfo(NamedTuple):
product_id: str
name: str
unit_price: Money
def __str__(self): return f"{self.name} ({self.product_id}) @ {self.unit_price}"
class OrderLineItem:
def __init__(self, product_info: ProductInfo, quantity: Quantity):
if not isinstance(product_info, ProductInfo): raise TypeError("product_info must be a ProductInfo Value Object.")
if not isinstance(quantity, Quantity): raise TypeError("quantity must be a Quantity Value Object.")
self._product_info = product_info
self._quantity = quantity
@property
def product_info(self) -> ProductInfo: return self._product_info
@property
def quantity(self) -> Quantity: return self._quantity
def calculate_subtotal(self) -> Money:
return self.product_info.unit_price.multiply(decimal.Decimal(self.quantity.value))
def __eq__(self, other):
if not isinstance(other, OrderLineItem): return NotImplemented
return self.product_info == other.product_info and self.quantity == other.quantity
def __hash__(self): return hash((self.product_info, self.quantity))
def __str__(self):
return (f"{self.quantity.value} x {self.product_info.name} "
f"({self.product_info.product_id}) - {self.calculate_subtotal()}")
def __repr__(self):
return (f"OrderLineItem(product_info={self.product_info!r}, "
f"quantity={self.quantity!r})")
# --- Domain Layer: Entities ---
class Customer:
def __init__(self, customer_id: str, name: str, email: str):
if not customer_id: raise ValueError("Customer ID cannot be empty.")
if not name: raise ValueError("Customer name cannot be empty.")
if not email: raise ValueError("Customer email cannot be empty.")
self._customer_id = customer_id
self._name = name
self._email = email
@property
def customer_id(self) -> str: return self._customer_id
@property
def name(self) -> str: return self._name
@property
def email(self) -> str: return self._email
def update_email(self, new_email: str):
if not new_email: raise ValueError("New email cannot be empty.")
self._email = new_email
print(f"Customer {self.customer_id} email updated to {new_email}.")
def __eq__(self, other):
if not isinstance(other, Customer): return NotImplemented
return self.customer_id == other.customer_id
def __hash__(self): return hash(self.customer_id)
def __str__(self): return f"Customer(ID: {self.customer_id}, Name: {self.name}, Email: {self.email})"
def __repr__(self):
return (f"Customer(customer_id='{self.customer_id}', name='{self.name}', "
f"email='{self.email}')")
# --- Domain Layer: Aggregates ---
class Order:
class Status:
PENDING = "PENDING"
PLACED = "PLACED"
PAID = "PAID"
SHIPPED = "SHIPPED"
CANCELLED = "CANCELLED"
def __init__(self, order_id: str, customer_id: str, order_date: datetime):
if not order_id: raise ValueError("Order ID cannot be empty.")
if not customer_id: raise ValueError("Customer ID cannot be empty.")
if not isinstance(order_date, datetime): raise TypeError("Order date must be a datetime object.")
self._order_id = order_id
self._customer_id = customer_id
self._order_date = order_date
self._line_items: List[OrderLineItem] = []
self._status = self.Status.PENDING
self._total_price: Optional[Money] = None
@property
def order_id(self) -> str: return self._order_id
@property
def customer_id(self) -> str: return self._customer_id
@property
def order_date(self) -> datetime: return self._order_date
@property
def line_items(self) -> List[OrderLineItem]: return list(self._line_items)
@property
def status(self) -> str: return self._status
@property
def total_price(self) -> Optional[Money]:
if not self._line_items: return None
currency = self._line_items[0].product_info.unit_price.currency
total = Money(decimal.Decimal('0.00'), currency)
for item in self._line_items: total = total.add(item.calculate_subtotal())
self._total_price = total
return self._total_price
def add_line_item(self, line_item: OrderLineItem):
if self.status != self.Status.PENDING: raise ValueError(f"Cannot add items to an order with status '{self.status}'.")
self._line_items.append(line_item)
self._total_price = None
print(f"Added line item to order {self.order_id}: {line_item}")
def remove_line_item(self, product_id: str):
if self.status != self.Status.PENDING: raise ValueError(f"Cannot remove items from an order with status '{self.status}'.")
initial_count = len(self._line_items)
self._line_items = [item for item in self._line_items if item.product_info.product_id != product_id]
if len(self._line_items) < initial_count:
self._total_price = None
print(f"Removed line item for product {product_id} from order {self.order_id}.")
else:
print(f"Product {product_id} not found in order {self.order_id}.")
def place_order(self):
if not self._line_items: raise ValueError("Cannot place an empty order.")
if self._status == self.Status.PENDING:
self._status = self.Status.PLACED
print(f"Order {self.order_id} has been placed.")
else:
raise ValueError(f"Order {self.order_id} cannot be placed from status '{self.status}'.")
def mark_as_paid(self):
if self._status == self.Status.PLACED:
self._status = self.Status.PAID
print(f"Order {self.order_id} has been marked as paid.")
else:
raise ValueError(f"Order {self.order_id} cannot be marked paid from status '{self.status}'.")
def mark_as_shipped(self):
if self._status == self.Status.PAID:
self._status = self.Status.SHIPPED
print(f"Order {self.order_id} has been marked as shipped.")
else:
raise ValueError(f"Order {self.order_id} cannot be marked shipped from status '{self.status}'.")
def cancel_order(self):
if self._status in [self.Status.PENDING, self.Status.PLACED, self.Status.PAID]:
self._status = self.Status.CANCELLED
print(f"Order {self.order_id} has been cancelled.")
else:
raise ValueError(f"Order {self.order_id} cannot be cancelled from status '{self.status}'.")
def __eq__(self, other):
if not isinstance(other, Order): return NotImplemented
return self.order_id == other.order_id
def __hash__(self): return hash(self.order_id)
def __str__(self):
items_str = "\n ".join(str(item) for item in self._line_items)
return (f"Order ID: {self.order_id}\n"
f" Customer ID: {self.customer_id}\n"
f" Date: {self.order_date.isoformat()}\n"
f" Status: {self.status}\n"
f" Total: {self.total_price if self.total_price else 'N/A'}\n"
f" Line Items:\n {items_str if items_str else 'No items'}")
def __repr__(self):
return (f"Order(order_id='{self.order_id}', customer_id='{self.customer_id}', "
f"order_date={self.order_date!r}, status='{self.status}', "
f"line_items={self._line_items!r})")
# --- Domain Layer: Services (Interfaces/Abstracts) ---
class OrderRepository(ABC):
@abstractmethod
def get_by_id(self, order_id: str) -> Optional[Order]: pass
@abstractmethod
def save(self, order: Order): pass
@abstractmethod
def delete(self, order_id: str): pass
class CustomerRepository(ABC):
@abstractmethod
def get_by_id(self, customer_id: str) -> Optional[Customer]: pass
@abstractmethod
def save(self, customer: Customer): pass
class OrderPlacementService:
def __init__(self, order_repo: OrderRepository, customer_repo: CustomerRepository):
self._order_repo = order_repo
self._customer_repo = customer_repo
def place_new_order(self, customer_id: str, product_data: List[dict]) -> Order:
customer = self._customer_repo.get_by_id(customer_id)
if not customer: raise ValueError(f"Customer with ID {customer_id} not found.")
new_order_id = str(uuid.uuid4())
order = Order(order_id=new_order_id, customer_id=customer_id, order_date=datetime.now())
for item_data in product_data:
product_id = item_data.get('product_id')
product_name = item_data.get('name')
unit_price_amount = decimal.Decimal(str(item_data.get('unit_price')))
unit_price_currency = item_data.get('currency', 'USD')
quantity_value = item_data.get('quantity')
if not all([product_id, product_name, unit_price_amount, quantity_value is not None]):
raise ValueError("Invalid product data provided for order line item.")
product_info = ProductInfo(
product_id=product_id,
name=product_name,
unit_price=Money(unit_price_amount, unit_price_currency)
)
quantity = Quantity(quantity_value)
line_item = OrderLineItem(product_info=product_info, quantity=quantity)
order.add_line_item(line_item)
order.place_order()
self._order_repo.save(order)
print(f"Order {order.order_id} successfully placed for customer {customer.name}.")
return order
# --- Infrastructure Layer: In-Memory Repositories ---
class InMemoryOrderRepository(OrderRepository):
def __init__(self): self._orders: dict[str, Order] = {}
def get_by_id(self, order_id: str) -> Optional[Order]:
print(f"Retrieving order {order_id} from in-memory repository.")
return self._orders.get(order_id)
def save(self, order: Order):
print(f"Saving order {order.order_id} to in-memory repository.")
self._orders[order.order_id] = order
def delete(self, order_id: str):
print(f"Deleting order {order_id} from in-memory repository.")
if order_id in self._orders: del self._orders[order_id]
class InMemoryCustomerRepository(CustomerRepository):
def __init__(self): self._customers: dict[str, Customer] = {}
def get_by_id(self, customer_id: str) -> Optional[Customer]:
print(f"Retrieving customer {customer_id} from in-memory repository.")
return self._customers.get(customer_id)
def save(self, customer: Customer):
print(f"Saving customer {customer.customer_id} to in-memory repository.")
self._customers[customer.customer_id] = customer
# --- Application Layer: Entry Point / Orchestration ---
def run_order_management_example():
print("--- Starting Order Management System Example ---")
# 1. Setup Repositories (Infrastructure Layer)
order_repo = InMemoryOrderRepository()
customer_repo = InMemoryCustomerRepository()
# 2. Create some initial customers (simulating data setup)
customer1_id = str(uuid.uuid4())
customer2_id = str(uuid.uuid4())
customer1 = Customer(customer1_id, "Alice Wonderland", "alice@example.com")
customer2 = Customer(customer2_id, "Bob The Builder", "bob@example.com")
customer_repo.save(customer1)
customer_repo.save(customer2)
print("\n--- Initial Customers Created ---")
print(customer_repo.get_by_id(customer1_id))
print(customer_repo.get_by_id(customer2_id))
# 3. Initialize Domain Service (Domain Layer, injected with Infrastructure dependencies)
order_placement_service = OrderPlacementService(order_repo, customer_repo)
# 4. Simulate placing an order for Customer 1
print("\n--- Placing Order 1 for Alice ---")
product_data_1 = [
{'product_id': 'P001', 'name': 'Laptop Pro', 'unit_price': 1200.00, 'currency': 'USD', 'quantity': 1},
{'product_id': 'P005', 'name': 'Wireless Mouse', 'unit_price': 25.50, 'currency': 'USD', 'quantity': 2},
]
try:
order1 = order_placement_service.place_new_order(customer1_id, product_data_1)
print("\n--- Order 1 Details ---")
print(order1)
print(f"Order 1 Total Price: {order1.total_price}")
# Simulate further actions on the order
order1.mark_as_paid()
order1.mark_as_shipped()
print(f"Order 1 Status after shipping: {order1.status}")
except ValueError as e:
print(f"Error placing order for Alice: {e}")
# 5. Simulate placing another order for Customer 2
print("\n--- Placing Order 2 for Bob ---")
product_data_2 = [
{'product_id': 'P010', 'name': 'Mechanical Keyboard', 'unit_price': 150.00, 'currency': 'USD', 'quantity': 1},
]
try:
order2 = order_placement_service.place_new_order(customer2_id, product_data_2)
print("\n--- Order 2 Details ---")
print(order2)
print(f"Order 2 Total Price: {order2.total_price}")
# Simulate cancelling the order
order2.cancel_order()
print(f"Order 2 Status after cancellation: {order2.status}")
except ValueError as e:
print(f"Error placing order for Bob: {e}")
# 6. Attempt to modify a placed order (should fail due to domain rules)
print("\n--- Attempting to modify placed Order 1 ---")
try:
# Retrieve the order again to simulate a new session or process
retrieved_order1 = order_repo.get_by_id(order1.order_id)
if retrieved_order1:
retrieved_order1.add_line_item(
OrderLineItem(
ProductInfo('P002', 'Monitor', Money(decimal.Decimal('300.00'), 'USD')),
Quantity(1)
)
)
except ValueError as e:
print(f"Successfully prevented modification: {e}")
print("\n--- End of Order Management System Example ---")
if __name__ == "__main__":
run_order_management_example()
--------------------------------------------------------------------You've got it! Here is the article reprinted with plain ASCII formatting, no unnecessary line breaks, proper indentation of code, and no ASCII graphics characters.
Unlocking Domain-Driven Design with LLM Chatbots: A Comprehensive Guide
======================================================================
In the intricate landscape of modern software development, Domain-Driven Design (DDD) stands as a powerful methodology for tackling complex business domains. It emphasizes a deep understanding of the core business, fostering a shared language between domain experts and developers, and structuring software to reflect the domain model. However, the initial phases of DDD, particularly the elicitation of the Ubiquitous Language and the identification of Bounded Contexts, can be time-consuming and require significant human interaction and expertise. This article explores how Large Language Model (LLM) chatbots can serve as a revolutionary tool to assist and accelerate the DDD process, whether powered by local or remote LLMs. We will delve into the benefits and liabilities of this approach, broadly and deeply cover all levels of DDD, and illustrate concepts with practical examples.
Introduction to Domain-Driven Design and LLMs
---------------------------------------------
Domain-Driven Design, pioneered by Eric Evans, is an approach to software development that focuses on modeling software to match a domain according to input from domain experts. Its core tenets include building a Ubiquitous Language, defining Bounded Contexts, and crafting rich domain models. The goal is to create software that is aligned with the business, adaptable to change, and maintainable over time.
Large Language Models, on the other hand, are advanced AI systems capable of understanding, generating, and processing human language. Their ability to synthesize information, engage in conversational dialogue, and even generate code makes them a compelling candidate for augmenting human intelligence in complex tasks. By combining the structured thinking of DDD with the conversational power of LLMs, we can create intelligent assistants that guide developers and domain experts through the DDD journey, making it more accessible and efficient.
Benefits of an LLM Chatbot for Domain-Driven Design
Integrating an LLM chatbot into the DDD workflow offers several compelling advantages, streamlining processes and enhancing collaboration.
An LLM chatbot can actively participate in conversations with domain experts and developers, identifying key terms, concepts, and synonyms, thereby accelerating Ubiquitous Language Development. It can then propose a consistent Ubiquitous Language, flagging ambiguities and suggesting clear definitions, which significantly speeds up the consensus-building phase. The chatbot acts as a tireless scribe and an intelligent thesaurus, ensuring that all stakeholders speak the same domain-specific language. By analyzing discussions, user stories, and existing documentation, an LLM can help identify natural boundaries within the domain, leading to Enhanced Bounded Context Identification. It can suggest potential Bounded Contexts by grouping related concepts and functionalities, and even propose relationships between these contexts, aiding in the creation of initial Context Maps, which helps teams to avoid monolithic designs and promotes a modular architecture.
The chatbot acts as an intelligent intermediary, translating complex business jargon into technical terms and vice versa, which results in Improved Communication Between Domain Experts and Developers. It can clarify misunderstandings, elaborate on concepts, and ensure that both parties have a shared understanding of the domain model, fostering a more productive and less error-prone communication channel. Based on conversational input and defined Ubiquitous Language, an LLM can generate initial drafts of DDD artifacts, providing Automated Generation of Initial DDD Artifacts. This includes basic class definitions for Entities, Value Objects, Aggregates, and even skeletal Domain Services, offering a concrete starting point for development teams, reducing boilerplate code, and allowing developers to focus on refining the domain logic.
When faced with design dilemmas, the chatbot can propose various DDD patterns and architectural choices, explaining their pros and cons in the context of the specific domain problem, thus providing Support for Exploring Design Alternatives. It can simulate potential impacts of different design decisions, helping teams make informed choices without extensive manual analysis. For teams new to DDD or those needing a quick refresher, the chatbot can serve as an always-available DDD expert, offering On-Demand DDD Expertise. It can explain complex patterns, provide examples, and answer questions about best practices, effectively democratizing access to DDD knowledge and reducing the learning curve.
Liabilities of an LLM Chatbot for Domain-Driven Design
While the benefits are substantial, it is crucial to acknowledge the potential drawbacks and challenges associated with relying on LLM chatbots for DDD.
LLMs can sometimes generate plausible but incorrect information, a phenomenon known as hallucination, posing a Risk of Hallucination and Inaccuracies. If not carefully validated by human experts, these inaccuracies could lead to flawed domain models and ultimately, incorrect software implementations, undermining the very purpose of DDD. The effectiveness of the chatbot is directly tied to the quality of the underlying LLM and its training data, indicating a Dependence on LLM Quality and Training Data. If the model lacks sufficient understanding of software engineering principles, DDD patterns, or the specific business domain, its suggestions may be suboptimal or irrelevant.
When using remote LLMs (e.g., cloud-based APIs), sensitive domain information must be transmitted to external servers, raising Data Privacy and Security Concerns (Especially with Remote LLMs). This is particularly critical for proprietary business logic or regulated industries, requiring careful consideration of data governance and compliance. An excessive dependence on the chatbot might reduce the critical thinking and problem-solving skills of developers and domain experts, potentially Stifling Human Creativity and Critical Thinking. The human element of deep domain exploration, nuanced understanding, and innovative design should always remain central, with the LLM acting as an assistant, not a replacement.
Integrating an LLM chatbot into existing development workflows can be complex, representing an Integration Complexity (Local vs. Remote). Choosing between local and remote LLMs involves trade-offs in terms of infrastructure, maintenance, and API management, each presenting its own set of technical challenges. Remote LLMs typically operate on a pay-per-use model, which can accumulate significant costs, especially during intensive DDD phases, leading to Cost Implications. Local LLMs, while avoiding per-token costs, require substantial upfront investment in powerful hardware and ongoing maintenance, making cost a critical factor in decision-making.
Core Concepts of Domain-Driven Design (DDD) and LLM Integration
Let us now explore how LLM chatbots can specifically assist with the fundamental building blocks of DDD.
Ubiquitous Language: This is the shared language developed by domain experts and developers, used consistently in all communication and within the code itself. An LLM chatbot can facilitate its creation by listening to conversations, extracting key nouns and verbs, and proposing a glossary of terms. For example, if a domain expert describes "a customer placing an order," the LLM can suggest 'Customer' and 'Order' as core entities and 'PlaceOrder' as a potential domain service operation. It can also identify synonyms (e.g., "client" vs. "customer") and help standardize terminology.
Bounded Contexts: A Bounded Context defines a logical boundary within which a specific domain model is applicable, and terms within that context have a precise, unambiguous meaning. The LLM can analyze functional requirements and discussions to identify natural groupings of responsibilities. For instance, in an e-commerce system, the LLM might suggest distinct Bounded Contexts like 'Sales', 'Inventory', 'Shipping', and 'Billing', based on how different aspects of the business are discussed and managed. It can then help define the interfaces and communication patterns between these contexts.
Entities: An Entity is an object defined by its identity, rather than its attributes. It has a life cycle and typically represents a distinct thing in the domain that needs to be tracked. An LLM, given a description of a business process, can identify potential entities. For an "Order Management System," the chatbot might suggest 'Order' and 'Customer' as entities. It can then propose initial attributes and methods based on common patterns.
--------------------------------------------------------------------
# Example: Initial Entity suggestion for an 'Order'
# Chatbot output based on user input: "Describe an order in our system."
class Order:
def __init__(self, order_id: str, customer_id: str, order_date: datetime):
self.order_id = order_id
self.customer_id = customer_id
self.order_date = order_date
self.line_items = [] # List of LineItem objects
self.status = "PENDING" # Initial status
def add_line_item(self, line_item):
# Logic to add a line item to the order
self.line_items.append(line_item)
def calculate_total(self) -> float:
# Logic to calculate the total price of the order
return sum(item.calculate_subtotal() for item in self.line_items)
def place_order(self):
# Logic to change order status to 'PLACED' and trigger events
if self.status == "PENDING":
self.status = "PLACED"
print(f"Order {self.order_id} has been placed.")
# Potentially publish an OrderPlaced event here
else:
print(f"Order {self.order_id} cannot be placed from status {self.status}.")
# This is a basic structure. An LLM can generate this and suggest further refinements.
--------------------------------------------------------------------
Value Objects: A Value Object is an object that describes a characteristic or attribute but has no conceptual identity. It is immutable and compared by its attributes. Examples include 'Money', 'Address', or 'Quantity'. An LLM can help identify these by looking for concepts that are typically exchanged or measured, and whose identity is not important, only their value.
--------------------------------------------------------------------
# Example: Value Object suggestion for 'Money'
# Chatbot output based on user input: "How do we handle currency and amounts?"
import decimal
class Money:
def __init__(self, amount: decimal.Decimal, currency: str):
if not isinstance(amount, decimal.Decimal):
raise TypeError("Amount must be a Decimal.")
if not isinstance(currency, str) or len(currency) != 3:
raise ValueError("Currency must be a 3-letter string (e.g., 'USD').")
self._amount = amount
self._currency = currency.upper()
@property
def amount(self) -> decimal.Decimal:
return self._amount
@property
def currency(self) -> str:
return self._currency
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("Cannot add Money with different currencies.")
return Money(self.amount + other.amount, self.currency)
def subtract(self, other: 'Money') -> 'Money':
if self.currency != other.currency:
raise ValueError("Cannot subtract Money with different currencies.")
return Money(self.amount - other.amount, self.currency)
def __eq__(self, other):
if not isinstance(other, Money):
return NotImplemented
return self.amount == other.amount and self.currency == other.currency
def __hash__(self):
return hash((self.amount, self.currency))
def __str__(self):
return f"{self.currency} {self.amount:.2f}"
def __repr__(self):
return f"Money(amount=Decimal('{self.amount}'), currency='{self.currency}')"
# This immutable class ensures monetary calculations are handled consistently.
--------------------------------------------------------------------
Aggregates: An Aggregate is a cluster of associated objects treated as a unit for data changes. It has an Aggregate Root, which is an Entity that controls access to other objects within the aggregate and ensures the aggregate's consistency. An LLM can assist in defining these consistency boundaries. For an 'Order', the LLM might suggest that 'Line Items' belong within the 'Order' aggregate, and that 'Customer' is a separate aggregate, merely referenced by 'Order' via a 'CustomerId'. It can also help define invariants that must always hold true for the aggregate.
Domain Services: A Domain Service is a stateless operation that performs significant domain logic that does not naturally fit within an Entity or Value Object. An LLM can identify complex business operations that involve multiple aggregates or external systems and suggest them as candidates for Domain Services. For example, 'PlaceOrderService' might be suggested as a service responsible for orchestrating the creation of an order, checking inventory, and publishing an 'OrderPlaced' event.
Domain Events: A Domain Event represents something that happened in the domain that domain experts care about. They are immutable facts about past occurrences. The LLM can help identify these by analyzing phrases like "when X happens," "after Y occurs," or "upon Z completion." For an 'Order' aggregate, 'OrderPlaced', 'OrderShipped', or 'PaymentReceived' could be identified as crucial domain events. The chatbot can also suggest the data that should be included in each event.
Repositories: A Repository provides a mechanism for encapsulating storage, retrieval, and search behavior that mimics a collection of objects. It abstracts the persistence layer from the domain model. An LLM can suggest repository interfaces for each aggregate root, outlining methods like 'get_by_id', 'save', and 'delete'.
Modules: Modules are used to organize the domain model into logical groupings, often aligning with Bounded Contexts or subdomains. The LLM can propose module structures based on the identified Bounded Contexts and the relationships between them, helping to maintain a clean and understandable codebase.
DDD Patterns and Diagrams with LLM Assistance
LLM chatbots can extend their utility beyond core concepts to assist with both strategic and tactical DDD patterns, and even help in generating representations like diagrams.
Strategic Design: This level of DDD focuses on understanding the overall business domain and dividing it into Bounded Contexts.
Context Mapping: This pattern visually represents the relationships and integrations between different Bounded Contexts. An LLM can analyze descriptions of business processes and system interactions to suggest a textual representation of a Context Map. For example, the LLM might describe a relationship where the 'Sales Context' acts as a Customer to the 'Inventory Context' which acts as a Supplier. This implies that Sales depends on Inventory for product availability. It might also note an 'Anti-Corruption Layer' protecting the Sales Context from the Inventory Context's model. The LLM can help define the nature of the relationship (e.g., Shared Kernel, Conformist, Anti-Corruption Layer) and suggest the necessary integration mechanisms.
Subdomains: The LLM can help categorize subdomains within a larger business domain as 'Core', 'Supporting', or 'Generic', based on their strategic importance and uniqueness to the business. For an e-commerce system, 'Order Fulfillment' might be identified as a core subdomain, while 'User Authentication' would be a generic subdomain.
Tactical Design: This level focuses on the detailed implementation of the domain model within a Bounded Context, using patterns like Aggregates, Entities, Value Objects, Domain Services, Repositories, and Factories. An LLM can generate initial code structures for these patterns, complete with comments and basic documentation, based on the Ubiquitous Language and domain descriptions provided.
Diagrams: While LLMs are primarily text-based, they can assist in creating or interpreting diagrammatic representations.
Event Storming: This collaborative workshop technique identifies domain events, commands, and aggregates. An LLM can "simulate" an Event Storming session by asking guiding questions and summarizing identified events and commands, or by analyzing transcripts of actual sessions to extract key elements.
UML (Unified Modeling Language): An LLM can generate basic UML class diagrams or sequence diagrams in a textual format (e.g., PlantUML syntax, or a simplified ASCII representation) from a textual description of entities, their relationships, and interactions. This provides a structured visual aid for understanding the domain model.
Constituents of an LLM Chatbot for DDD
Building an effective LLM chatbot for DDD involves several key components working in concert.
The User Interface (UI) is the primary interaction point for developers and domain experts, which can be a web-based chat application, a desktop client, or a command-line interface (CLI). The UI needs to be intuitive, allowing users to ask questions, provide domain descriptions, and receive structured responses. The LLM Integration Layer handles the communication with the underlying Large Language Model. For remote LLMs, it involves making API calls (e.g., HTTP requests) to services like OpenAI GPT, Anthropic Claude, or Google Gemini, while for local LLMs, it entails loading the model into memory and interacting with it via local libraries such as 'ollama', 'transformers', or 'llama.cpp'. This layer manages authentication, request formatting, and response parsing.
The Prompt Engineering Module is a critical component responsible for crafting effective prompts that guide the LLM to produce relevant and accurate DDD-specific outputs. It translates user queries into detailed instructions for the LLM, often including context, desired output format, and specific DDD principles to adhere to. For example, if a user asks "Define the Order aggregate," the prompt engineering module might prepend instructions like "Act as a DDD expert. Define the Order aggregate, including its root, invariants, and contained entities/value objects. Provide Python code." To maintain a coherent and useful conversation, the chatbot needs to remember past interactions and the evolving DDD context, which is managed by the Context Management/Memory module. This module stores conversation history, identified Ubiquitous Language terms, Bounded Context definitions, and other relevant domain knowledge, allowing the LLM to build upon previous statements and provide contextually aware responses, avoiding repetitive information or contradictory suggestions.
To enhance accuracy and reduce hallucinations, the chatbot can be augmented with a knowledge base containing authoritative DDD resources, project-specific documentation, or examples, leveraging a Knowledge Base/Retrieval Augmented Generation (RAG) module. The RAG module retrieves relevant information from this knowledge base based on the user's query and injects it into the LLM's prompt, ensuring that the LLM's responses are grounded in factual and specific information, rather than solely relying on its general training data. LLMs often produce free-form text, so the Output Parser/Formatter module is responsible for parsing the LLM's response and formatting it into a structured, usable output. This could involve extracting code blocks, identifying key terms for the Ubiquitous Language glossary, or converting textual descriptions into a format suitable for diagram generation tools. For instance, it might extract a Python class definition from the LLM's response and present it cleanly.
An essential component for continuous improvement is the Feedback Loop, which allows users to correct inaccuracies, refine suggestions, or rate the usefulness of the chatbot's responses. This feedback can then be used to fine-tune prompts, update the knowledge base, or even retrain the LLM (if using a local model), leading to increasingly accurate and helpful interactions over time.
LLM Choices: Local vs. Remote
The decision to use a local or remote LLM significantly impacts the chatbot's architecture, performance, cost, and data handling.
Remote LLMs are cloud-hosted models provided by vendors like OpenAI (GPT series), Anthropic (Claude), or Google (Gemini). Their advantages include High Performance and Scale, as they are typically hosted on powerful infrastructure, offering high performance and the ability to handle a large volume of requests without requiring local hardware investment. They also provide access to Large Models, as vendors often provide access to their largest and most capable models, which generally exhibit superior understanding and generation capabilities. Additionally, they offer Easy Access and Maintenance, as integration is usually straightforward via well-documented APIs, and the vendor handles all infrastructure, updates, and maintenance. However, there are drawbacks: Cost, as usage is typically billed per token, which can become expensive, especially for extensive use or large context windows. Data Privacy and Security are concerns, as sending sensitive domain data to a third-party cloud service raises privacy and security issues, especially for proprietary business logic or regulated industries, requiring careful review of vendor policies and compliance certifications. Internet Dependency is another factor, as a stable internet connection is mandatory for operation, and Vendor Lock-in can occur, as relying heavily on a single vendor's API can create dependency and make switching providers challenging. Integration typically involves HTTP requests to a RESTful API, authenticated with API keys.
Local LLMs are models that run directly on an organization's own hardware, such as Llama 2, Mistral, Falcon, or Phi-3. Their advantages include Data Privacy and Security, as sensitive data remains within the organization's control, addressing privacy concerns. There is No Internet Dependency, as once downloaded, the models can operate entirely offline. They can be Cost-Effective (after initial setup), as while requiring upfront hardware investment, there are no per-token costs, making them potentially more economical for high-volume internal use. Organizations also have Full Control and Customization over the model, allowing for fine-tuning with specific domain data and custom modifications. However, there are disadvantages: Hardware Requirements, as running powerful LLMs locally demands significant computational resources (GPUs, ample RAM), which can be a substantial upfront investment. Performance Variability can be an issue, as performance can vary greatly depending on the local hardware and the chosen model's size, with smaller models potentially having reduced capabilities. Model Management falls on the organization, as they are responsible for managing model deployment, updates, and infrastructure maintenance. Finally, Context Window Limitations mean that many local models, especially smaller ones, have more restrictive context windows compared to their remote counterparts, limiting the amount of conversation history they can process at once. Integration involves using libraries like 'ollama', 'transformers', or 'llama.cpp' to load and interact with the model programmatically.
Building the Chatbot: A Conceptual Walkthrough
Let us outline the steps to construct an LLM chatbot for DDD, using our 'Order Management System' as a running example.
Step 1: Define the Goal. Clearly articulate what specific DDD tasks the chatbot will assist with. For our Order Management System, the goal is to help define the Ubiquitous Language, identify Bounded Contexts, and generate initial code for core domain entities like 'Order' and 'LineItem'.
Step 2: Choose LLM Strategy. Decide between a local or remote LLM. For a proof-of-concept or if data privacy is not a primary concern initially, a remote LLM like GPT-4 might be chosen for its superior capabilities. If the system needs to handle highly sensitive order data, a local LLM running on internal servers would be preferred. Let us assume we start with a remote LLM for ease of initial development.
Step 3: Design the Interaction Flow. Determine how users will interact with the chatbot. An example flow could be: User: "Help me define the core concepts for an Order Management System." Chatbot: "Certainly! Let's start with the Ubiquitous Language. What are the key nouns and verbs related to orders?" User: "Customers place orders, orders contain line items, products are in line items, orders are shipped." Chatbot: "Excellent. I suggest 'Customer', 'Order', 'LineItem', 'Product', and 'Shipment' as core terms. How do we define an 'Order'?" This flow guides the user through DDD concepts.
Step 4: Implement Core Components.
* UI: A simple web interface using a framework like Flask or FastAPI for the backend and a basic HTML/JavaScript frontend for chat.
* LLM Integration: Use the chosen LLM's Python SDK (e.g., 'openai' library) to send prompts and receive responses.
* Prompt Engineering: Develop a set of initial system prompts and user prompt templates.
--------------------------------------------------------------------
# Example: Basic prompt for defining an entity
def generate_entity_prompt(entity_name: str, domain_description: str) -> str:
return f"""
You are an expert in Domain-Driven Design (DDD).
Your task is to help define the core domain model for an Order Management System.
Based on the following domain description:
"{domain_description}"
Please define the DDD Entity '{entity_name}'.
Provide its key attributes, identify its unique identity, and suggest initial methods.
Explain why it is an Entity (identity-based) rather than a Value Object.
Provide a Python class definition for this entity, including type hints and docstrings.
Ensure the code follows clean code principles.
"""
--------------------------------------------------------------------
Step 5: Incorporate DDD Knowledge (RAG). Create a small knowledge base of common DDD patterns, definitions, and best practices. When the user asks about "Aggregates," the RAG system can retrieve the definition and principles of aggregates and inject them into the LLM's prompt, ensuring the LLM's explanation is accurate and comprehensive.
Step 6: Iterate and Refine. Continuously test the chatbot with various DDD scenarios. Gather feedback from developers and domain experts. If the LLM hallucinates or provides unhelpful suggestions, refine the prompts, update the RAG knowledge base, or consider fine-tuning the model (if local) or exploring different LLM parameters. For example, if the LLM consistently forgets to include an 'order_date' in the 'Order' entity, the prompt can be explicitly updated to "ensure 'order_date' is included."
Conclusion
The integration of LLM chatbots into the Domain-Driven Design process presents a significant opportunity to enhance efficiency, foster collaboration, and democratize access to DDD expertise. From accelerating the development of a Ubiquitous Language and identifying Bounded Contexts to generating initial code artifacts, LLMs can act as powerful assistants, guiding teams through the complexities of domain modeling. However, it is imperative to approach this technology with a clear understanding of its liabilities, particularly concerning potential inaccuracies, data privacy, and the risk of over-reliance. The human element of critical thinking, deep domain understanding, and creative problem-solving must always remain at the core of DDD. By strategically leveraging LLM chatbots as intelligent tools, not replacements, organizations can unlock new levels of productivity and build more robust, business-aligned software systems.
Addendum: Full Running Example Code for Order Management System
This addendum provides a more complete, albeit simplified, Python implementation of the core domain model for an Order Management System, demonstrating the DDD concepts discussed in the article. This code is designed to be runnable and illustrates the structure an LLM chatbot could help generate and refine.
--------------------------------------------------------------------
# order_management_system/domain/value_objects.py
import decimal
from typing import NamedTuple
# A Value Object representing monetary amounts.
# It is immutable and compared by its value (amount and currency).
class Money:
def __init__(self, amount: decimal.Decimal, currency: str):
if not isinstance(amount, decimal.Decimal):
raise TypeError("Amount must be a Decimal.")
if not isinstance(currency, str) or len(currency) != 3:
raise ValueError("Currency must be a 3-letter string (e.g., 'USD').")
# Store as private attributes to enforce immutability
self._amount = amount
self._currency = currency.upper()
@property
def amount(self) -> decimal.Decimal:
return self._amount
@property
def currency(self) -> str:
return self._currency
def add(self, other: 'Money') -> 'Money':
"""Adds another Money object to this one, returning a new Money object."""
if self.currency != other.currency:
raise ValueError("Cannot add Money with different currencies.")
return Money(self.amount + other.amount, self.currency)
def multiply(self, factor: decimal.Decimal) -> 'Money':
"""Multiplies the Money amount by a factor, returning a new Money object."""
if not isinstance(factor, decimal.Decimal):
raise TypeError("Factor must be a Decimal.")
return Money(self.amount * factor, self.currency)
def __eq__(self, other):
"""Compares two Money objects for equality based on amount and currency."""
if not isinstance(other, Money):
return NotImplemented
return self.amount == other.amount and self.currency == other.currency
def __hash__(self):
"""Generates a hash for the Money object, allowing it to be used in sets/dicts."""
return hash((self.amount, self.currency))
def __str__(self):
"""Returns a user-friendly string representation."""
return f"{self.currency} {self.amount:.2f}"
def __repr__(self):
"""Returns a developer-friendly string representation."""
return f"Money(amount=Decimal('{self.amount}'), currency='{self.currency}')"
# A Value Object representing a quantity of items.
# Also immutable and compared by its value.
class Quantity:
def __init__(self, value: int):
if not isinstance(value, int) or value < 0:
raise ValueError("Quantity must be a non-negative integer.")
self._value = value
@property
def value(self) -> int:
return self._value
def add(self, other: 'Quantity') -> 'Quantity':
"""Adds another Quantity object, returning a new Quantity."""
return Quantity(self.value + other.value)
def __eq__(self, other):
"""Compares two Quantity objects for equality."""
if not isinstance(other, Quantity):
return NotImplemented
return self.value == other.value
def __hash__(self):
"""Generates a hash for the Quantity object."""
return hash(self.value)
def __str__(self):
"""Returns a user-friendly string representation."""
return str(self.value)
def __repr__(self):
"""Returns a developer-friendly string representation."""
return f"Quantity(value={self.value})"
# A Value Object representing a product reference within an order line item.
# It's immutable and identified by its product ID and name.
class ProductInfo(NamedTuple):
product_id: str
name: str
unit_price: Money
def __str__(self):
return f"{self.name} ({self.product_id}) @ {self.unit_price}"
# A Value Object representing a single line item within an order.
# It contains product information, quantity, and calculates its subtotal.
class OrderLineItem:
def __init__(self, product_info: ProductInfo, quantity: Quantity):
if not isinstance(product_info, ProductInfo):
raise TypeError("product_info must be a ProductInfo Value Object.")
if not isinstance(quantity, Quantity):
raise TypeError("quantity must be a Quantity Value Object.")
self._product_info = product_info
self._quantity = quantity
@property
def product_info(self) -> ProductInfo:
return self._product_info
@property
def quantity(self) -> Quantity:
return self._quantity
def calculate_subtotal(self) -> Money:
"""Calculates the subtotal for this line item."""
return self.product_info.unit_price.multiply(decimal.Decimal(self.quantity.value))
def __eq__(self, other):
"""Compares two OrderLineItem objects for equality."""
if not isinstance(other, OrderLineItem):
return NotImplemented
return self.product_info == other.product_info and self.quantity == other.quantity
def __hash__(self):
"""Generates a hash for the OrderLineItem object."""
return hash((self.product_info, self.quantity))
def __str__(self):
return (f"{self.quantity.value} x {self.product_info.name} "
f"({self.product_info.product_id}) - {self.calculate_subtotal()}")
def __repr__(self):
return (f"OrderLineItem(product_info={self.product_info!r}, "
f"quantity={self.quantity!r})")
# order_management_system/domain/entities.py
import uuid
from datetime import datetime
from typing import List, Optional
# Assuming Value Objects are imported from the same domain package
# from .value_objects import Money, Quantity, ProductInfo, OrderLineItem
# An Entity representing a Customer.
# It has a unique identity (customer_id) and a lifecycle.
class Customer:
def __init__(self, customer_id: str, name: str, email: str):
if not customer_id:
raise ValueError("Customer ID cannot be empty.")
if not name:
raise ValueError("Customer name cannot be empty.")
if not email:
raise ValueError("Customer email cannot be empty.")
self._customer_id = customer_id
self._name = name
self._email = email
@property
def customer_id(self) -> str:
return self._customer_id
@property
def name(self) -> str:
return self._name
@property
def email(self) -> str:
return self._email
def update_email(self, new_email: str):
"""Updates the customer's email address."""
if not new_email:
raise ValueError("New email cannot be empty.")
self._email = new_email
print(f"Customer {self.customer_id} email updated to {new_email}.")
# Potentially publish a CustomerEmailUpdated event here
def __eq__(self, other):
"""Compares two Customer objects based on their identity."""
if not isinstance(other, Customer):
return NotImplemented
return self.customer_id == other.customer_id
def __hash__(self):
"""Generates a hash for the Customer object based on its identity."""
return hash(self.customer_id)
def __str__(self):
return f"Customer(ID: {self.customer_id}, Name: {self.name}, Email: {self.email})"
def __repr__(self):
return (f"Customer(customer_id='{self.customer_id}', name='{self.name}', "
f"email='{self.email}')")
# order_management_system/domain/aggregates.py
# Assuming Value Objects and Entities are imported
# from .value_objects import Money, Quantity, ProductInfo, OrderLineItem
# from .entities import Customer # Customer is a separate aggregate, referenced by ID
# An Aggregate Root representing an Order.
# It encapsulates OrderLineItems and ensures the consistency of the order.
class Order:
class Status:
PENDING = "PENDING"
PLACED = "PLACED"
PAID = "PAID"
SHIPPED = "SHIPPED"
CANCELLED = "CANCELLED"
def __init__(self, order_id: str, customer_id: str, order_date: datetime):
if not order_id:
raise ValueError("Order ID cannot be empty.")
if not customer_id:
raise ValueError("Customer ID cannot be empty.")
if not isinstance(order_date, datetime):
raise TypeError("Order date must be a datetime object.")
self._order_id = order_id
self._customer_id = customer_id
self._order_date = order_date
self._line_items: List[OrderLineItem] = []
self._status = self.Status.PENDING
self._total_price: Optional[Money] = None # Calculated on demand or when placed
@property
def order_id(self) -> str:
return self._order_id
@property
def customer_id(self) -> str:
return self._customer_id
@property
def order_date(self) -> datetime:
return self._order_date
@property
def line_items(self) -> List[OrderLineItem]:
# Return a copy to prevent external modification of the internal list
return list(self._line_items)
@property
def status(self) -> str:
return self._status
@property
def total_price(self) -> Optional[Money]:
"""Calculates the total price of the order from its line items."""
if not self._line_items:
return None
# Assuming all line items have the same currency as the first one
# In a real system, more robust currency handling would be needed.
currency = self._line_items[0].product_info.unit_price.currency
total = Money(decimal.Decimal('0.00'), currency)
for item in self._line_items:
total = total.add(item.calculate_subtotal())
self._total_price = total # Cache the total price
return self._total_price
def add_line_item(self, line_item: OrderLineItem):
"""Adds a new line item to the order."""
if self.status != self.Status.PENDING:
raise ValueError(f"Cannot add items to an order with status '{self.status}'.")
self._line_items.append(line_item)
# Invalidate cached total price
self._total_price = None
print(f"Added line item to order {self.order_id}: {line_item}")
def remove_line_item(self, product_id: str):
"""Removes a line item from the order by product ID."""
if self.status != self.Status.PENDING:
raise ValueError(f"Cannot remove items from an order with status '{self.status}'.")
initial_count = len(self._line_items)
self._line_items = [
item for item in self._line_items if item.product_info.product_id != product_id
]
if len(self._line_items) < initial_count:
# Invalidate cached total price
self._total_price = None
print(f"Removed line item for product {product_id} from order {self.order_id}.")
else:
print(f"Product {product_id} not found in order {self.order_id}.")
def place_order(self):
"""Transitions the order status to PLACED, if currently PENDING."""
if not self._line_items:
raise ValueError("Cannot place an empty order.")
if self._status == self.Status.PENDING:
self._status = self.Status.PLACED
print(f"Order {self.order_id} has been placed.")
# Here, an OrderPlaced domain event would typically be published.
else:
raise ValueError(f"Order {self.order_id} cannot be placed from status '{self.status}'.")
def mark_as_paid(self):
"""Transitions the order status to PAID, if currently PLACED."""
if self._status == self.Status.PLACED:
self._status = self.Status.PAID
print(f"Order {self.order_id} has been marked as paid.")
# Here, a PaymentReceived domain event would typically be published.
else:
raise ValueError(f"Order {self.order_id} cannot be marked paid from status '{self.status}'.")
def mark_as_shipped(self):
"""Transitions the order status to SHIPPED, if currently PAID."""
if self._status == self.Status.PAID:
self._status = self.Status.SHIPPED
print(f"Order {self.order_id} has been marked as shipped.")
# Here, an OrderShipped domain event would typically be published.
else:
raise ValueError(f"Order {self.order_id} cannot be marked shipped from status '{self.status}'.")
def cancel_order(self):
"""Cancels the order, if not already shipped or cancelled."""
if self._status in [self.Status.PENDING, self.Status.PLACED, self.Status.PAID]:
self._status = self.Status.CANCELLED
print(f"Order {self.order_id} has been cancelled.")
# Here, an OrderCancelled domain event would typically be published.
else:
raise ValueError(f"Order {self.order_id} cannot be cancelled from status '{self.status}'.")
def __eq__(self, other):
"""Compares two Order objects based on their identity."""
if not isinstance(other, Order):
return NotImplemented
return self.order_id == other.order_id
def __hash__(self):
"""Generates a hash for the Order object based on its identity."""
return hash(self.order_id)
def __str__(self):
items_str = "\n ".join(str(item) for item in self._line_items)
return (f"Order ID: {self.order_id}\n"
f" Customer ID: {self.customer_id}\n"
f" Date: {self.order_date.isoformat()}\n"
f" Status: {self.status}\n"
f" Total: {self.total_price if self.total_price else 'N/A'}\n"
f" Line Items:\n {items_str if items_str else 'No items'}")
def __repr__(self):
return (f"Order(order_id='{self.order_id}', customer_id='{self.customer_id}', "
f"order_date={self.order_date!r}, status='{self.status}', "
f"line_items={self._line_items!r})")
# order_management_system/domain/services.py
from abc import ABC, abstractmethod
# Assuming domain models are imported
# from .aggregates import Order
# from .entities import Customer
# from .value_objects import ProductInfo, Quantity, Money
# Abstract base class for a generic Repository.
# Repositories abstract the persistence layer for aggregates.
class OrderRepository(ABC):
@abstractmethod
def get_by_id(self, order_id: str) -> Optional[Order]:
"""Retrieves an Order aggregate by its ID."""
pass
@abstractmethod
def save(self, order: Order):
"""Saves or updates an Order aggregate."""
pass
@abstractmethod
def delete(self, order_id: str):
"""Deletes an Order aggregate by its ID."""
pass
class CustomerRepository(ABC):
@abstractmethod
def get_by_id(self, customer_id: str) -> Optional[Customer]:
"""Retrieves a Customer entity by its ID."""
pass
@abstractmethod
def save(self, customer: Customer):
"""Saves or updates a Customer entity."""
pass
# A Domain Service that orchestrates the process of placing an order.
# It involves multiple aggregates (Order, potentially Inventory) and is stateless.
class OrderPlacementService:
def __init__(self, order_repo: OrderRepository, customer_repo: CustomerRepository):
self._order_repo = order_repo
self._customer_repo = customer_repo
# In a real system, an InventoryService or ProductRepository might also be injected
# to check product availability.
def place_new_order(self, customer_id: str, product_data: List[dict]) -> Order:
"""
Orchestrates the creation and placement of a new order.
Args:
customer_id (str): The ID of the customer placing the order.
product_data (List[dict]): A list of dictionaries, each containing
'product_id', 'name', 'unit_price', 'quantity'.
Returns:
Order: The newly placed Order aggregate.
Raises:
ValueError: If customer not found or product data is invalid.
"""
customer = self._customer_repo.get_by_id(customer_id)
if not customer:
raise ValueError(f"Customer with ID {customer_id} not found.")
new_order_id = str(uuid.uuid4())
order = Order(order_id=new_order_id, customer_id=customer_id, order_date=datetime.now())
for item_data in product_data:
product_id = item_data.get('product_id')
product_name = item_data.get('name')
unit_price_amount = decimal.Decimal(str(item_data.get('unit_price')))
unit_price_currency = item_data.get('currency', 'USD')
quantity_value = item_data.get('quantity')
if not all([product_id, product_name, unit_price_amount, quantity_value is not None]):
raise ValueError("Invalid product data provided for order line item.")
product_info = ProductInfo(
product_id=product_id,
name=product_name,
unit_price=Money(unit_price_amount, unit_price_currency)
)
quantity = Quantity(quantity_value)
line_item = OrderLineItem(product_info=product_info, quantity=quantity)
order.add_line_item(line_item)
order.place_order() # This changes the order status and applies domain logic
self._order_repo.save(order) # Persist the new order
print(f"Order {order.order_id} successfully placed for customer {customer.name}.")
return order
# order_management_system/infrastructure/repositories.py
# This layer handles actual persistence, e.g., to a database.
# For simplicity, we'll use in-memory repositories.
# Assuming domain models are imported
# from ..domain.aggregates import Order
# from ..domain.entities import Customer
# from ..domain.services import OrderRepository, CustomerRepository
class InMemoryOrderRepository(OrderRepository):
def __init__(self):
self._orders: dict[str, Order] = {}
def get_by_id(self, order_id: str) -> Optional[Order]:
"""Retrieves an Order from memory."""
print(f"Retrieving order {order_id} from in-memory repository.")
# Return a deep copy to prevent external modification of the stored object
# In a real system, ORM would handle this.
return self._orders.get(order_id) # Simplified: returning direct reference
def save(self, order: Order):
"""Saves or updates an Order in memory."""
print(f"Saving order {order.order_id} to in-memory repository.")
self._orders[order.order_id] = order # Simplified: storing direct reference
def delete(self, order_id: str):
"""Deletes an Order from memory."""
print(f"Deleting order {order_id} from in-memory repository.")
if order_id in self._orders:
del self._orders[order_id]
class InMemoryCustomerRepository(CustomerRepository):
def __init__(self):
self._customers: dict[str, Customer] = {}
def get_by_id(self, customer_id: str) -> Optional[Customer]:
"""Retrieves a Customer from memory."""
print(f"Retrieving customer {customer_id} from in-memory repository.")
return self._customers.get(customer_id)
def save(self, customer: Customer):
"""Saves or updates a Customer in memory."""
print(f"Saving customer {customer.customer_id} to in-memory repository.")
self._customers[customer.customer_id] = customer
# order_management_system/application/main.py
# This is the application layer, orchestrating domain and infrastructure.
# Import all necessary components
# from order_management_system.domain.value_objects import Money, Quantity, ProductInfo, OrderLineItem
# from order_management_system.domain.entities import Customer
# from order_management_system.domain.aggregates import Order
# from order_management_system.domain.services import OrderPlacementService
# from order_management_system.infrastructure.repositories import InMemoryOrderRepository, InMemoryCustomerRepository
# To make the example runnable as a single file, we'll define everything here.
# In a real project, these would be in their respective files as commented above.
import decimal
import uuid
from datetime import datetime
from typing import List, Optional, NamedTuple
from abc import ABC, abstractmethod
# --- Domain Layer: Value Objects ---
class Money:
def __init__(self, amount: decimal.Decimal, currency: str):
if not isinstance(amount, decimal.Decimal):
raise TypeError("Amount must be a Decimal.")
if not isinstance(currency, str) or len(currency) != 3:
raise ValueError("Currency must be a 3-letter string (e.g., 'USD').")
self._amount = amount
self._currency = currency.upper()
@property
def amount(self) -> decimal.Decimal: return self._amount
@property
def currency(self) -> str: return self._currency
def add(self, other: 'Money') -> 'Money':
if self.currency != other.currency: raise ValueError("Cannot add Money with different currencies.")
return Money(self.amount + other.amount, self.currency)
def multiply(self, factor: decimal.Decimal) -> 'Money':
if not isinstance(factor, decimal.Decimal): raise TypeError("Factor must be a Decimal.")
return Money(self.amount * factor, self.currency)
def __eq__(self, other):
if not isinstance(other, Money): return NotImplemented
return self.amount == other.amount and self.currency == other.currency
def __hash__(self): return hash((self.amount, self.currency))
def __str__(self): return f"{self.currency} {self.amount:.2f}"
def __repr__(self): return f"Money(amount=Decimal('{self.amount}'), currency='{self.currency}')"
class Quantity:
def __init__(self, value: int):
if not isinstance(value, int) or value < 0: raise ValueError("Quantity must be a non-negative integer.")
self._value = value
@property
def value(self) -> int: return self._value
def add(self, other: 'Quantity') -> 'Quantity': return Quantity(self.value + other.value)
def __eq__(self, other):
if not isinstance(other, Quantity): return NotImplemented
return self.value == other.value
def __hash__(self): return hash(self.value)
def __str__(self): return str(self.value)
def __repr__(self): return f"Quantity(value={self.value})"
class ProductInfo(NamedTuple):
product_id: str
name: str
unit_price: Money
def __str__(self): return f"{self.name} ({self.product_id}) @ {self.unit_price}"
class OrderLineItem:
def __init__(self, product_info: ProductInfo, quantity: Quantity):
if not isinstance(product_info, ProductInfo): raise TypeError("product_info must be a ProductInfo Value Object.")
if not isinstance(quantity, Quantity): raise TypeError("quantity must be a Quantity Value Object.")
self._product_info = product_info
self._quantity = quantity
@property
def product_info(self) -> ProductInfo: return self._product_info
@property
def quantity(self) -> Quantity: return self._quantity
def calculate_subtotal(self) -> Money:
return self.product_info.unit_price.multiply(decimal.Decimal(self.quantity.value))
def __eq__(self, other):
if not isinstance(other, OrderLineItem): return NotImplemented
return self.product_info == other.product_info and self.quantity == other.quantity
def __hash__(self): return hash((self.product_info, self.quantity))
def __str__(self):
return (f"{self.quantity.value} x {self.product_info.name} "
f"({self.product_info.product_id}) - {self.calculate_subtotal()}")
def __repr__(self):
return (f"OrderLineItem(product_info={self.product_info!r}, "
f"quantity={self.quantity!r})")
# --- Domain Layer: Entities ---
class Customer:
def __init__(self, customer_id: str, name: str, email: str):
if not customer_id: raise ValueError("Customer ID cannot be empty.")
if not name: raise ValueError("Customer name cannot be empty.")
if not email: raise ValueError("Customer email cannot be empty.")
self._customer_id = customer_id
self._name = name
self._email = email
@property
def customer_id(self) -> str: return self._customer_id
@property
def name(self) -> str: return self._name
@property
def email(self) -> str: return self._email
def update_email(self, new_email: str):
if not new_email: raise ValueError("New email cannot be empty.")
self._email = new_email
print(f"Customer {self.customer_id} email updated to {new_email}.")
def __eq__(self, other):
if not isinstance(other, Customer): return NotImplemented
return self.customer_id == other.customer_id
def __hash__(self): return hash(self.customer_id)
def __str__(self): return f"Customer(ID: {self.customer_id}, Name: {self.name}, Email: {self.email})"
def __repr__(self):
return (f"Customer(customer_id='{self.customer_id}', name='{self.name}', "
f"email='{self.email}')")
# --- Domain Layer: Aggregates ---
class Order:
class Status:
PENDING = "PENDING"
PLACED = "PLACED"
PAID = "PAID"
SHIPPED = "SHIPPED"
CANCELLED = "CANCELLED"
def __init__(self, order_id: str, customer_id: str, order_date: datetime):
if not order_id: raise ValueError("Order ID cannot be empty.")
if not customer_id: raise ValueError("Customer ID cannot be empty.")
if not isinstance(order_date, datetime): raise TypeError("Order date must be a datetime object.")
self._order_id = order_id
self._customer_id = customer_id
self._order_date = order_date
self._line_items: List[OrderLineItem] = []
self._status = self.Status.PENDING
self._total_price: Optional[Money] = None
@property
def order_id(self) -> str: return self._order_id
@property
def customer_id(self) -> str: return self._customer_id
@property
def order_date(self) -> datetime: return self._order_date
@property
def line_items(self) -> List[OrderLineItem]: return list(self._line_items)
@property
def status(self) -> str: return self._status
@property
def total_price(self) -> Optional[Money]:
if not self._line_items: return None
currency = self._line_items[0].product_info.unit_price.currency
total = Money(decimal.Decimal('0.00'), currency)
for item in self._line_items: total = total.add(item.calculate_subtotal())
self._total_price = total
return self._total_price
def add_line_item(self, line_item: OrderLineItem):
if self.status != self.Status.PENDING: raise ValueError(f"Cannot add items to an order with status '{self.status}'.")
self._line_items.append(line_item)
self._total_price = None
print(f"Added line item to order {self.order_id}: {line_item}")
def remove_line_item(self, product_id: str):
if self.status != self.Status.PENDING: raise ValueError(f"Cannot remove items from an order with status '{self.status}'.")
initial_count = len(self._line_items)
self._line_items = [item for item in self._line_items if item.product_info.product_id != product_id]
if len(self._line_items) < initial_count:
self._total_price = None
print(f"Removed line item for product {product_id} from order {self.order_id}.")
else:
print(f"Product {product_id} not found in order {self.order_id}.")
def place_order(self):
if not self._line_items: raise ValueError("Cannot place an empty order.")
if self._status == self.Status.PENDING:
self._status = self.Status.PLACED
print(f"Order {self.order_id} has been placed.")
else:
raise ValueError(f"Order {self.order_id} cannot be placed from status '{self.status}'.")
def mark_as_paid(self):
if self._status == self.Status.PLACED:
self._status = self.Status.PAID
print(f"Order {self.order_id} has been marked as paid.")
else:
raise ValueError(f"Order {self.order_id} cannot be marked paid from status '{self.status}'.")
def mark_as_shipped(self):
if self._status == self.Status.PAID:
self._status = self.Status.SHIPPED
print(f"Order {self.order_id} has been marked as shipped.")
else:
raise ValueError(f"Order {self.order_id} cannot be marked shipped from status '{self.status}'.")
def cancel_order(self):
if self._status in [self.Status.PENDING, self.Status.PLACED, self.Status.PAID]:
self._status = self.Status.CANCELLED
print(f"Order {self.order_id} has been cancelled.")
else:
raise ValueError(f"Order {self.order_id} cannot be cancelled from status '{self.status}'.")
def __eq__(self, other):
if not isinstance(other, Order): return NotImplemented
return self.order_id == other.order_id
def __hash__(self): return hash(self.order_id)
def __str__(self):
items_str = "\n ".join(str(item) for item in self._line_items)
return (f"Order ID: {self.order_id}\n"
f" Customer ID: {self.customer_id}\n"
f" Date: {self.order_date.isoformat()}\n"
f" Status: {self.status}\n"
f" Total: {self.total_price if self.total_price else 'N/A'}\n"
f" Line Items:\n {items_str if items_str else 'No items'}")
def __repr__(self):
return (f"Order(order_id='{self.order_id}', customer_id='{self.customer_id}', "
f"order_date={self.order_date!r}, status='{self.status}', "
f"line_items={self._line_items!r})")
# --- Domain Layer: Services (Interfaces/Abstracts) ---
class OrderRepository(ABC):
@abstractmethod
def get_by_id(self, order_id: str) -> Optional[Order]: pass
@abstractmethod
def save(self, order: Order): pass
@abstractmethod
def delete(self, order_id: str): pass
class CustomerRepository(ABC):
@abstractmethod
def get_by_id(self, customer_id: str) -> Optional[Customer]: pass
@abstractmethod
def save(self, customer: Customer): pass
class OrderPlacementService:
def __init__(self, order_repo: OrderRepository, customer_repo: CustomerRepository):
self._order_repo = order_repo
self._customer_repo = customer_repo
def place_new_order(self, customer_id: str, product_data: List[dict]) -> Order:
customer = self._customer_repo.get_by_id(customer_id)
if not customer: raise ValueError(f"Customer with ID {customer_id} not found.")
new_order_id = str(uuid.uuid4())
order = Order(order_id=new_order_id, customer_id=customer_id, order_date=datetime.now())
for item_data in product_data:
product_id = item_data.get('product_id')
product_name = item_data.get('name')
unit_price_amount = decimal.Decimal(str(item_data.get('unit_price')))
unit_price_currency = item_data.get('currency', 'USD')
quantity_value = item_data.get('quantity')
if not all([product_id, product_name, unit_price_amount, quantity_value is not None]):
raise ValueError("Invalid product data provided for order line item.")
product_info = ProductInfo(
product_id=product_id,
name=product_name,
unit_price=Money(unit_price_amount, unit_price_currency)
)
quantity = Quantity(quantity_value)
line_item = OrderLineItem(product_info=product_info, quantity=quantity)
order.add_line_item(line_item)
order.place_order()
self._order_repo.save(order)
print(f"Order {order.order_id} successfully placed for customer {customer.name}.")
return order
# --- Infrastructure Layer: In-Memory Repositories ---
class InMemoryOrderRepository(OrderRepository):
def __init__(self): self._orders: dict[str, Order] = {}
def get_by_id(self, order_id: str) -> Optional[Order]:
print(f"Retrieving order {order_id} from in-memory repository.")
return self._orders.get(order_id)
def save(self, order: Order):
print(f"Saving order {order.order_id} to in-memory repository.")
self._orders[order.order_id] = order
def delete(self, order_id: str):
print(f"Deleting order {order_id} from in-memory repository.")
if order_id in self._orders: del self._orders[order_id]
class InMemoryCustomerRepository(CustomerRepository):
def __init__(self): self._customers: dict[str, Customer] = {}
def get_by_id(self, customer_id: str) -> Optional[Customer]:
print(f"Retrieving customer {customer_id} from in-memory repository.")
return self._customers.get(customer_id)
def save(self, customer: Customer):
print(f"Saving customer {customer.customer_id} to in-memory repository.")
self._customers[customer.customer_id] = customer
# --- Application Layer: Entry Point / Orchestration ---
def run_order_management_example():
print("--- Starting Order Management System Example ---")
# 1. Setup Repositories (Infrastructure Layer)
order_repo = InMemoryOrderRepository()
customer_repo = InMemoryCustomerRepository()
# 2. Create some initial customers (simulating data setup)
customer1_id = str(uuid.uuid4())
customer2_id = str(uuid.uuid4())
customer1 = Customer(customer1_id, "Alice Wonderland", "alice@example.com")
customer2 = Customer(customer2_id, "Bob The Builder", "bob@example.com")
customer_repo.save(customer1)
customer_repo.save(customer2)
print("\n--- Initial Customers Created ---")
print(customer_repo.get_by_id(customer1_id))
print(customer_repo.get_by_id(customer2_id))
# 3. Initialize Domain Service (Domain Layer, injected with Infrastructure dependencies)
order_placement_service = OrderPlacementService(order_repo, customer_repo)
# 4. Simulate placing an order for Customer 1
print("\n--- Placing Order 1 for Alice ---")
product_data_1 = [
{'product_id': 'P001', 'name': 'Laptop Pro', 'unit_price': 1200.00, 'currency': 'USD', 'quantity': 1},
{'product_id': 'P005', 'name': 'Wireless Mouse', 'unit_price': 25.50, 'currency': 'USD', 'quantity': 2},
]
try:
order1 = order_placement_service.place_new_order(customer1_id, product_data_1)
print("\n--- Order 1 Details ---")
print(order1)
print(f"Order 1 Total Price: {order1.total_price}")
# Simulate further actions on the order
order1.mark_as_paid()
order1.mark_as_shipped()
print(f"Order 1 Status after shipping: {order1.status}")
except ValueError as e:
print(f"Error placing order for Alice: {e}")
# 5. Simulate placing another order for Customer 2
print("\n--- Placing Order 2 for Bob ---")
product_data_2 = [
{'product_id': 'P010', 'name': 'Mechanical Keyboard', 'unit_price': 150.00, 'currency': 'USD', 'quantity': 1},
]
try:
order2 = order_placement_service.place_new_order(customer2_id, product_data_2)
print("\n--- Order 2 Details ---")
print(order2)
print(f"Order 2 Total Price: {order2.total_price}")
# Simulate cancelling the order
order2.cancel_order()
print(f"Order 2 Status after cancellation: {order2.status}")
except ValueError as e:
print(f"Error placing order for Bob: {e}")
# 6. Attempt to modify a placed order (should fail due to domain rules)
print("\n--- Attempting to modify placed Order 1 ---")
try:
# Retrieve the order again to simulate a new session or process
retrieved_order1 = order_repo.get_by_id(order1.order_id)
if retrieved_order1:
retrieved_order1.add_line_item(
OrderLineItem(
ProductInfo('P002', 'Monitor', Money(decimal.Decimal('300.00'), 'USD')),
Quantity(1)
)
)
except ValueError as e:
print(f"Successfully prevented modification: {e}")
print("\n--- End of Order Management System Example ---")
if __name__ == "__main__":
run_order_management_examples()
No comments:
Post a Comment