Fractal Commands

Fractal Commands is a minimal command bus for building SOLID logic in your Python applications.

Installation

pip install fractal-commands

Background

A command is a plain data object describing an intent: add this road, approve this member. A handler carries it out. The bus is the thin thing in between — it knows which handler answers which command, and nothing else.

Keeping the two apart is what makes the pattern useful. The caller states what it wants and stays ignorant of how it happens; the handler owns the how and never has to know who asked. Commands cross that seam as data, so they are easy to log, queue, replay, or map onto from events.

Usage

from dataclasses import dataclass

from fractal_commands import Command, CommandBus, CommandHandler


@dataclass
class Greet(Command):
    name: str


class GreetHandler(CommandHandler[Greet]):
    command = Greet

    def handle(self, command: Greet):
        return f"hello {command.name}"


bus = CommandBus()
bus.add_handler(GreetHandler())

bus.handle(Greet("world"))
# {'GreetHandler': 'hello world'}

Every handler registered for a command runs. handle collects the return values of the ones that produced something, keyed by handler class name; handlers that return None — most write handlers — simply do not appear.

await bus.handle_async(command) is the same thing for async handlers.

A command with no handler is an error

Dispatching a command nobody handles raises NoCommandHandlerError:

bus = CommandBus()
bus.handle(Greet("world"))
# NoCommandHandlerError: no handler is registered for Greet, so the command was
# dropped without being executed

This is deliberate, and it is worth explaining, because the obvious alternative — shrug and return an empty result — is what this library was extracted to stop doing.

Handlers usually register themselves through a decorator at import time. That makes registration a side effect of the import graph, and import graphs lose edges: a module stops being imported, and every command of that type silently stops being handled. Nothing raises. No event is published, no row is written, and the call returns normally. Tests that assert on the service layer see a perfectly healthy no-op, so the failure surfaces days later somewhere else entirely, as missing data.

Raising turns that into a one-line stack trace at the first dispatch.

When silence is legitimate

A bus fed by fan-out is the honest exception. An event projector that maps events onto commands will produce commands this particular deployment has no handler for, and that is normal rather than broken. Pass strict=False to log the miss at ERROR and carry on:

bus = CommandBus(strict=False)
bus.handle(Greet("world"))
# ERROR:fractal_commands.command_bus:no handler is registered for Greet, so the
# command was dropped without being executed
# {}

It is also the migration path if you are adopting this library in an application that has been relying on the old silence: start with strict=False, fix what the logs show, then turn it on.

Entity commands

The three commands generic CRUD needs are included, built on fractal-specifications and fractal-repositories:

from fractal_commands import AddEntityCommand, UpdateEntityCommand, DeleteEntityCommand

Each carries the specification the handler checks the entity against, so the business rule travels with the intent instead of living in the handler.

Development

make dev-install
make test
make lint
make format