Clean Architecture: How to decouple the Repository from a CommandHandler?

Question

We are just starting with event driven (and clean) architecture. So far we have two main entrypoints, a consumer (reader from a Redis Stream / Kafka Topic), and an API. As this is almost a modular monolith and not a proper isolated microservice, we added an extra layer that we called the Processor, it basically redirects the message from a specific topic to a specific Module. The basic architecture we came up so far was:

API / Consumer => MessageBus => ModuleProcessor => CommandHandler => ModuleService

Our very simple message bus is something like:

class MessageBus:

    strategies = {
        "mytopic1": ModuleABCProcessor,
        "mytopic2": ModuleBCDProcessor,
    }  # type: Dict[str, Type[interfaces.ProcessorInterface]]

    def handle(self, msg: commons.Message):
        logger.info("Message Bus | Sending command to the appropriate processor...")
        processor: Type[interfaces.ProcessorInterface] = self.strategies[msg.topic]
        processor.process_msg(msg=msg)

The processor is more specific then the Bus, and can now parse the message (we are using the envelope pattern) into command objects (we are using Pydantic), and pass it to the proper Handler.

class ModuleAProcessor(ProcessorInterface):
    command_strategies = {
        "mytopic1.actionA": (cmd.CommandObjectA, handlers.ObjectACommandHandler),
        "mytopic1.actionB": (cmd.CommandObjectB, handlers.ObjectBCommandHandler),
    }  # type: Dict[str, Tuple[Type[Command], Type[HandlerInterface]]]

    @classmethod
    def process(cls, msg: Message) -> None:
        logger.info(f"ModuleA Processor | Handling command {msg.headers.subtopic}...")
        strategy = cls.command_strategies[msg.headers.subtopic]
        parsed_command = strategy[0](**msg.body.dict())
        handler = strategy[1](cmd=parsed_command)
        handler.handle()

Now the process method sends the Command to the appropriate Handler. So far so good, the logic works for multiple topics, multiple subtopics/actions and the organization suits our project well.

class ObjectACommandHandler(interfaces.HandlerInterface):
    def __init__(self, cmd: CommandA) -> None:
        self.cmd = cmd
        logger.info(f"I'm handling command id: {self.cmd.uuid}")

    def handle(self):
        try:
            # I'm running some business logic
            # Depend on Module's A repository
            # and a couple of Module's B model
            self._successful_event(self)
        except:
            self.failed_event(self)

    def _successful_event(self):
        # will dispatch a successful event
        pass

    def _failed_event(self):
        # will dispatch a failed event
        pass

The problem starts at the Handler. Usually to perform a use case, we will need a repository and maybe one or two dependencies (a part from the Command object itself). But we are unsure on how to do this:

1. Should the business logic stay on the CommandHandler? Or should we make the CommandHandler a "dummy" handler, and make it only pass parsed information to a Service?
2. How to inject the repository? Should the repository be injected on the Processor layer, so that the CommandHandler already has all the dependencies, or should the CommandHandler gather all dependencies and inject the dependencies to the Service?
3. we also need to instantiate a couple of models for the CommandHandler (or Service) to use. Where should we build these objects? We thought of having a factory, but again, where would we call this factory? The CommandHandler or maybe on an even higher layer (the Processor perhaps?)?

Our current folder structure is something like this:

.
├── app
│   ├── consumer.py
│   ├── dispatcher.py
│   ├── bus.py
│   ├── infra
│   │   ├── cache/
│   │   ├── db/
│   │   └── stream/
│   ├── modules
│   │   ├── module_a
│   │   │   ├── __init__.py
│   │   │   ├── builders.py
│   │   │   ├── commands.py
│   │   │   ├── events.py
│   │   │   ├── handlers.py
│   │   │   ├── models.py
│   │   │   ├── orm.py
│   │   │   ├── repositories.py
│   │   │   ├── routes.py
│   │   │   ├── schemas.py
│   │   │   └── services.py
│   │   ├── module_b
│   │   │   ├── __init__.py
│   │   │   ├── commands.py
│   │   │   ├── events.py
│   │   │   ├── handlers.py
│   │   │   ├── models.py
│   │   │   ├── orm.py
│   │   │   ├── repositories.py
│   │   │   ├── routes.py
│   │   │   ├── schemas.py
│   │   │   └── services.py
│   ├── processor
│   │   ├── __init__.py
│   │   ├── module_a.py
│   │   ├── module_b.py
│   │   ├── ...
│   ├── core
│   │   ├── __init__.py
│   │   ├── envelopes.py
│   │   ├── events.py
│   │   ├── interfaces.py
│   │   ├── routes.py
│   │   ├── ...
├── poetry.lock
├── pyproject.toml
└── settings.toml

We have been considering making the handlers a first class citzen at the same level as the processor, and make them able to import stuff from several modules. But we are unsure.

Any ideas would be great, we have been a bit stuck on this implementation.

Thanks!

Answer 1

If anyone is interested in the future, I came to the following design. I added dependencies.py module to app > core, which is basically an object that maps the Handlers to their dependencies, something like:

handler_dependencies = {
    "moduleA": {
        MyCommandHandler: {
            "repo": ModuleRepository(),
            "dep1": AnotherDependency()
        }
    }
}

So MyCommandHandler will be something like:

class MyCommandHandler:
    def __init__(self, repo: ARepositoryInterface, dep1: AnotherDependencyInterface):
...

And then, at the processor level, I inject the dependencies like:

@classmethod
def process(cls, msg: Message) -> None:
    logger.info(f"ModuleA Processor | Handling command {msg.headers.subtopic}...")
    strategy = cls.command_strategies[msg.headers.subtopic]
    parsed_command = strategy[0](**msg.body.dict())
    handler = strategy[1]
    handler_dependencies = dependencies['moduleA'][handler]
    command_handler = handler(cmd=parsed_command, **handler_dependencies)
    handler.handle()

I could also inspect the handler's signature and the dependencies object at startup to ensure that the dependencies match (the dict key and the param name), so the str is not much of an issue (misspelling and etc).

The solution seems to work, bring a fair good amount of decoupling between the business and the dependencies, and it's flexible enough so you can change dependencies by configuration file later on (you would just need another dict mapping a config_key (like in_memory or postgres for the repository layer for example) to the correct dependency object. This also avoids external dependency injection packages.