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.

PlantUML diagram

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.

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

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.

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.

0

One thought to “How to implement and use Command Bus in Python with Injector?”

Leave a Reply

Your email address will not be published. Required fields are marked *

This site uses Akismet to reduce spam. Learn how your comment data is processed.