Dependency Injection in Castlecraft Engineer

castlecraft-engineer utilizes the punq library for dependency injection (DI). DI is a design pattern that allows for loose coupling between components by providing dependencies to objects rather than having them create dependencies themselves. This promotes modularity, testability, and maintainability.

The primary tool for configuring the DI container is the ContainerBuilder class found in castlecraft_engineer.common.di.

ContainerBuilder

The ContainerBuilder provides a fluent interface to progressively register components and services with the punq.Container.

Initialization

from castlecraft_engineer.common.di import ContainerBuilder

builder = ContainerBuilder()

When a ContainerBuilder is initialized, it creates an internal punq.Container and registers the container itself, making it available for injection if needed.

Core Registration Methods

register(self, type_or_name: Any, **kwargs) -> "ContainerBuilder"

This is the fundamental method for registering components directly with the underlying punq.Container. * type_or_name: The type (e.g., a class or an abstract base class) or a string name that will be used to request the dependency. * **kwargs: Keyword arguments passed directly to punq.Container.register(). Common punq arguments include: * instance: A pre-existing object to be returned when type_or_name is resolved. * factory: A callable (function or class) that creates an instance of type_or_name. If factory is a class, punq will instantiate it. If it's a function, punq will call it. * scope: Defines the lifecycle of the registered component (e.g., punq.Scope.singleton, punq.Scope.transient). * name: A string name to register the component under, allowing multiple registrations for the same type. If only type_or_name is provided and it's a concrete class, punq will attempt to instantiate it, resolving its constructor dependencies automatically.

Example:

class MyService:
    pass

class MyServiceImpl(MyService):
    pass

builder.register(MyService, instance=MyServiceImpl()) # Register a specific instance
builder.register(MyServiceImpl) # Register MyServiceImpl as itself (transient by default)

Helper with_* Methods

The ContainerBuilder provides several with_* methods to simplify the registration of common infrastructure components.

with_database(self, name: str = "default", db_config: Optional[Dict]) -> "ContainerBuilder"

Registers synchronous database components for a named connection. This allows you to register multiple databases, such as a default write database and a read-replica.

• name: A unique name for the connection (e.g., "default", "read_replica"). Defaults to "default".
• db_config: An optional dictionary to programmatically configure the connection, overriding environment variables. Keys can include connection_string and enable_sql_log.

When using the default name, you can resolve the session by its type (Session or AsyncSession). For named connections, you must resolve the session using its specific registration name (e.g., "db_sync_session_read_replica").

Example: Registering a default database and a read-replica

builder = ContainerBuilder()

# Register the default database using environment variables
builder.with_async_database(name="default")

# Register a read-replica with a specific connection string
read_replica_config = {
    "async_connection_string": "postgresql+asyncpg://user:pass@read-replica-host/db"
}
builder.with_async_database(name="read_replica", db_config=read_replica_config)

container = builder.build()

# Resolve the default session by type
default_session = container.resolve(AsyncSession)

# Resolve the read-replica session by its generated name
read_session = container.resolve(AsyncSession, name="db_async_session_read_replica")

with_async_database(self, name: str = "default", db_config: Optional[Dict]) -> "ContainerBuilder"

This is the asynchronous equivalent of with_database. It registers AsyncEngine and AsyncSession components. The db_config can include keys like async_connection_string.

with_cache(self, is_async: bool = False, cache_config: Optional[Dict]) -> "ContainerBuilder"

Registers Redis cache components.

• is_async: If True, registers an async client; otherwise, a sync client.
• cache_config: An optional dictionary to programmatically configure the Redis connection, overriding environment variables.

with_authentication(self, auth_config: Optional[Dict]) -> "ContainerBuilder"

Registers the AuthenticationService as a singleton.

• auth_config: An optional dictionary to programmatically configure the AuthenticationService, overriding environment variables.

with_command_bus(self) -> "ContainerBuilder"

Registers the CommandBus (castlecraft_engineer.abstractions.command_bus.CommandBus) as a singleton. The CommandBus itself is initialized with the DI container to resolve command handlers. The resolved CommandBus instance is also made available as builder.command_bus.

with_query_bus(self) -> "ContainerBuilder"

Registers the QueryBus (castlecraft_engineer.abstractions.query_bus.QueryBus) as a singleton. Similar to the CommandBus, it's initialized with the DI container to resolve query handlers. The resolved QueryBus instance is also made available as builder.query_bus.

with_event_bus(self) -> "ContainerBuilder"

Registers the EventBus (castlecraft_engineer.abstractions.event_bus.EventBus) as a singleton. It's initialized with the DI container, which can be used if the EventBus implementation needs to resolve event handlers or other dependencies. The resolved EventBus instance is also made available as builder.event_bus.

with_authorization(self, engine_name: Optional[str] = None) -> "ContainerBuilder"

Sets up and registers authorization components.

• engine_name: Optionally specify the authorization engine name (e.g., "allow_all", "deny_all"), overriding the ENV_AUTHORIZATION_ENGINE environment variable.

Note: with_authentication() should generally be called before this method.

Building the Container

build(self) -> punq.Container

After all desired components have been registered, this method returns the fully configured punq.Container instance.

container = builder.with_async_database().with_command_bus().build()

# Now you can resolve dependencies
from sqlalchemy.ext.asyncio import AsyncSession
from castlecraft_engineer.abstractions.command_bus import CommandBus

async_session = container.resolve(AsyncSession) # Gets a new session from the factory
command_bus_instance = container.resolve(CommandBus)

create_injector(container: punq.Container)

The di.py module also provides a create_injector utility function. This function takes a punq.Container and returns a decorator factory (inject). The inject decorator can be used to automatically inject dependencies into the keyword arguments of a function or method based on their type annotations.

How it Works

The inject decorator: 1. Inspects the signature of the decorated function/method. 2. For each parameter that has a type annotation and was not explicitly passed by the caller: a. It attempts to resolve an instance of that type from the provided container. b. If successful, the resolved dependency is added to the keyword arguments passed to the original function. 3. If a dependency cannot be resolved, an error is logged, but the original function is still called (it might handle the missing dependency or raise an error itself).

Example Usage

from castlecraft_engineer.common.di import ContainerBuilder, create_injector
from castlecraft_engineer.abstractions.command_bus import CommandBus

# 1. Configure and build the container
builder = ContainerBuilder()
builder.with_command_bus()
container = builder.build()

# 2. Create the injector decorator
inject = create_injector(container)

# 3. Define a function that needs dependencies
class MyCommand:
    pass

@inject
async def handle_some_action(command: MyCommand, bus: CommandBus, explicit_arg: str = "default"):
    print(f"Explicit arg: {explicit_arg}")
    print(f"Injected CommandBus: {bus}")
    # await bus.execute(command) # bus is now available
    return f"Handled {command} with {bus}"

async def main():
    # CommandBus will be injected automatically
    await handle_some_action(MyCommand(), explicit_arg="passed_value")

    # If bus was not registered, it would log an error and bus would be None (or raise AttributeError if accessed)
    # For example, if CommandBus was not registered:
    # await handle_some_action(MyCommand()) # Would log error for missing CommandBus

# import asyncio
# asyncio.run(main())

The create_injector provides a convenient way to perform DI for functions or methods without needing to manually resolve each dependency from the container within the function body. This is particularly useful for application entry points, route handlers in web frameworks, or any function that orchestrates services.

Best Practices

• Centralized Configuration: Configure your DI container at the application's entry point (e.g., in main.py or a dedicated startup module).
• Register Interfaces: Prefer registering interfaces (abstract base classes) and their concrete implementations. This allows for easier swapping of implementations.
• Scope Management: Understand punq's scopes (transient, singleton) to manage the lifecycle of your services appropriately.
• Avoid Service Locator Pattern: Inject dependencies directly into constructors or methods rather than resolving them from a globally accessible container within your components. The ContainerBuilder and create_injector aim to simplify these practices within the castlecraft-engineer ecosystem.