Cosmic Python Book

Part II of the book focuses on extending domain modeling techniques to distributed systems using asynchronous message passing. The goal is to compose systems from small components that interact via messages rather than relying solely on synchronous HTTP APIs, which can lead to a 'distributed big ball of mud'.

Key patterns covered include:

• Domain Events: Used to trigger workflows that cross consistency boundaries.
• Message Bus: Provides a unified way of invoking use cases from any endpoint.
• CQRS (Command Query Responsibility Segregation): Separates read and write operations to avoid compromises in event-driven architectures and to enable performance and scalability improvements.
• Dependency Injection: Used to manage component wiring and tidy up loose ends in the architecture.

The project utilizes four key design patterns to build a rich object model that is decoupled from technical concerns (like databases) and supports aggressive refactoring:

• Repository pattern: Provides an abstraction over persistent storage.
• Service Layer pattern: Defines the boundaries and entry points for use cases.
• Unit of Work pattern: Manages atomic operations to ensure data consistency.
• Aggregate pattern: Enforces data integrity within domain boundaries.

When transforming an application into a message processor where the entire app operates via a message bus, consider the following architectural trade-offs:

Pros:

• Simplified Service Layer: Handlers and services are treated as the same thing, reducing architectural complexity.
• Standardized Inputs: Provides a consistent data structure for all inputs entering the system.

Cons:

• Unpredictable Execution Flow: From a web/request-response perspective, a message bus can be unpredictable because the completion of a process is not immediately obvious.
• Data Duplication: There is a maintenance cost due to the duplication of fields and structures between model objects and events. Adding a field to a model typically requires adding it to the corresponding event(s).

To follow along with the architectural changes in this chapter, clone the repository and checkout the specific branch for this chapter:

git clone https://github.com/cosmicpython/code.git
cd code
git checkout chapter_09_all_messagebus

Alternatively, you can checkout the previous chapter's state using:

git checkout chapter_08_events_and_message_bus

To explore the specific implementation of the Repository and Unit of Work patterns using Django, you can check out the appendix_django branch of the code repository.

git clone https://github.com/cosmicpython/code.git
cd code
git checkout appendix_django

To follow along with the Unit of Work pattern implementation, clone the repository and checkout the specific branch for this chapter:

git clone https://github.com/cosmicpython/code.git
cd code
git checkout chapter_06_uow

Alternatively, you can checkout the Chapter 4 branch to see the service layer without the Unit of Work pattern:

git checkout chapter_04_service_layer

git clone https://github.com/cosmicpython/code.git
cd code
git checkout chapter_06_uow
# or to code along, checkout Chapter 4:
git checkout chapter_04_service_layer

In entrypoints like Flask, instead of manually configuring the UnitOfWork and starting the ORM, call the bootstrap() function to obtain a ready-to-use MessageBus instance.

from allocation import bootstrap, views

app = Flask(__name__)
bus = bootstrap.bootstrap()

@app.route("/add_batch", methods=["POST"])
def add_batch():
    cmd = commands.CreateBatch(request.json["ref"], request.json["sku"], request.json["qty"], eta)
    bus.handle(cmd)
    return "OK", 201

Semantic validation checks if a message is meaningful within the current state of the system (e.g., 'Does this product exist?'). This is best handled in the service layer using preconditions.

Use a common base exception class for invalid messages to make error reporting easier. For example, mapping a ProductNotFound exception to a 404 HTTP status code in a web API.

class MessageUnprocessable(Exception):
    def __init__(self, message):
        self.message = message

class ProductNotFound(MessageUnprocessable):
    def __init__(self, message):
        super().__init__(message)
        self.sku = message.sku

def product_exists(event, uow):
    product = uow.products.get(event.sku)
    if product is None:
        raise ProductNotFound(event)

Implementing the Aggregate pattern involves several trade-offs:

Pros:

• Encapsulation: Helps decide which domain model classes are public and which are internal.
• Performance: Modeling operations around explicit consistency boundaries helps avoid ORM performance problems.
• Reasoning: Putting the aggregate in sole charge of state changes to subsidiary models makes the system easier to reason about and helps control invariants.

Cons:

• Complexity: Adds another concept (beyond Entities and Value Objects) for developers to learn.
• Strictness: Requires a mental shift to follow the rule of modifying only one aggregate at a time.
• Consistency: Dealing with eventual consistency between different aggregates can be complex.