Skip to content

Dependency Injection with Punq

castlecraft-engineer utilizes the punq library for managing Dependency Injection (DI). DI is a design pattern that allows for loose coupling between software components, making applications more modular, testable, and maintainable. Instead of components creating their dependencies directly, these dependencies are "injected" into them, typically by a DI container.

1. The Role of punq

punq is a lightweight and intuitive DI library for Python. It provides a simple way to register services (dependencies) and resolve them when needed. Key features of punq that castlecraft-engineer leverages include:

  • Explicit Registration: Dependencies are explicitly registered with a container.
  • Type Hinting: punq often uses type hints to resolve dependencies.
  • Scopes (Singleton, Transient): While castlecraft-engineer primarily uses punq for transient resolution by default for handlers and services, punq supports different lifestyles for registered components.

2. The ContainerBuilder Class

To simplify the setup of the punq container and the registration of common framework components, castlecraft-engineer provides the ContainerBuilder class, located in castlecraft_engineer.common.di.ContainerBuilder.

The ContainerBuilder offers a fluent API to progressively configure your DI container. It abstracts away some of the direct punq.Container setup for common patterns.

Key ContainerBuilder Methods:

  • __init__(self, container: Optional[punq.Container] = None):

    • Initializes the builder. You can optionally pass an existing punq.Container if you need to pre-configure it.

  • register(self, interface: Type[T], instance: Optional[Type[T]] = None, **kwargs) -> "ContainerBuilder":

    • The primary method for registering custom services. You provide an interface (or abstract type) and an instance (concrete implementation).
    • If instance is not provided, it assumes interface is the concrete class to be instantiated.
    • **kwargs can be used to specify constructor arguments for the instance if they are not resolvable by the container.

  • with_database(self, connection_string: Optional[str] = None, echo: bool = False) -> "ContainerBuilder":

    • Registers a synchronous SQLAlchemy session factory (Session) based on SQL_CONNECTION_STRING environment variable or the provided connection_string.

  • with_async_database(self, connection_string: Optional[str] = None, echo: bool = False) -> "ContainerBuilder":

    • Registers an asynchronous SQLAlchemy session factory (AsyncSession) based on SQL_ASYNC_CONNECTION_STRING environment variable or the provided connection_string.

  • with_cache(self, redis_url: Optional[str] = None, sentinel_config: Optional[Dict[str, Any]] = None) -> "ContainerBuilder":

    • Registers synchronous (redis.Redis) and asynchronous (redis.asyncio.Redis) Redis cache client factories. Configuration is typically read from environment variables (e.g., REDIS_CACHE_URL).

  • with_command_bus(self) -> "ContainerBuilder":

    • Registers the CommandBus and makes it available as self.command_bus on the builder instance for handler registration (e.g., builder.command_bus.register(MyCommand, MyCommandHandler)).

  • with_query_bus(self) -> "ContainerBuilder":

    • Registers the QueryBus and makes it available as self.query_bus on the builder instance for handler registration.

  • with_event_bus(self) -> "ContainerBuilder":

    • Registers the in-process EventBus and makes it available as self.event_bus on the builder instance for handler subscription.

  • with_authentication(self) -> "ContainerBuilder":

    • Registers the AuthenticationService and its dependencies (like cache clients for JWKS).

  • with_authorization(self, service: Optional[Type[AuthorizationService]] = None) -> "ContainerBuilder":

    • Registers an AuthorizationService. If service is not provided, it defaults to DenyAllAuthorizationService. You should register your custom or AllowAllAuthorizationService here.

  • build(self) -> punq.Container:

    • Finalizes the container setup and returns the configured punq.Container instance.

3. Creating and Using the DI Container

The typical workflow involves: 1. Instantiating ContainerBuilder. 2. Using its fluent methods to register framework components and your application-specific services, repositories, and handlers. 3. Calling builder.build() to get the punq.Container. 4. Using container.resolve(TypeToResolve) to get instances of your services, typically at the entry point of a request or operation (e.g., in an API endpoint to resolve a CommandBus or QueryBus).

4. How Components are Resolved

When you request a type from the container (e.g., container.resolve(MyService)), punq inspects the constructor of MyService (or the registered concrete type). It then tries to resolve each constructor parameter by looking up its type in the container. This process happens recursively until all dependencies are resolved.

This is why it's crucial to register all necessary dependencies. For example, if MyCommandHandler depends on MyRepository and AsyncSession: 

class MyCommandHandler(CommandHandler[MyCommand, None]):
    def __init__(self, session: AsyncSession, repo: MyRepository):
        self.session = session
        self.repo = repo
    # ...
You would need to ensure that AsyncSession (via builder.with_async_database()) and MyRepository (via builder.register(MyRepository)) are registered so punq can successfully instantiate MyCommandHandler when it's needed by the CommandBus.

5. Example: Setting Up and Using the Container

Here's a simplified example demonstrating how to set up a container and resolve a command handler.

import asyncio
import punq
from dataclasses import dataclass
from typing import Type
from sqlalchemy.ext.asyncio import AsyncSession

from castlecraft_engineer.common.di import ContainerBuilder
from castlecraft_engineer.abstractions.command import Command
from castlecraft_engineer.abstractions.command_handler import CommandHandler
from castlecraft_engineer.abstractions.command_bus import CommandBus
from castlecraft_engineer.authorization.base_service import AuthorizationService
from castlecraft_engineer.authorization.default_services import AllowAllAuthorizationService

# --- Define a simple Command and Handler ---
@dataclass(frozen=True)
class LogMessageCommand(Command):
    message: str

class LogMessageHandler(CommandHandler[LogMessageCommand, None]):
    def __init__(self, session: AsyncSession, auth_service: AuthorizationService):
        # In a real app, session might be used with a repository
        self.session = session
        self.auth_service = auth_service
        print(f"LogMessageHandler instantiated with session: {type(session)}, auth: {type(auth_service)}")

    async def authorize(self, command: LogMessageCommand) -> None:
        print(f"Authorizing LogMessageCommand for: {command.message}")
        # Example: await self.auth_service.check_permission(...)
        pass # Allow for this example

    async def execute(self, command: LogMessageCommand) -> None:
        print(f"HANDLER EXECUTING: Logging message - '{command.message}'")
        # await self.session.commit() # If any DB operations were done

# --- DI Setup Function ---
def configure_dependencies() -> punq.Container:
    builder = ContainerBuilder()

    # 1. Register database session factory (using a mock for simplicity)
    class MockAsyncSession:
        async def __aenter__(self): return self
        async def __aexit__(self, exc_type, exc, tb): pass
        async def commit(self): print("MockSession: commit called")

    # Register the MockAsyncSession class itself as the factory for AsyncSession
    builder.register(AsyncSession, factory=MockAsyncSession)

    # 2. Register Authorization Service
    builder.with_authorization(service=AllowAllAuthorizationService)

    # 3. Register Command Bus and the handler
    builder.with_command_bus()
    # LogMessageHandler depends on AsyncSession and AuthorizationService
    builder.command_bus.register(LogMessageCommand, LogMessageHandler)

    # 4. Register other services if any
    # builder.register(MyOtherService)

    return builder.build()

# --- Main Execution ---
async def run_example():
    container = configure_dependencies()

    # Resolve the CommandBus
    command_bus = container.resolve(CommandBus)

    # Create and execute a command
    log_cmd = LogMessageCommand(message="Hello from DI example!")
    await command_bus.execute(log_cmd)

if __name__ == "__main__":
    asyncio.run(run_example())

This example illustrates how ContainerBuilder simplifies registering common components like the CommandBus and its handlers. The CommandBus internally uses the container to resolve and instantiate LogMessageHandler along with its dependencies (AsyncSession, AuthorizationService) when the LogMessageCommand is executed.

By leveraging DI, castlecraft-engineer promotes a clean separation of concerns, making your application components easier to manage, test, and evolve.