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) -> "ContainerBuilder"

Registers synchronous database components: * sqlalchemy.Engine (named "db_sync_engine"): The synchronous SQLAlchemy engine instance, created via get_engine(). * sqlalchemy.orm.sessionmaker[Session] (named "db_sync_session_factory"): The SyncSessionFactory instance. * sqlalchemy.orm.Session: A factory for creating new synchronous sessions (lambda: SyncSessionFactory()). This method ensures that synchronous database access is configured and ready for use.

with_async_database(self) -> "ContainerBuilder"

Registers asynchronous database components: * sqlalchemy.ext.asyncio.AsyncEngine (named "db_async_engine"): The asynchronous SQLAlchemy engine instance, created via get_async_engine(). * sqlalchemy.ext.asyncio.async_sessionmaker[AsyncSession] (named "db_async_session_factory"): The AsyncSessionFactory instance. * sqlalchemy.ext.asyncio.AsyncSession: A factory for creating new asynchronous sessions (lambda: AsyncSessionFactory()). This method sets up asynchronous database access.

with_cache(self, is_async: bool = False) -> "ContainerBuilder"

Registers Redis cache connection components. * is_async (default False): * If False: Registers a synchronous redis.Redis client instance (named "cache_sync") obtained via get_redis_cache_connection(). * If True: Registers a factory for an asynchronous redis.asyncio.Redis client (named "cache_async", singleton scope) obtained via get_redis_cache_async_connection().

with_authentication(self) -> "ContainerBuilder"

Registers the AuthenticationService (castlecraft_engineer.application.auth.AuthenticationService) as a singleton. The factory for AuthenticationService attempts to resolve registered cache clients ("cache_async" first, then "cache_sync") to pass to the AuthenticationService constructor. This allows the authentication service to utilize caching if cache components have been registered.

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) -> "ContainerBuilder"

Sets up and registers authorization components using castlecraft_engineer.authorization.setup.setup_authorization(container). This typically registers an AuthorizationService implementation based on environment variables or pre-existing registrations. Note: with_authentication() should generally be called before with_authorization() if your authorization setup depends on authentication components being available.

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.