PicoContainer API Reference¶

class PicoContainer(_ResolutionMixin):
    """The pico-ioc dependency injection container.

    Created by :func:`init`. Provides synchronous (``get``) and asynchronous
    (``aget``) resolution, scope management, lifecycle hooks, health checks,
    and dependency graph export.

    Args:
        component_factory: The :class:`ComponentFactory` holding all provider
            bindings.
        caches: The :class:`ScopedCaches` instance managing instance storage.
        scopes: The :class:`ScopeManager` coordinating scope activation.
        observers: Optional list of :class:`ContainerObserver` instances.
        container_id: Optional explicit container identifier.
        profiles: Active profile names.
    """

    _container_id_var: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar("pico_container_id", default=None)
    _container_registry: Dict[str, "PicoContainer"] = {}

    class _Ctx:
        def __init__(self, container_id: str, profiles: Tuple[str, ...], created_at: float) -> None:
            self.container_id = container_id
            self.profiles = profiles
            self.created_at = created_at
            self.resolve_count = 0
            self.cache_hit_count = 0

    def __init__(
        self,
        component_factory: ComponentFactory,
        caches: ScopedCaches,
        scopes: ScopeManager,
        observers: Optional[List["ContainerObserver"]] = None,
        container_id: Optional[str] = None,
        profiles: Tuple[str, ...] = (),
    ) -> None:
        self._factory = component_factory
        self._caches = caches
        self.scopes = scopes
        self._locator: Optional[ComponentLocator] = None
        self._observers = list(observers or [])
        self.container_id = container_id or self._generate_container_id()
        import time as _t

        self.context = PicoContainer._Ctx(container_id=self.container_id, profiles=profiles, created_at=_t.time())
        PicoContainer._container_registry[self.container_id] = self

    @staticmethod
    def _generate_container_id() -> str:
        import secrets
        import time as _t

        return f"c{_t.time_ns():x}{secrets.randbits(16):04x}"

    @classmethod
    def get_current(cls) -> Optional["PicoContainer"]:
        """Return the container that is active in the current context, or ``None``."""
        cid = cls._container_id_var.get()
        return cls._container_registry.get(cid) if cid else None

    @classmethod
    def get_current_id(cls) -> Optional[str]:
        """Return the container ID that is active in the current context, or ``None``."""
        return cls._container_id_var.get()

    @classmethod
    def all_containers(cls) -> Dict[str, "PicoContainer"]:
        """Return a snapshot dict of all live containers, keyed by container ID."""
        return dict(cls._container_registry)

    def activate(self) -> contextvars.Token:
        return PicoContainer._container_id_var.set(self.container_id)

    def deactivate(self, token: contextvars.Token) -> None:
        PicoContainer._container_id_var.reset(token)

    @contextmanager
    def as_current(self):
        token = self.activate()
        try:
            yield self
        finally:
            self.deactivate(token)

    def attach_locator(self, locator: ComponentLocator) -> None:
        self._locator = locator

    def _cache_for(self, key: KeyT):
        md = self._locator._metadata.get(key) if self._locator else None
        sc = md.scope if md else SCOPE_SINGLETON
        return self._caches.for_scope(self.scopes, sc)

    def has(self, key: KeyT) -> bool:
        """Check whether a component is registered for *key*.

        Args:
            key: The resolution key (type or string).

        Returns:
            ``True`` if a provider or cached instance exists for *key*.
        """
        cache = self._cache_for(key)
        return cache.get(key) is not None or self._factory.has(key)

    def _canonical_key(self, key: KeyT) -> KeyT:
        if self._factory.has(key):
            return key

        if isinstance(key, type) and self._locator:
            cands: List[Tuple[bool, Any]] = []
            for k, md in self._locator._metadata.items():
                typ = md.provided_type or md.concrete_class
                if not isinstance(typ, type):
                    continue
                try:
                    if typ is not key and issubclass(typ, key):
                        cands.append((md.primary, k))
                except Exception:
                    continue
            if cands:
                prim = [k for is_p, k in cands if is_p]
                return prim[0] if prim else cands[0][1]

        if isinstance(key, str) and self._locator:
            for k, md in self._locator._metadata.items():
                if md.pico_name == key:
                    return k

        return key

    def _resolve_or_create_internal(self, key: KeyT) -> Tuple[Any, float, bool]:
        key = self._canonical_key(key)
        cache = self._cache_for(key)
        cached = cache.get(key)

        if cached is not None:
            self.context.cache_hit_count += 1
            for o in self._observers:
                o.on_cache_hit(key)
            return cached, 0.0, True

        import time as _tm

        t0 = _tm.perf_counter()

        token_container = self.activate()
        requester = None

        try:
            provider = self._factory.get(key, origin=requester)
            try:
                instance_or_awaitable = provider()
            except ProviderNotFoundError:
                raise
            except Exception as creation_error:
                raise ComponentCreationError(key, creation_error) from creation_error

            took_ms = (_tm.perf_counter() - t0) * 1000
            return instance_or_awaitable, took_ms, False

        finally:
            self.deactivate(token_container)

    def _run_configure_methods(self, instance: Any) -> Any:
        if not _needs_async_configure(instance):
            for m in _iter_configure_methods(instance):
                configure_deps = analyze_callable_dependencies(m)
                args = self._resolve_args(configure_deps)
                res = m(**args)
                if inspect.isawaitable(res):
                    raise AsyncResolutionError(
                        f"Component {type(instance).__name__} returned an awaitable from synchronous "
                        f"@configure method '{m.__name__}'. You must use 'await container.aget()' "
                        "or make the method synchronous."
                    )
            return instance

        async def runner():
            for m in _iter_configure_methods(instance):
                configure_deps = analyze_callable_dependencies(m)
                args = self._resolve_args(configure_deps)
                r = m(**args)
                if inspect.isawaitable(r):
                    await r
            return instance

        return runner()

    @overload
    def get(self, key: type) -> Any: ...
    @overload
    def get(self, key: str) -> Any: ...
    def get(self, key: KeyT) -> Any:
        """Resolve a component synchronously.

        Args:
            key: The type or string key to resolve.

        Returns:
            The component instance.

        Raises:
            ProviderNotFoundError: If no provider is bound to *key*.
            AsyncResolutionError: If the provider returns an awaitable
                (use :meth:`aget` instead).
            ComponentCreationError: If the provider fails during creation.
        """
        instance_or_awaitable, took_ms, was_cached = self._resolve_or_create_internal(key)

        if was_cached:
            return instance_or_awaitable

        instance = instance_or_awaitable
        if inspect.isawaitable(instance):
            raise AsyncResolutionError(key)

        md = self._locator._metadata.get(key) if self._locator else None
        scope = md.scope if md else SCOPE_SINGLETON
        if scope != SCOPE_SINGLETON:
            instance_or_awaitable_configured = self._run_configure_methods(instance)
            if inspect.isawaitable(instance_or_awaitable_configured):
                raise AsyncResolutionError(key)
            instance = instance_or_awaitable_configured

        final_instance = self._maybe_wrap_with_aspects(key, instance)
        cache = self._cache_for(key)
        cache.put(key, final_instance)
        self.context.resolve_count += 1
        for o in self._observers:
            o.on_resolve(key, took_ms)

        return final_instance

    async def aget(self, key: KeyT) -> Any:
        """Resolve a component asynchronously.

        Awaits any coroutine returned by the provider, ``__ainit__``, and
        async ``@configure`` methods. Always safe to use, even for sync
        components.

        Args:
            key: The type or string key to resolve.

        Returns:
            The component instance.

        Raises:
            ProviderNotFoundError: If no provider is bound to *key*.
            ComponentCreationError: If the provider fails during creation.
        """
        instance_or_awaitable, took_ms, was_cached = self._resolve_or_create_internal(key)

        instance = instance_or_awaitable
        if was_cached:
            if isinstance(instance, UnifiedComponentProxy):
                await instance._async_init_if_needed()
            return instance

        if inspect.isawaitable(instance_or_awaitable):
            instance = await instance_or_awaitable

        md = self._locator._metadata.get(key) if self._locator else None
        scope = md.scope if md else SCOPE_SINGLETON
        if scope != SCOPE_SINGLETON:
            instance_or_awaitable_configured = self._run_configure_methods(instance)
            if inspect.isawaitable(instance_or_awaitable_configured):
                instance = await instance_or_awaitable_configured
            else:
                instance = instance_or_awaitable_configured

        final_instance = self._maybe_wrap_with_aspects(key, instance)

        if isinstance(final_instance, UnifiedComponentProxy):
            await final_instance._async_init_if_needed()

        cache = self._cache_for(key)
        cache.put(key, final_instance)
        self.context.resolve_count += 1
        for o in self._observers:
            o.on_resolve(key, took_ms)

        return final_instance

    def _iterate_cleanup_targets(self) -> Iterable[Any]:
        yield from (obj for _, obj in self._caches.all_items())

        if self._locator:
            seen = set()
            for md in self._locator._metadata.values():
                fc = md.factory_class
                if fc and fc not in seen:
                    seen.add(fc)
                    inst = self.get(fc) if self._factory.has(fc) else fc()
                    yield inst

    def _call_cleanup_method(self, method: Callable[..., Any]) -> Any:
        deps_requests = analyze_callable_dependencies(method)
        return method(**self._resolve_args(deps_requests))

    def cleanup_all(self) -> None:
        """Invoke all ``@cleanup`` methods on cached components (sync)."""
        for obj in self._iterate_cleanup_targets():
            for _, m in inspect.getmembers(obj, predicate=inspect.ismethod):
                meta = getattr(m, PICO_META, {})
                if meta.get("cleanup", False):
                    res = self._call_cleanup_method(m)
                    if inspect.isawaitable(res):
                        LOGGER.warning(f"Async cleanup method {m} called during sync shutdown. Awaitable ignored.")

    async def cleanup_all_async(self) -> None:
        """Invoke all ``@cleanup`` methods on cached components (async).

        Awaits async cleanup methods and closes the :class:`EventBus` if present.
        """
        for obj in self._iterate_cleanup_targets():
            for _, m in inspect.getmembers(obj, predicate=inspect.ismethod):
                meta = getattr(m, PICO_META, {})
                if meta.get("cleanup", False):
                    res = self._call_cleanup_method(m)
                    if inspect.isawaitable(res):
                        await res

        try:
            from .event_bus import EventBus

            for _, obj in self._caches.all_items():
                if isinstance(obj, EventBus):
                    await obj.aclose()
        except Exception:
            pass

    def activate_scope(self, name: str, scope_id: Any):
        return self.scopes.activate(name, scope_id)

    def deactivate_scope(self, name: str, token: Optional[contextvars.Token]) -> None:
        self.scopes.deactivate(name, token)

    def info(self, msg: str) -> None:
        LOGGER.info(f"[{self.container_id[:8]}] {msg}")

    @contextmanager
    def scope(self, name: str, scope_id: Any):
        """Context manager that activates a scope for the duration of a block.

        Args:
            name: The scope name (e.g. ``"request"``).
            scope_id: A unique identifier for this scope instance.

        Yields:
            This container instance.

        Example:
            >>> with container.scope("request", request_id):
            ...     ctx = container.get(RequestContext)
        """
        tok = self.activate_scope(name, scope_id)
        try:
            yield self
        finally:
            self.deactivate_scope(name, tok)

    def health_check(self) -> Dict[str, bool]:
        """Run all ``@health``-decorated methods and return their results.

        Returns:
            Dict mapping ``"ClassName.method_name"`` to ``True``/``False``.
        """
        out: Dict[str, bool] = {}
        for k, obj in self._caches.all_items():
            for name, m in inspect.getmembers(obj, predicate=callable):
                if getattr(m, PICO_META, {}).get("health_check", False):
                    try:
                        out[f"{getattr(k, '__name__', k)}.{name}"] = bool(m())
                    except Exception:
                        out[f"{getattr(k, '__name__', k)}.{name}"] = False
        return out

    def stats(self) -> Dict[str, Any]:
        """Return container statistics.

        Returns:
            Dict with keys ``container_id``, ``profiles``,
            ``uptime_seconds``, ``total_resolves``, ``cache_hits``,
            ``cache_hit_rate``, and ``registered_components``.
        """
        import time as _t

        resolves = self.context.resolve_count
        hits = self.context.cache_hit_count
        total = resolves + hits
        return {
            "container_id": self.container_id,
            "profiles": self.context.profiles,
            "uptime_seconds": _t.time() - self.context.created_at,
            "total_resolves": resolves,
            "cache_hits": hits,
            "cache_hit_rate": (hits / total) if total > 0 else 0.0,
            "registered_components": len(self._locator._metadata) if self._locator else 0,
        }

    def shutdown(self) -> None:
        """Synchronously shut down the container.

        Calls all ``@cleanup`` methods and removes the container from the
        global registry.
        """
        self.cleanup_all()
        PicoContainer._container_registry.pop(self.container_id, None)

    async def ashutdown(self) -> None:
        """Asynchronously shut down the container.

        Awaits async ``@cleanup`` methods, closes the ``EventBus``, and
        removes the container from the global registry.
        """
        await self.cleanup_all_async()
        PicoContainer._container_registry.pop(self.container_id, None)

    def build_resolution_graph(self):
        """Build the static dependency graph from registered metadata.

        Returns:
            Dict mapping each key to a tuple of its dependency keys.
        """
        return _build_resolution_graph(self._locator)

    def export_graph(
        self,
        path: str,
        *,
        include_scopes: bool = True,
        include_qualifiers: bool = False,
        rankdir: str = "LR",
        title: Optional[str] = None,
    ) -> None:
        """Export the dependency graph as a Graphviz DOT file.

        Args:
            path: Filesystem path for the output ``.dot`` file.
            include_scopes: Annotate nodes with their scope.
            include_qualifiers: Annotate nodes with their qualifiers.
            rankdir: Graph layout direction (``"LR"`` or ``"TB"``).
            title: Optional graph title.

        Raises:
            RuntimeError: If no locator is attached to the container.
        """
        _export_graph(
            self._locator,
            path,
            include_scopes=include_scopes,
            include_qualifiers=include_qualifiers,
            rankdir=rankdir,
            title=title,
        )