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

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! 🙌