How to implement and use Command Bus in Python with Injector?

What's a command bus?

Command Bus is an incarnation of Mediator design pattern. It provides a way to decouple the code structure that sends a command to its receiver. It becomes handy with CQRS implementation with regard to the write stack. Commands are implemented as immutable data structures. It can be done with e.g. dataclasses or attrs. Command handlers are places where the logic is executed.

from dataclasses import dataclass

@dataclass(frozen=True)
class Enroll:
    student_id: int
    course_id: int


class EnrolHandler:
    def __call__(self, command: Enroll) -> None:
        ...

The concept is straightforward and it makes it pretty simple to use the functionalities. A caller just need a command instance and a command bus. The latter calls the appropriate handler. But how does a command bus look like? How to get one?

Which command bus to use?

Some time ago, I've wrote a little pure Python library implementing Command- and Event Buses - pybuses. And frankly - I haven't been using it for a very long time. It is mostly because I fell in love with injector - a mature dependency injection container, modelled after Java's Guice. In the meantime, I come across a C# project created to demonstrate the basics of modular monoliths. I found out it uses a custom Command/Event Bus implementation that leverages dependency injection container - Autofac. The code is so elegant and simple that I decided to just rewrite it using Injector. That became my personal standard.

To start with, we need a base generic class that will be used for injector bindings configuration. Concrete handlers can inherit from it to make mypy happy.

from typing import Generic, TypeVar

TCommand = TypeVar("TCommand")


class Handler(Generic[TCommand]):
    def __call__(self, command: TCommand) -> None:
        raise NotImplementedError

Now, using the previous example with Enrol command, its handler configuration would look like this:

from injector import Injector, Module, provider

class EnrolHandler(Handler[Enrol]):
    def __call__(self, command: Enrol) -> None:
        print(f"command: {command}")


class Enrolment(Module):
    @provider
    def enrol_handler(self) -> Handler[Enrol]:
        return EnrolHandler()

container = Injector([Enrolment()], auto_bind=False)

Why am I using a class for EnrolHandler even though it has a single method __call__? Because in a real world-code I'd be injecting some collaborating classes, e.g repository or some port. Command handlers have the same responsibilities as Use Cases / Interactors. If you're not sure what I am writing about, see my previous posts about the Clean Architecture.

Finally, a command bus itself gets an instance of Injector and will use it to look for Handler of a given command.

class CommandBus:
    def __init__(self, container: Injector) -> None:
        self._container = container

    def handle(self, command: TCommand) -> None:
        command_cls: Type[TCommand] = type(command)
        handler = self._container.get(Handler[command_cls])
        handler(command)

Summary

It's that simple. Aw working example can be found here. In the project, CommandBus is also configured using a provider (just like Handler[Enrol]), so it can later be injected into views or background tasks.

Event Bus implementation is trickier because one needs to use injector's multibind. You can find a full example in my repository that contains example Python project written using the Clean Architecture.