orchestrata_ultimate.di.container

Reference › orchestrata_ultimate › di › container

The whole dependency-injection system of orchestrata, described in the di narrative. Its only project imports are the IComponent interface and ComponentScope enum from core.protocols, and the @traces decorator; the in-tree consumer is the gateway server, which uses DIContainer in its composition root.

• ComponentRegistration is the record stored per interface: the implementation class, the scope (default singleton), a config dictionary, an explicit dependency list and an optional factory callable.
• ResolutionContext is created for each resolution pass. Its resolving stack detects circular dependencies, its cache holds singleton instances, and context_id isolates request-scoped components.
• DIContainer registers with register (a class or factory) and register_instance, both guarded by allow_override; _validate_registration checks each registration. resolve(interface, context_id) is a coroutine that descends through _resolve_with_context, _create_instance, _resolve_dependencies and _resolve_param, using inspect on the constructor (_extract_dependencies) to inject dependencies recursively. Scopes are singleton (cached), transient (rebuilt on every resolve) and request (cached per context id). set_context and get_context carry keyed values, has, get_all_registrations and get_registry_info introspect, and shutdown stops components in reverse dependency order.
• ContainerBuilder is the fluent, configuration-driven path: with_config, with_component, with_instance, with_context, then build.
• ContainerRegistry publishes containers by name in a class-level dictionary (register, get, has) and shutdown_all tears them all down.

At a glance

Class diagram

Dependency diagram

API

Classes

ComponentRegistration (dataclass)

Decorators: dataclass · Line 31

Registration information for a component

ResolutionContext (dataclass)

Decorators: traces(spec='024', story='Resolution', reqs=['REQ-024-025', 'REQ-024-030']), dataclass · Line 48

Context for dependency resolution

DIContainer

Line 65

Dependency injection container for Orchestrata components.

ContainerBuilder

Line 685

Fluent builder for configuring DI containers.

ContainerRegistry

Decorators: traces(spec='024', story='Container Registry', reqs=['REQ-024-045']) · Line 838

Global registry for DI containers.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/di/container.py

# mypy: ignore-errors
"""
Dependency Injection Container for Orchestrata.

Provides a flexible, type-safe dependency injection system
that supports multiple scopes and configuration-driven assembly.
"""

from __future__ import annotations

import inspect
import logging
import threading
from collections.abc import Callable
from copy import copy
from dataclasses import dataclass, field
from typing import Any, TypeVar, cast

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import ComponentScope, IComponent

# Type variables
T = TypeVar("T", bound=IComponent)

# ============================================================================
# Data Classes
# ============================================================================


@dataclass
class ComponentRegistration:
    """Registration information for a component"""

    interface: type[IComponent]
    implementation: type[IComponent] | None = None
    scope: ComponentScope = ComponentScope.SINGLETON
    config: dict[str, Any] = field(default_factory=dict)
    dependencies: list[str] = field(default_factory=list)
    factory: Callable | None = None


@traces(
    spec="024",
    story="Resolution",
    reqs=["REQ-024-025", "REQ-024-030"],
)
@dataclass
class ResolutionContext:
    """Context for dependency resolution"""

    resolving: list[str] = field(
        default_factory=list
    )  # Stack of interfaces being resolved
    cache: dict[str, IComponent] = field(
        default_factory=dict
    )  # Cache for singleton instances
    context_id: str = ""  # For REQUEST scope isolation


# ============================================================================
# Dependency Injection Container
# ============================================================================


class DIContainer:
    """
    Dependency injection container for Orchestrata components.

    Supports:
    - Interface-based registration
    - Multiple scopes (singleton, transient, request)
    - Constructor injection
    - Configuration-driven assembly
    - Circular dependency detection
    """

    def __init__(self, name: str = "default", allow_override: bool = False) -> None:
        """
        Initialize DI container.

        Args:
            name: Container name for logging
            allow_override: If True, allows re-registering same interface
        """
        self.name = name
        self._registry: dict[str, ComponentRegistration] = {}
        self._context: dict[str, Any] = {}
        self._singletons: dict[str, IComponent] = {}  # Separate cache for singletons
        self._lock = threading.Lock()
        self._logger = logging.getLogger(f"di.container.{name}")
        self._allow_override = allow_override

        # Register self as a service
        self._context["di_container"] = self

    @traces(
        spec="024",
        story="Core Registration",
        reqs=[
            "REQ-024-001",
            "REQ-024-002",
            "REQ-024-003",
            "REQ-024-004",
            "REQ-024-007",
            "REQ-024-008",
            "REQ-024-009",
            "REQ-024-010",
            "REQ-024-051",
        ],
    )
    def register(
        self,
        interface: type[T],
        implementation: type[T] | None = None,
        *,
        scope: ComponentScope = ComponentScope.SINGLETON,
        config: dict[str, Any] | None = None,
        factory: Callable[..., T] | None = None,
        dependencies: list[str] | None = None,
    ) -> DIContainer:
        """
        Register a component implementation.

        Args:
            interface: Interface class
            implementation: Implementation class (optional if using factory)
            scope: Component scope (SINGLETON, TRANSIENT, REQUEST)
            config: Component configuration dict
            factory: Factory function for creating instances
            dependencies: Explicit dependency override list

        Returns:
            Self for method chaining

        Raises:
            ValueError: If neither implementation nor factory provided
            ValueError: If duplicate interface registration without allow_override
        """
        interface_name = interface.__name__

        # Check for duplicate registration (REQ-024-010)
        with self._lock:
            if interface_name in self._registry and not self._allow_override:
                raise ValueError(f"Interface {interface_name} already registered")

        self._validate_registration(interface_name, interface, implementation, factory)

        # Shallow copy config to prevent mutation issues (REQ-024-007)
        config_copy = copy(config) if config else {}

        # Use explicit dependencies or extract from constructor
        dependencies_list = (
            dependencies
            if dependencies is not None
            else (self._extract_dependencies(implementation) if implementation else [])
        )

        registration = ComponentRegistration(
            interface=interface,
            implementation=implementation,
            scope=scope,
            config=config_copy,
            dependencies=dependencies_list,
            factory=factory,
        )

        with self._lock:
            self._registry[interface_name] = registration
        self._logger.debug(
            f"Registered {interface_name} -> "
            f"{implementation.__name__ if implementation else 'factory'}"
        )

        return self  # REQ-024-009: Return self for chaining

    @traces(
        spec="024",
        story="Core Registration",
        reqs=["REQ-024-005", "REQ-024-006"],
    )
    def _validate_registration(
        self,
        interface_name: str,
        interface: type,
        implementation: type | None,
        factory: Callable | None,
    ) -> None:
        """Validate that the registration arguments are consistent.

        Args:
            interface_name: String name of the interface (for messages).
            interface: The interface type.
            implementation: The implementation type, if provided.
            factory: The factory callable, if provided.

        Raises:
            ValueError: If neither implementation nor factory is provided,
                        or if implementation does not implement interface.
        """
        if implementation is None and factory is None:
            raise ValueError(
                f"Must provide either implementation or factory for {interface_name}"
            )
        if implementation is not None and not issubclass(implementation, interface):
            raise ValueError(
                f"{implementation.__name__} must implement {interface_name}"
            )

    @traces(
        spec="024",
        story="Instance Registration",
        reqs=[
            "REQ-024-036",
            "REQ-024-037",
            "REQ-024-039",
        ],
    )
    def register_instance(self, interface: type[T], instance: T) -> DIContainer:
        """
        Register an existing instance.

        Args:
            interface: Interface class
            instance: Component instance

        Returns:
            Self for method chaining

        Raises:
            ValueError: If instance does not implement interface
        """
        interface_name = interface.__name__

        if not isinstance(instance, interface):
            raise ValueError(f"Instance must implement {interface_name}")

        # Create a factory that returns the instance
        @traces(spec="024", story="Instance Registration", reqs=["REQ-024-038"])
        def factory() -> T:
            return instance

        self.register(
            interface=interface, factory=factory, scope=ComponentScope.SINGLETON
        )

        return self  # REQ-024-040: Return self for chaining

    @traces(
        spec="024",
        story="Resolution",
        reqs=["REQ-024-011", "REQ-024-012", "REQ-024-023", "REQ-024-052"],
    )
    async def resolve(self, interface: type[T], context_id: str | None = None) -> T:
        """
        Resolve component instance.

        Args:
            interface: Interface to resolve
            context_id: Context ID for REQUEST scope isolation

        Returns:
            Component instance

        Raises:
            ValueError: If interface not registered
            RuntimeError: If circular dependency detected
        """
        # `interface_name` is typed `Any` (not `str`): `.__name__` is
        # normally a plain string, but the REQ-024-023 generic-stripping
        # check below intentionally probes it (not `interface` itself) for
        # `__origin__`, since some dynamic/generic-alias types return a
        # name-like object carrying its own `__origin__` rather than a bare
        # str (see test_resolve_with_generic_alias_type).
        interface_name: Any = interface.__name__

        # Handle generic type parameters (REQ-024-023)
        # For now, strip generic parameters and use base type
        if hasattr(interface_name, "__origin__"):
            actual_interface = interface_name.__origin__
            interface_name = actual_interface.__name__

        # Snapshot registry under lock; resolution itself is lock-free
        with self._lock:
            if interface_name not in self._registry:
                raise ValueError(f"Interface {interface_name} not registered")

        # Create context with ID for REQUEST scope (REQ-024-015, REQ-024-025)
        context = ResolutionContext()
        if context_id:
            context.context_id = context_id
            # Use context_id-based cache for REQUEST scope
            cache_key = f"{interface_name}:{context_id}"
            if cache_key in self._context:
                return self._context[cache_key]
        else:
            context.context_id = "default"
            cache_key = interface_name

        try:
            instance = await self._resolve_with_context(interface_name, context)
            # The string-keyed internal registry can't carry the TypeVar
            # binding through `_resolve_with_context`'s `IComponent` return
            # type; the caller's `interface: type[T]` is what guarantees the
            # resolved instance is actually a `T` (assuming a correctly
            # populated registry).
            return cast(T, instance)
        except KeyError:
            raise ValueError(f"Interface {interface_name} not registered")
        except RecursionError:
            raise RuntimeError(
                f"Circular dependency detected: {' -> '.join(context.resolving)}"
            )

    @traces(
        spec="024",
        story="Resolution",
        reqs=[
            "REQ-024-014",
            "REQ-024-016",
            "REQ-024-026",
            "REQ-024-027",
            "REQ-024-028",
            "REQ-024-029",
            "REQ-024-030",
        ],
    )
    async def _resolve_with_context(
        self, interface_name: str, context: ResolutionContext
    ) -> IComponent:
        """
        Resolve component with context.

        Args:
            interface_name: Interface name to resolve
            context: Resolution context

        Returns:
            Component instance
        """
        # Circular dependency detection (REQ-024-026, REQ-024-027, REQ-024-028)
        if interface_name in context.resolving:
            cycle = " -> ".join(context.resolving) + f" -> {interface_name}"
            raise RecursionError(f"Circular dependency: {cycle}")

        # Check cache/singletons (REQ-024-013, REQ-024-015)
        registration = self._registry[interface_name]

        cached = self._cached_instance(interface_name, registration, context)
        if cached is not None:
            return cached

        # For TRANSIENT (REQ-024-014): Always create new instance (no caching)
        context.resolving.append(interface_name)

        try:
            instance = await self._create_instance(registration, context)

            # Call initialize() for IComponent instances (REQ-024-016)
            if isinstance(instance, IComponent):
                await instance.initialize(registration.config)

            self._store_instance(interface_name, registration, context, instance)
            return instance

        finally:
            # Unwind resolving stack (REQ-024-030)
            context.resolving.pop()

    @traces(spec="024", story="Resolution", reqs=["REQ-024-013", "REQ-024-015"])
    def _cached_instance(
        self,
        interface_name: str,
        registration: ComponentRegistration,
        context: ResolutionContext,
    ) -> IComponent | None:
        """Return a cached instance for SINGLETON/REQUEST scopes, else None."""
        # For SINGLETON (REQ-024-013): global singleton cache
        if registration.scope == ComponentScope.SINGLETON:
            return self._singletons.get(interface_name)
        # For REQUEST scope (REQ-024-015): context-specific cache
        if registration.scope == ComponentScope.REQUEST:
            cache_key = f"{interface_name}:{context.context_id}"
            return context.cache.get(cache_key)
        return None

    @traces(spec="024", story="Resolution", reqs=["REQ-024-017", "REQ-024-053"])
    def _store_instance(
        self,
        interface_name: str,
        registration: ComponentRegistration,
        context: ResolutionContext,
        instance: IComponent,
    ) -> None:
        """Cache a freshly created instance according to its scope (REQ-024-017)."""
        if registration.scope == ComponentScope.SINGLETON:
            # Double-checked locking pattern (REQ-024-053)
            if interface_name not in self._singletons:
                with self._lock:
                    if interface_name not in self._singletons:
                        self._singletons[interface_name] = instance
        elif registration.scope == ComponentScope.REQUEST:
            cache_key = f"{interface_name}:{context.context_id}"
            context.cache[cache_key] = instance

    @traces(
        spec="024",
        story="Resolution",
        reqs=["REQ-024-020", "REQ-024-021", "REQ-024-024"],
    )
    async def _create_instance(
        self,
        registration: ComponentRegistration,
        context: ResolutionContext,
    ) -> Any:
        """Instantiate a component from its registration.

        Uses the factory if present, otherwise resolves constructor
        dependencies and calls the implementation class.

        Args:
            registration: Component registration.
            context: Current resolution context.

        Returns:
            New component instance (not yet initialized).

        Raises:
            Exceptions from factory or initialization propagate (REQ-024-024)
        """
        # Use factory if registered (REQ-024-020)
        if registration.factory is not None:
            # Pass config to factory if it accepts it (REQ-024-021)
            sig = inspect.signature(registration.factory)
            if "config" in sig.parameters or "kwargs" in sig.parameters:
                return registration.factory(config=registration.config)
            return registration.factory()

        dependencies = await self._resolve_dependencies(registration, context)
        if registration.implementation is None:
            # Unreachable via the public API: register()'s own validation
            # (_validate_registration) already requires implementation or
            # factory. This guards direct/programmatic construction of an
            # invalid ComponentRegistration (e.g. in tests) and narrows the
            # type for the call below.
            raise ValueError(
                f"No implementation or factory for {registration.interface.__name__}"
            )
        return registration.implementation(**dependencies)

    @traces(spec="024", story="Resolution", reqs=["REQ-024-018"])
    async def _resolve_dependencies(
        self, registration: ComponentRegistration, context: ResolutionContext
    ) -> dict[str, Any]:
        """
        Resolve constructor dependencies by inspecting signature.

        Args:
            registration: Component registration
            context: Resolution context

        Returns:
            Dictionary of resolved dependencies
        """
        dependencies: dict[str, Any] = {}

        if registration.implementation is None:
            return dependencies

        constructor = registration.implementation.__init__
        signature = inspect.signature(constructor)

        for param_name, param in signature.parameters.items():
            if param_name == "self":
                continue
            dependencies[param_name] = await self._resolve_param(
                param_name, param, registration, context
            )

        return dependencies

    @traces(spec="024", story="Resolution", reqs=["REQ-024-019"])
    async def _resolve_param(
        self,
        param_name: str,
        param: inspect.Parameter,
        registration: ComponentRegistration,
        context: ResolutionContext,
    ) -> Any:
        """Resolve a single constructor parameter.

        Skips parameters without type annotations or with built-in types.

        Args:
            param_name: Name of the parameter.
            param: The inspect.Parameter object.
            registration: The component registration being constructed.
            context: Current resolution context.

        Returns:
            Resolved value for the parameter.

        Raises:
            ValueError: If the parameter cannot be resolved.
        """
        param_type = param.annotation

        # Skip untyped parameters or use context default (REQ-024-019)
        if param_type == inspect.Parameter.empty:
            if param_name in self._context:
                return self._context[param_name]
            if param.default != inspect.Parameter.empty:
                return param.default
            raise ValueError(f"Cannot resolve untyped parameter: {param_name}")

        # Resolve IComponent interfaces recursively
        if inspect.isclass(param_type) and issubclass(param_type, IComponent):
            return await self._resolve_with_context(param_type.__name__, context)

        # Use context values or defaults
        if param_name in self._context:
            return self._context[param_name]
        if param.default != inspect.Parameter.empty:
            return param.default

        # Try config values
        config_key = f"{registration.interface.__name__.lower()}.{param_name}"
        if config_key in registration.config:
            return registration.config[config_key]

        raise ValueError(f"Cannot resolve parameter: {param_name}")

    def _extract_dependencies(self, implementation: type[IComponent]) -> list[str]:
        """
        Extract dependency interfaces from constructor.

        Args:
            implementation: Implementation class

        Returns:
            List of dependency interface names
        """
        dependencies = []

        constructor = implementation.__init__
        signature = inspect.signature(constructor)

        for param_name, param in signature.parameters.items():
            if param_name == "self":
                continue

            param_type = param.annotation
            if (
                param_type != inspect.Parameter.empty
                and inspect.isclass(param_type)
                and issubclass(param_type, IComponent)
            ):
                dependencies.append(param_type.__name__)

        return dependencies

    @traces(spec="024", story="Introspection", reqs=["REQ-024-049"])
    def set_context(self, key: str, value: Any) -> None:
        """
        Set context value for dependency resolution.

        Args:
            key: Context key
            value: Context value
        """
        self._context[key] = value

    @traces(spec="024", story="Introspection", reqs=["REQ-024-050"])
    def get_context(self, key: str, default: Any = None) -> Any:
        """
        Get context value.

        Args:
            key: Context key
            default: Default value if key not found

        Returns:
            Context value
        """
        return self._context.get(key, default)

    async def shutdown(self) -> None:
        """Shutdown all components in reverse dependency order.

        Stops both context-tracked components (e.g. values stashed via
        ``set_context``) and SINGLETON-scope components resolved through
        ``resolve()`` — those live in ``self._singletons``, a separate cache
        from ``self._context`` (defect found during D4 audit: this method
        used to iterate only ``self._context``, so a singleton resolved via
        the normal ``resolve()`` path never had ``stop()`` called; only
        components manually stashed into ``self._context`` were reached).
        """
        self._logger.info("Shutting down DI container")

        stopped_ids: set[int] = set()

        # Shutdown context-tracked components (original behavior, unchanged).
        for interface_name, instance in list(self._context.items()):
            if isinstance(instance, IComponent):
                stopped_ids.add(id(instance))
                try:
                    await instance.stop()
                except Exception as e:
                    self._logger.error(f"Error stopping {interface_name}: {e}")

        # Shutdown SINGLETON-scope components resolved via resolve(), in
        # reverse (most-recently-resolved first). Skip anything already
        # stopped above to avoid calling stop() twice on the same instance.
        for interface_name, instance in reversed(list(self._singletons.items())):
            if id(instance) in stopped_ids:
                continue
            if isinstance(instance, IComponent):
                try:
                    await instance.stop()
                except Exception as e:
                    self._logger.error(f"Error stopping {interface_name}: {e}")

        self._context.clear()
        self._singletons.clear()

    @traces(spec="024", story="Introspection", reqs=["REQ-024-046"])
    def get_registry_info(self) -> dict[str, Any]:
        """
        Get information about registered components.

        Returns:
            Dictionary with registry information
        """
        info = {
            "name": self.name,
            "registered_components": len(self._registry),
            "components": {},
        }

        for interface_name, registration in self._registry.items():
            info["components"][interface_name] = {
                "implementation": (
                    registration.implementation.__name__
                    if registration.implementation
                    else "factory"
                ),
                "scope": registration.scope.value,
                "dependencies": registration.dependencies,
                "has_config": bool(registration.config),
            }

        return info

    @traces(spec="024", story="Introspection", reqs=["REQ-024-047"])
    def has(self, interface: type[T]) -> bool:
        """
        Check if interface has a registration.

        Args:
            interface: Interface to check

        Returns:
            True if registered, False otherwise
        """
        interface_name = interface.__name__
        with self._lock:
            return interface_name in self._registry

    @traces(spec="024", story="Introspection", reqs=["REQ-024-048"])
    def get_all_registrations(self) -> dict[str, ComponentRegistration]:
        """
        Get raw registration mapping.

        Returns:
            Dictionary of interface names to ComponentRegistration
        """
        with self._lock:
            return self._registry.copy()


# ============================================================================
# Container Builder
# ============================================================================


class ContainerBuilder:
    """
    Fluent builder for configuring DI containers.
    """

    def __init__(self, name: str = "default", allow_override: bool = False) -> None:
        """
        Initialize container builder.

        Args:
            name: Container name
            allow_override: If True, allows re-registering same interface
        """
        self._container = DIContainer(name, allow_override=allow_override)
        self._configs: dict[str, dict[str, Any]] = {}

    def with_config(self, config: dict[str, Any]) -> ContainerBuilder:
        """
        Add configuration to container.

        Args:
            config: Configuration dictionary

        Returns:
            Self for chaining
        """
        self._configs.update(config)
        return self

    @traces(spec="024", story="Container Builder", reqs=["REQ-024-031"])
    def with_component(
        self,
        interface: type[T],
        implementation: type[T] | None = None,
        *,
        scope: ComponentScope = ComponentScope.SINGLETON,
        config_key: str | None = None,
        factory: Callable[..., T] | None = None,
        dependencies: list[str] | None = None,
    ) -> ContainerBuilder:
        """
        Register a component.

        Args:
            interface: Interface class
            implementation: Implementation class
            scope: Component scope
            config_key: Key in config for component-specific config
            factory: Factory function
            dependencies: Explicit dependency list

        Returns:
            Self for chaining
        """
        # Get component config
        config = {}
        if config_key and config_key in self._configs:
            config = self._configs[config_key]
        elif interface.__name__.lower() in self._configs:
            config = self._configs[interface.__name__.lower()]

        self._container.register(
            interface=interface,
            implementation=implementation,
            scope=scope,
            config=config,
            factory=factory,
            dependencies=dependencies,
        )

        return self

    def with_instance(self, interface: type[T], instance: T) -> ContainerBuilder:
        """
        Register an existing instance.

        Args:
            interface: Interface class
            instance: Component instance

        Returns:
            Self for chaining
        """
        self._container.register_instance(interface, instance)
        return self

    def with_context(self, key: str, value: Any) -> ContainerBuilder:
        """
        Add context value.

        Args:
            key: Context key
            value: Context value

        Returns:
            Self for chaining
        """
        self._container.set_context(key, value)
        return self

    @traces(
        spec="024",
        story="Container Builder",
        reqs=["REQ-024-032", "REQ-024-034", "REQ-024-035"],
    )
    def build(self) -> DIContainer:
        """
        Build the configured container.

        Validates registrations and returns the container this builder has
        been assembling.

        Note: this returns the SAME ``DIContainer`` instance the builder has
        been mutating since ``__init__`` (not an independent copy) — further
        ``with_*`` calls on this builder after ``build()`` still mutate the
        container already handed to the caller. (REQ-024-035 originally
        promised an "independent" container; this comment was corrected to
        match actual behavior during the D4 audit rather than changing the
        long-standing return-same-instance contract multiple call sites
        depend on.)

        Returns:
            The configured DI container (same object across repeat build() calls)

        Raises:
            ValueError: If registrations are invalid
        """
        # Validate registrations (REQ-024-034)
        for interface_name, registration in self._container._registry.items():
            if registration.implementation is None and registration.factory is None:
                raise ValueError(
                    f"Invalid registration for {interface_name}: no implementation or factory"
                )
            if registration.implementation is not None and not issubclass(
                registration.implementation, registration.interface
            ):
                raise ValueError(
                    f"Invalid registration for {interface_name}: implementation doesn't implement interface"
                )

        return self._container


# ============================================================================
# Global Container Registry
# ============================================================================


@traces(
    spec="024",
    story="Container Registry",
    reqs=["REQ-024-045"],
)
class ContainerRegistry:
    """
    Global registry for DI containers.

    Allows accessing containers by name from anywhere in the application.
    """

    _containers: dict[str, DIContainer] = {}

    @classmethod
    @traces(spec="024", story="Container Registry", reqs=["REQ-024-041"])
    def register(cls, name: str, container: DIContainer) -> None:
        """
        Register a container.

        Args:
            name: Container name
            container: DI container
        """
        cls._containers[name] = container

    @classmethod
    @traces(spec="024", story="Container Registry", reqs=["REQ-024-042"])
    def get(cls, name: str = "default") -> DIContainer:
        """
        Get container by name.

        Args:
            name: Container name

        Returns:
            DI container

        Raises:
            KeyError: If container not found
        """
        if name not in cls._containers:
            raise KeyError(f"Container '{name}' not registered")
        return cls._containers[name]

    @classmethod
    @traces(spec="024", story="Container Registry", reqs=["REQ-024-043"])
    def has(cls, name: str) -> bool:
        """
        Check if container is registered.

        Args:
            name: Container name

        Returns:
            True if registered, False otherwise
        """
        return name in cls._containers

    @classmethod
    async def shutdown_all(cls) -> None:
        """Shutdown all registered containers."""
        for container in cls._containers.values():
            await container.shutdown()
        cls._containers.clear()