Putting it all together

Up to this point, we have defined messages, their corresponding handlers, and organized them into modules. Now it’s finally time to put everything together.

Creating an application

Below is the factory function for the application. The kwargs in the constructor are the dependencies we used earlier across the modules: TodoRepository, NotificationService, and TodosCounter. Next, modules are linked to the app using app.include_submodule(). Finally, the transaction middleware is configured using on_enter_transaction_context, on_exit_transaction_context, and transaction_middleware decorators.

from collections.abc import Callable
from datetime import datetime
from typing import Any

from analytics import TodosCounter, analytics
from notifications import NotificationService, notifications
from todos import TodoRepository, todos

from lato import Application, TransactionContext


def create_app() -> Application:
    # create an application with dependencies used across the handlers
    app = Application(
        "Tutorial",
        todo_repository=TodoRepository(),  # used by todos module
        notification_service=NotificationService(),  # used by notifications module
        todos_counter=TodosCounter(),  # used ny analytics module
    )
    # add modules to the app
    app.include_submodule(todos)
    app.include_submodule(notifications)
    app.include_submodule(analytics)

    # add transaction context middlewares
    @app.on_enter_transaction_context
    def on_enter_transaction_context(ctx: TransactionContext):
        print("Begin transaction")
        ctx.set_dependencies(
            now=datetime.now(),
        )

    @app.on_exit_transaction_context
    def on_exit_transaction_context(ctx: TransactionContext, exception=None):
        print("End transaction")

    @app.transaction_middleware
    def logging_middleware(ctx: TransactionContext, call_next: Callable) -> Any:
        handler = ctx.current_handler
        message_name = ctx.get_dependency("message").__class__.__name__
        handler_name = f"{handler.source}.{handler.fn.__name__}"
        print(f"Executing {handler_name}({message_name})")
        result = call_next()
        print(f"Result from {handler_name}: {result}")
        return result

    @app.transaction_middleware
    def analytics_middleware(ctx: TransactionContext, call_next: Callable) -> Any:
        result = call_next()
        todos_counter = ctx.get_dependency(TodosCounter)
        print(
            f" todos stats: {todos_counter.completed_todos}/{todos_counter.created_todos}"
        )
        return result

    return app

In on_enter_transaction_context, a transaction-level dependency now is being set. This means that every time a new command is executed using app.execute(), a new value for now will be passed to handlers that require it.

The last piece we are missing to run our test is the conftest.py implementing the app fixture:

import pytest
from application import create_app


@pytest.fixture
def app():
    return create_app()

We could also run the app from the command line using the following script:

from uuid import UUID
from commands import CreateTodo, CompleteTodo
from queries import  GetAllTodos
from application import create_app

app = create_app()

app.execute(CreateTodo(todo_id=UUID(int=1), title="Publish the tutorial"))

all_todos = app.execute(GetAllTodos())
print(all_todos)

app.execute(CompleteTodo(todo_id=UUID(int=1)))

all_todos = app.execute(GetAllTodos())
print(all_todos)

And the output is:

Begin transaction
Executing todos.handle_create_todo(CreateTodo)
 todos stats: 0/0
Result from todos.handle_create_todo: None
Executing analytics.handle_create_todo(CreateTodo)
 todos stats: 0/1
Result from analytics.handle_create_todo: None
End transaction
Begin transaction
Executing todos.get_all_todos(GetAllTodos)
 todos stats: 0/1
Result from todos.get_all_todos: [TodoReadModel(id=UUID('00000000-0000-0000-0000-000000000001'), title='Publish the tutorial', description='', is_due=False, is_completed=False)]
End transaction
[TodoReadModel(id=UUID('00000000-0000-0000-0000-000000000001'), title='Publish the tutorial', description='', is_due=False, is_completed=False)]
Begin transaction
Executing todos.handle_complete_todo(CompleteTodo)
Executing notifications.on_todo_was_completed(TodoWasCompleted)
Executing todos.get_todo_details(GetTodoDetails)
 todos stats: 0/1
Result from todos.get_todo_details: TodoReadModel(id=UUID('00000000-0000-0000-0000-000000000001'), title='Publish the tutorial', description='', is_due=False, is_completed=True)
TodoReadModel(id=UUID('00000000-0000-0000-0000-000000000001'), title='Publish the tutorial', description='', is_due=False, is_completed=True)
A todo Publish the tutorial was completed
 todos stats: 0/1
Result from notifications.on_todo_was_completed: None
Executing analytics.on_todo_was_completed(TodoWasCompleted)
 todos stats: 1/1
Result from analytics.on_todo_was_completed: None
 todos stats: 1/1
Result from todos.handle_complete_todo: None
End transaction
Begin transaction
Executing todos.get_all_todos(GetAllTodos)
 todos stats: 1/1
Result from todos.get_all_todos: [TodoReadModel(id=UUID('00000000-0000-0000-0000-000000000001'), title='Publish the tutorial', description='', is_due=False, is_completed=True)]
End transaction
[TodoReadModel(id=UUID('00000000-0000-0000-0000-000000000001'), title='Publish the tutorial', description='', is_due=False, is_completed=True)]

Understanding the transaction context

When the command is executed via app.execute(), transaction middlewares fire from top to bottom. In our case, logging_middleware is called first. Upon reaching its call_next(), the analytics_middleware is called. As this is the last middleware, it’s call_next passes the execution to the the handler. When the handler execution finishes, middlewares resume execution from bottom to top.

Next, let’s see how to integrate the application with a FastAPI server.