dioxide v1.1.2: Multi-Binding for Plugin Systems

#python #dependencyinjection #architecture #opensource

dioxide v1.1.2 adds multi-binding support. The multi=True parameter on @adapter.for_() registers multiple adapters for the same port, and list[Port] type hints collect them at resolution time.

from dioxide import adapter, service, Container, Profile

@adapter.for_(PluginPort, multi=True, priority=10)
class ValidationPlugin:
    def process(self, data: str) -> str:
        return validate(data)

@adapter.for_(PluginPort, multi=True, priority=20)
class TransformPlugin:
    def process(self, data: str) -> str:
        return data.upper()

@service
class DataProcessor:
    def __init__(self, plugins: list[PluginPort]) -> None:
        self.plugins = plugins  # Both plugins, ordered by priority

The container resolves list[PluginPort] to a list containing all registered multi-binding adapters for the active profile, sorted by priority (lower values first).

Plugin Systems

Mutation testing frameworks need to collect all mutation operators. With multi-binding, each operator is a decorated class that the container discovers during scanning:

class MutationOperator(Protocol):
    def can_mutate(self, node: AST) -> bool: ...
    def mutate(self, node: AST) -> list[AST]: ...

@adapter.for_(MutationOperator, multi=True)
class ArithmeticOperator:
    def can_mutate(self, node: AST) -> bool:
        return isinstance(node, ast.BinOp)

    def mutate(self, node: AST) -> list[AST]:
        # Replace + with -, * with /, etc.
        ...

@adapter.for_(MutationOperator, multi=True)
class ComparisonOperator:
    def can_mutate(self, node: AST) -> bool:
        return isinstance(node, ast.Compare)

    def mutate(self, node: AST) -> list[AST]:
        # Replace == with !=, < with >=, etc.
        ...

@service
class MutationEngine:
    def __init__(self, operators: list[MutationOperator]) -> None:
        self.operators = operators

    def mutate_node(self, node: AST) -> list[AST]:
        for op in self.operators:
            if op.can_mutate(node):
                return op.mutate(node)
        return []

To add an operator, create a class decorated with @adapter.for_(MutationOperator, multi=True) in a module that the container scans.

Ordered Processing Pipelines

The priority parameter controls injection order. Lower values come first, negative values run before the default of 0:

class PipelineStep(Protocol):
    async def execute(self, context: dict) -> dict: ...

@adapter.for_(PipelineStep, multi=True, priority=-10)
class AuthenticationStep:
    async def execute(self, context: dict) -> dict:
        context["user"] = await authenticate(context["token"])
        return context

@adapter.for_(PipelineStep, multi=True, priority=0)
class ValidationStep:
    async def execute(self, context: dict) -> dict:
        validate_request(context)
        return context

@adapter.for_(PipelineStep, multi=True, priority=100)
class AuditStep:
    async def execute(self, context: dict) -> dict:
        await log_request(context)
        return context

@service
class RequestPipeline:
    def __init__(self, steps: list[PipelineStep]) -> None:
        self.steps = steps  # Ordered: Auth -> Validation -> Audit

    async def run(self, context: dict) -> dict:
        for step in self.steps:
            context = await step.execute(context)
        return context

Profile-Filtered Collections

Multi-bindings respect the profile system. Only adapters matching the active profile are included in the resolved list:

@adapter.for_(NotificationChannel, multi=True, profile=Profile.PRODUCTION)
class EmailChannel:
    async def notify(self, message: str) -> None:
        await send_email(message)

@adapter.for_(NotificationChannel, multi=True, profile=Profile.PRODUCTION)
class SlackChannel:
    async def notify(self, message: str) -> None:
        await post_to_slack(message)

@adapter.for_(NotificationChannel, multi=True, profile=Profile.TEST)
class FakeChannel:
    def __init__(self):
        self.messages = []

    async def notify(self, message: str) -> None:
        self.messages.append(message)

@service
class NotificationDispatcher:
    def __init__(self, channels: list[NotificationChannel]) -> None:
        self.channels = channels

    async def broadcast(self, message: str) -> None:
        for channel in self.channels:
            await channel.notify(message)

With Profile.PRODUCTION, channels contains [EmailChannel, SlackChannel]. With Profile.TEST, it contains [FakeChannel].

Constraints and Behavior

A port must be either single-binding or multi-binding across all profiles. Mixing multi=True and multi=False adapters for the same port raises an error at scan() time:

# This fails at scan() time:
@adapter.for_(SomePort, profile=Profile.PRODUCTION)  # single
class SingleAdapter: ...

@adapter.for_(SomePort, multi=True, profile=Profile.TEST)  # multi
class MultiAdapter: ...

If no multi-binding adapters are registered for a port, list[Port] resolves to an empty list rather than raising an error. This supports plugin architectures where zero implementations is valid.

The type hint must be exactly list[Port]. Other generic types like Sequence[Port] or Iterable[Port] are not supported.