Creating Providers and Modules

Overview

Providers wire services into the DI container. Modules group providers and enforce visibility.

Core Pattern

Provider = registers/boots/shuts down one bounded concern. Module = groups providers, defines imports/exports.

Provider Lifecycle

from lexigram.di import Provider
from lexigram.contracts.core.di import ContainerRegistrarProtocol, BootContainerProtocol
from lexigram.contracts.core.health import HealthCheckResult, HealthStatus

class MyProvider(Provider):
    name = "my_provider"
    priority = ProviderPriority.NORMAL

    async def register(self, container: ContainerRegistrarProtocol) -> None:
        container.singleton(MyProtocol, MyImpl)
        container.transient(OtherProtocol, lambda: OtherImpl(...))

    async def boot(self, container: BootContainerProtocol) -> None:
        svc = await container.resolve(MyProtocol)
        await svc.connect()

    async def shutdown(self) -> None:
        await self._cleanup()

    async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
        return HealthCheckResult(component=self.name, status=HealthStatus.HEALTHY)

Rules

• register() gets ContainerRegistrarProtocol — binds only, no resolution
• boot() gets BootContainerProtocol — may resolve AND register
• Optional hooks: on_error(error, phase), shutdown(), health_check(timeout)
• No business logic on Provider classes
• All I/O in boot/shutdown is async

Provider Priorities

Shutdown runs in reverse order.

Module Patterns

Static Module (no config)

from lexigram.di import module, Module

@module(providers=[MyProvider], imports=[ConfigModule], exports=[MyProtocol])
class MyModule(Module):
    pass

Dynamic Module (with config)

from lexigram.di import module, Module, DynamicModule

@module()
class MyModule(Module):
    @classmethod
    def configure(cls, config: MyConfig | None = None) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[MyProvider(config=config)],
            exports=[MyProtocol],
        )

    @classmethod
    def stub(cls, config: MyConfig | None = None) -> DynamicModule:
        return DynamicModule(
            module=cls,
            providers=[MyStubProvider(config=config)],
            exports=[MyProtocol],
        )

Three factory conventions: configure(), scope(), stub().

Package File Layout

src/lexigram/mypackage/    __init__.py (lazy exports), config.py,
                           module.py, exceptions.py, di/provider.py
tests/                     pyproject.toml

Shared Queue Lifecycle

Coordinate subscribe/unsubscribe when multiple providers share a queue:

class WorkerProvider(Provider):
    async def boot(self, container):
        queue = await container.resolve(QueueProtocol)
        self.sub = await queue.subscribe("orders.new", self.handle)

    async def shutdown(self):
        await self.sub.unsubscribe()   # each provider owns its handles

Never share a Subscription across providers — always unsubscribe in shutdown().

Common Mistakes

• Registering after container freeze — raises ContainerError
• Resolving in register() — use boot() instead
• Business logic on Provider — belongs in services, not providers
• Forgetting exports in DynamicModule — consumers can't resolve the type
• Not defining stub() — forces test modules to use real implementations
• Declaring config_key without config_model — raises ConfigurationError at boot unless a LexigramConfig is registered or provider.config is set directly
• Not unsubscribing in shutdown() — subscription handles leak, consumers pile up
• Sharing a Subscription across providers — race on unsubscribe tears down other consumers