core.container.handler_factory

src.core.container.handler_factory

Handler Factory Generator - Auto-wire handler dependencies from registry.

This module provides automatic dependency injection for CQRS handlers based on their init type hints. It introspects handler constructors and resolves dependencies from the container.

Architecture: - Uses Python's inspect module to analyze handler signatures - Maps protocol types to container factory functions - Creates request-scoped handler instances with injected dependencies

Usage

from src.core.container.handler_factory import create_handler

In router

async def get_handler(session: AsyncSession = Depends(get_db_session)): return await create_handler(RegisterUserHandler, session)

Reference

• docs/architecture/cqrs-registry.md
• docs/architecture/dependency-injection.md

Functions

get_type_name

get_type_name(annotation: Any) -> str

Extract type name from annotation.

Handles both class types and string forward references.

Parameters:

Name	Type	Description	Default
annotation	Any	Type annotation (class or string).	required

Returns:

Type	Description
str	Type name as string.

Source code in src/core/container/handler_factory.py

def get_type_name(annotation: Any) -> str:
    """Extract type name from annotation.

    Handles both class types and string forward references.

    Args:
        annotation: Type annotation (class or string).

    Returns:
        Type name as string.
    """
    if annotation is None:
        return "None"

    # Handle Optional types (Union with None)
    origin = getattr(annotation, "__origin__", None)
    if origin is not None:
        # For Union types, get the first non-None type
        args = getattr(annotation, "__args__", ())
        for arg in args:
            if arg is not type(None):
                return get_type_name(arg)
        return "None"

    # Handle class types
    if isinstance(annotation, type):
        return annotation.__name__

    # Handle string annotations
    if isinstance(annotation, str):
        return annotation.split(".")[-1]

    # Fallback
    return str(annotation).split(".")[-1].rstrip("'>")

analyze_handler_dependencies

analyze_handler_dependencies(
    handler_class: type,
) -> dict[str, dict[str, Any]]

Analyze handler init to discover dependencies.

Introspects the handler's __init__ method to discover required dependencies and their types.

Parameters:

Name	Type	Description	Default
handler_class	type	Handler class to analyze.	required

Returns:

Type	Description
dict[str, dict[str, Any]]	Dict mapping parameter names to dependency info dicts.
dict[str, dict[str, Any]]	Each info dict contains keys: type_name (str), annotation, is_optional (bool).

Source code in src/core/container/handler_factory.py

def analyze_handler_dependencies(handler_class: type) -> dict[str, dict[str, Any]]:
    """Analyze handler __init__ to discover dependencies.

    Introspects the handler's ``__init__`` method to discover required
    dependencies and their types.

    Args:
        handler_class: Handler class to analyze.

    Returns:
        Dict mapping parameter names to dependency info dicts.
        Each info dict contains keys: type_name (str), annotation, is_optional (bool).
    """
    try:
        # Use getattr to avoid mypy's unsound __init__ access warning
        init_method = getattr(handler_class, "__init__", None)
        if init_method is None:
            return {}
        # Get type hints (resolves forward references)
        hints = get_type_hints(init_method)
    except Exception:
        # Fallback to signature annotations if get_type_hints fails
        init_method = getattr(handler_class, "__init__", None)
        if init_method is None:
            return {}
        sig = inspect.signature(init_method)
        hints = {
            name: param.annotation
            for name, param in sig.parameters.items()
            if name != "self" and param.annotation != inspect.Parameter.empty
        }

    # Remove 'return' from hints
    hints.pop("return", None)

    dependencies: dict[str, dict[str, Any]] = {}

    for param_name, annotation in hints.items():
        if param_name == "self":
            continue

        type_name = get_type_name(annotation)

        # Check if optional (Union with None)
        is_optional = False
        origin = getattr(annotation, "__origin__", None)
        if origin is not None:
            args = getattr(annotation, "__args__", ())
            is_optional = type(None) in args

        dependencies[param_name] = {
            "type_name": type_name,
            "annotation": annotation,
            "is_optional": is_optional,
        }

    return dependencies

create_handler async

create_handler(
    handler_class: type[T],
    session: AsyncSession,
    **overrides: Any
) -> T

Create handler instance with auto-wired dependencies.

Introspects handler init and resolves dependencies: - Repositories: Created with session - Singletons: Retrieved from container - Overrides: Provided explicitly

Parameters:

Name	Type	Description	Default
handler_class	type[T]	Handler class to instantiate.	required
session	AsyncSession	Database session for repositories.	required
**overrides	Any	Explicit dependency overrides.	{}

Returns:

Type	Description
T	Handler instance with injected dependencies.

Raises:

Type	Description
ValueError	If dependency cannot be resolved.

Example

handler = await create_handler(RegisterUserHandler, session) result = await handler.handle(command)

Source code in src/core/container/handler_factory.py

async def create_handler(
    handler_class: type[T],
    session: AsyncSession,
    **overrides: Any,
) -> T:
    """Create handler instance with auto-wired dependencies.

    Introspects handler __init__ and resolves dependencies:
    - Repositories: Created with session
    - Singletons: Retrieved from container
    - Overrides: Provided explicitly

    Args:
        handler_class: Handler class to instantiate.
        session: Database session for repositories.
        **overrides: Explicit dependency overrides.

    Returns:
        Handler instance with injected dependencies.

    Raises:
        ValueError: If dependency cannot be resolved.

    Example:
        >>> handler = await create_handler(RegisterUserHandler, session)
        >>> result = await handler.handle(command)
    """
    dependencies = analyze_handler_dependencies(handler_class)
    resolved: dict[str, Any] = {}

    for param_name, dep_info in dependencies.items():
        type_name = dep_info["type_name"]
        is_optional = dep_info["is_optional"]

        # Check for explicit override
        if param_name in overrides:
            resolved[param_name] = overrides[param_name]
            continue

        # Try to resolve dependency
        try:
            if _is_repository_type(type_name):
                resolved[param_name] = _get_repository_instance(type_name, session)
            elif _is_session_service_type(type_name):
                resolved[param_name] = _get_session_service_instance(type_name, session)
            elif _is_singleton_type(type_name):
                resolved[param_name] = _get_singleton_instance(type_name)
            elif is_optional:
                resolved[param_name] = None
            else:
                raise ValueError(
                    f"Cannot resolve dependency '{param_name}' "
                    f"of type '{type_name}' for {handler_class.__name__}"
                )
        except ValueError:
            if is_optional:
                resolved[param_name] = None
            else:
                raise

    return handler_class(**resolved)

get_supported_dependencies

get_supported_dependencies() -> dict[str, list[str]]

Get list of supported dependency types.

Returns:

Type	Description
dict[str, list[str]]	Dict with 'repositories' and 'singletons' lists.

Source code in src/core/container/handler_factory.py

def get_supported_dependencies() -> dict[str, list[str]]:
    """Get list of supported dependency types.

    Returns:
        Dict with 'repositories' and 'singletons' lists.
    """
    return {
        "repositories": list(REPOSITORY_TYPES.keys()),
        "singletons": list(SINGLETON_TYPES.keys()),
    }

handler_factory

handler_factory(handler_class: type[T]) -> Any

Generate FastAPI dependency for auto-wired handler creation.

Creates a dependency function compatible with FastAPI's Depends() that automatically injects all handler dependencies based on type hints.

The factory function is cached per handler class to enable testing: tests can override the dependency using the same key.

Parameters:

Name	Type	Description	Default
handler_class	type[T]	Handler class to generate dependency for.	required

Returns:

Type	Description
Any	FastAPI-compatible dependency function (cached).

Example

from src.core.container.handler_factory import handler_factory from src.application.commands.handlers.register_user_handler import ( RegisterUserHandler, )

In router:

async def create_user( handler: RegisterUserHandler = Depends(handler_factory(RegisterUserHandler)) ): result = await handler.handle(command)

In tests:

app.dependency_overrides[handler_factory(RegisterUserHandler)] = ( lambda: mock_handler )

Note

This replaces manual factory functions like get_register_user_handler. The returned function uses create_handler internally for dependency resolution.

Source code in src/core/container/handler_factory.py

def handler_factory(handler_class: type[T]) -> Any:
    """Generate FastAPI dependency for auto-wired handler creation.

    Creates a dependency function compatible with FastAPI's Depends() that
    automatically injects all handler dependencies based on type hints.

    The factory function is cached per handler class to enable testing:
    tests can override the dependency using the same key.

    Args:
        handler_class: Handler class to generate dependency for.

    Returns:
        FastAPI-compatible dependency function (cached).

    Example:
        from src.core.container.handler_factory import handler_factory
        from src.application.commands.handlers.register_user_handler import (
            RegisterUserHandler,
        )

        # In router:
        async def create_user(
            handler: RegisterUserHandler = Depends(handler_factory(RegisterUserHandler))
        ):
            result = await handler.handle(command)

        # In tests:
        app.dependency_overrides[handler_factory(RegisterUserHandler)] = (
            lambda: mock_handler
        )

    Note:
        This replaces manual factory functions like `get_register_user_handler`.
        The returned function uses `create_handler` internally for dependency
        resolution.
    """
    # Return cached factory if exists (enables consistent test overrides)
    if handler_class in _handler_factory_cache:
        return _handler_factory_cache[handler_class]

    from fastapi import Depends
    from src.core.container.infrastructure import get_db_session

    async def _factory(
        session: AsyncSession = Depends(get_db_session),
    ) -> T:
        return await create_handler(handler_class, session)

    # Preserve handler class name for debugging/introspection
    _factory.__name__ = f"get_{handler_class.__name__.lower()}"
    _factory.__doc__ = f"Auto-wired factory for {handler_class.__name__}."

    # Cache for reuse
    _handler_factory_cache[handler_class] = _factory

    return _factory

clear_handler_factory_cache

clear_handler_factory_cache() -> None

Clear the handler factory cache.

Useful for test isolation when testing handler_factory itself.

Source code in src/core/container/handler_factory.py

def clear_handler_factory_cache() -> None:
    """Clear the handler factory cache.

    Useful for test isolation when testing handler_factory itself.
    """
    _handler_factory_cache.clear()

get_all_handler_factories

get_all_handler_factories() -> dict[str, Any]

Generate factory functions for all registered handlers.

Uses the CQRS registry to discover all handlers and creates auto-wired factory functions for each.

Returns:

Type	Description
dict[str, Any]	Dict mapping handler name to factory function.

Example

factories = get_all_handler_factories() factories['RegisterUserHandler']

Source code in src/core/container/handler_factory.py

def get_all_handler_factories() -> dict[str, Any]:
    """Generate factory functions for all registered handlers.

    Uses the CQRS registry to discover all handlers and creates
    auto-wired factory functions for each.

    Returns:
        Dict mapping handler name to factory function.

    Example:
        >>> factories = get_all_handler_factories()
        >>> factories['RegisterUserHandler']
        <function get_registeruserhandler at 0x...>
    """
    from src.application.cqrs.computed_views import get_all_handler_classes

    factories: dict[str, Any] = {}
    for handler_class in get_all_handler_classes():
        factories[handler_class.__name__] = handler_factory(handler_class)

    return factories