injector_command_bus.py

# tested on Python3.9 with just injector installed (pip install injector==0.18.4)
from dataclasses import dataclass
from typing import TypeVar, Generic

from injector import Injector, Module, provider


TCommand = TypeVar("TCommand")


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



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



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()


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)


container = Injector([Enrolment()], auto_bind=False)
command_bus = CommandBus(container)
command_bus.handle(Enrol(student_id=123000, course_id=666))

I really doubt this is a good idea to pass container to CommandBus constructor. A way better to provide all required handlers to constructor directly.

I really doubt this is a good idea to pass container to CommandBus constructor. A way better to provide all required handlers to constructor directly.

Care to elaborate on why do you think so?

The container is a private detail of CommandBus anyway. Using Handler generics it provides a nice and easy way to extend a list of handled commands and their handlers without modifying any code. This is an example of the Open-Closed Principle.

What you propose would force developers to modify the creation logic of CommandBus each time a new handler is added.

Besides, what about constructing a handler? It's not a function. It's an instance of a class that has its own dependencies etc. Container deals with building it. With it being passed, we either have to have it build in advance (not practical) or use plain functions (limiting).

What are in your opinion the benefits of other implementation? What's wrong with this one?

You said this is a nice and easy way to extend a list of commands, but you use DI container like a service allocator and pass it to constructor directly. This is basically the main thing I dont like.

That's why I prefer to pass handlers like this:

container = Injector(...)
....
command_bus = CommandBus(handlers=[
   container.get(EnrolHandler),
])
...
command_bus.handle(Enrol(student_id=123000, course_id=666))

or even simpler:

command_bus = CommandBus(handlers=container.get(typing.List[BaseHandler]))

But I'm not sure this specific implemenation of DI supports it (I use inject library).

You said this is a nice and easy way to extend a list of commands, but you use DI container like a service allocator and pass it to constructor directly. This is basically the main thing I dont like.

I think you mean Service Locator but I don't agree it is an instance of it. I think it might look like it because this gist is taken out of the context. Basically, such a CommandBus is more like "framework" code to me and glue between the container and actual, production code.

A complete example could look like this (pseudocode, let's say it's Flask):

@app.route('/api/courses/<course_id>', methods=['POST'])
def enrol_view(bus: CommandBus):  # bus is injected using framework integration e.g. flask-injector lib
    command = ... # build command out of request.json, irrelevant for the example
    try:
        bus.handle(command)
    except AlreadyEnrolled:
        return '', 409
    else:
        return '', 204

CommandBus is used at the boundaries of the project - e.g. API views or background (Celery/Rq/whatever) tasks. It's never directly created.

The second thing that may not be obvious is that I use scopes and creating objects on-demand instead of singletons. So let's say EnrolHandler looks like:

class EnrolHandler:
    _session: Session  # sqlalchemy.orm.Session, a stateful object (!)

   def __call__(self, command: Enrol) -> None:
       self._session.add(...)

Since _session is a stateful object that tracks models, I cannot create EnrolHandler in advance. Each request will have its own _session that will be the same for all objects created during the same request/background task etc. I cannot have it if EnrolHandler is created in advance with all dependencies

My bad, I meant Service Locator of course. Which container do you pass to constructor of CommandBus in this case? Do you have a separate container having only BaseHandler classes?

Besides that your examples are quite close / or almost the same what I'm doing in my code.

and if CommandBus uses global container inside it becomes a super object which in runtime can get any other object it wants, I guess this is what Service Locator pattern is about.

No, Service Locator is about passing and using global container all around in various places and layers in code. That's actually the antipattern that is very easy to get with inject as it relies on a global state.

I disagree CommandBus becomes super object - container is kept in a private field (_container) which should discourage people from using it from the outside. And CommandBus itself won't be using the container to get any object - that's why these Handler generics are for. You can't just pass anything to CommandBus.handle method and expect it to get from container something it shouldn't

What would be the scenario when CommandBus gets something undesirable from the container?

Hi @Enforcer

Very nice and clean implementation. I have one question:

How do you deal with dependencies in handlers? For example:

class EnrolHandler(Handler[Enrol]):
    def __init__(self, someService: SomeService) -> None:
        self._someService = someService

    def __call__(self, command: Enrol) -> None:
        self._someService.do(command.student_id)

Cheers!

Hi @luru-eb
one needs to tell container how to build EnrolHandler; it's still injected as you can see here https://gist.github.com/Enforcer/920ecebdabb5c4b75c67bc43aac5b325#file-injector_command_bus-py-L42

You just need to extend Injector's module:

class Enrolment(Module):
    @provider
    def some_service(self) -> SomeService:
        return SomeService()

    @provider
    def enrol_handler(self, some_service: SomeService) -> Handler[Enrol]:
        return EnrolHandler(some_service=some_service)

Thank you so much! 🙌