Di Module API Reference¶

def configure(self, binder: Binder) -> None:
    """Binds the `ConfigProtocol` to its default `Config` implementation.

    Args:
        binder (Binder): The `injector` binder instance.

    Returns:
        None:
    """
    binder.bind(ConfigProtocol, to=Config, scope=singleton)

@classmethod
def current(cls) -> DIContainer:
    """Returns the active singleton instance of the `DIContainer`.

    Returns:
        DIContainer: The singleton instance.
    """
    with cls._lock:
        if cls._instance is None:
            cls._instance = cls()
            cls._log_static(
                logging.WARNING, "DIContainer.current auto-initialized singleton"
            )
        return cls._instance

def factories(self) -> Sequence[tuple[type[Any] | str, str | None]]:
    """Returns a list of all registered factory keys.

    Returns:
        Sequence[tuple[type[Any] | str, str | None]]: A sequence of
            (key, name) tuples for all registered factories.
    """
    with self._lock:
        return list(self._store.keys())

@contextmanager
def override(
    self,
    key: type[T] | str,
    factory_or_value: Callable[[], T | Awaitable[T]] | T,
    name: str | None = None,
) -> Iterator[None]:
    """Temporarily overrides a service registration within a context block.

    This is primarily useful for testing, allowing a service to be replaced
    with a mock or stub. The original registration is restored upon exiting
    the context.

    Args:
        key (type[T] | str): The service key to override.
        factory_or_value: The temporary factory or value.
        name (str | None): An optional name for the registration.

    Yields:
        None:
    """
    with self._lock:
        store_key = (key, name)
        original_factory = self._store.get(store_key)
        original_instance = self._services.get(store_key)
        self.register(key, factory_or_value, name)
        if store_key in self._services:
            del self._services[store_key]
        self._log(
            logging.DEBUG,
            "Overriding service",
            extra={"service_name": _key_name(key), "svc_alias": name},
        )
    try:
        yield
    finally:
        with self._lock:
            if original_factory is not None:
                self._store[store_key] = original_factory
                if original_instance is not None:
                    self._services[store_key] = original_instance
                else:
                    self._services.pop(store_key, None)
                self._log(
                    logging.DEBUG,
                    "Restored service",
                    extra={"service_name": _key_name(key), "svc_alias": name},
                )
            else:
                self.unregister(key, name)
                self._log(
                    logging.DEBUG,
                    "Removed service override",
                    extra={"service_name": _key_name(key), "svc_alias": name},
                )

def register(
    self,
    key: type[T] | str,
    factory_or_value: Callable[[], T | Awaitable[T]] | T,
    name: str | None = None,
) -> None:
    """Registers a factory or a pre-resolved value for a given service key.

    Args:
        key (type[T] | str): The service key, which can be a protocol type
            or a unique string identifier.
        factory_or_value: The factory function that creates the service,
            or the service instance itself.
        name (str | None): An optional name for the registration, allowing
            multiple implementations of the same key.

    Returns:
        None:

    Raises:
        BijuxError: If the registration key is invalid or conflicts with an
            existing registration.
    """
    if not (isinstance(key, str) or inspect.isclass(key)):
        raise BijuxError("Service key must be a type or str", http_status=400)
    try:
        store_key = (key, name)
        if isinstance(key, str) and any(
            isinstance(k, type) and k.__name__ == key for k, _ in self._store
        ):
            raise BijuxError(
                f"Key {key} conflicts with existing type name", http_status=400
            )
        if isinstance(key, type) and any(k == key.__name__ for k, _ in self._store):
            raise BijuxError(
                f"Type {key.__name__} conflicts with existing string key",
                http_status=400,
            )
        self._store[store_key] = factory_or_value
        if isinstance(factory_or_value, ObservabilityProtocol) and not isinstance(
            factory_or_value, type
        ):
            self._obs = factory_or_value
        self._log(
            logging.DEBUG,
            "Registered service",
            extra={"service_name": _key_name(key), "svc_alias": name},
        )
    except (TypeError, KeyError) as exc:
        self._log(
            logging.ERROR,
            f"Failed to register service: {exc}",
            extra={"service_name": _key_name(key), "name": name},
        )
        raise BijuxError(
            f"Failed to register service {_key_name(key)}: {exc}", http_status=400
        ) from exc

@classmethod
def reset(cls) -> None:
    """Resets the singleton instance, shutting down all services.

    This method is intended for use in testing environments to ensure a
    clean state between tests. It clears all registered services and
    factories.
    """
    inst = None
    with cls._lock:
        inst = cls._instance
        cls._instance = None
        cls._obs = None
    if inst is None:
        cls._log_static(logging.DEBUG, "DIContainer reset (no instance)")
        return
    try:
        asyncio.run(inst.shutdown())
    except Exception as exc:
        cls._log_static(logging.ERROR, f"Error during shutdown: {exc}")
    inst._services.clear()
    inst._store.clear()
    inst._obs = None
    cls._log_static(logging.DEBUG, "DIContainer reset")

@classmethod
async def reset_async(cls) -> None:
    """Asynchronously resets the singleton instance.

    This method is intended for use in testing environments. All services
    and factories are cleared.
    """
    instance = None
    with cls._lock:
        if cls._instance is not None:
            instance = cls._instance
            cls._instance = None
            cls._obs = None
    if instance is not None:
        await instance.shutdown()
        instance._services.clear()
        instance._store.clear()
        instance._obs = None
    cls._log_static(logging.DEBUG, "DIContainer reset")

def resolve(self, key: type[T] | str, name: str | None = None) -> T:
    """Resolves and returns a service instance synchronously.

    If the service factory is asynchronous, this method will run the
    async factory to completion.

    Args:
        key (type[T] | str): The service key to resolve.
        name (str | None): An optional name for the registration.

    Returns:
        T: The resolved service instance.

    Raises:
        KeyError: If the service is not registered.
        BijuxError: If the factory fails, returns None, or a circular
            dependency is detected.
    """
    return self._resolve_common(key, name, async_mode=False)

async def resolve_async(self, key: type[T] | str, name: str | None = None) -> T:
    """Resolves and returns a service instance asynchronously.

    This method should be used when the caller is in an async context. It
    can resolve both synchronous and asynchronous factories.

    Args:
        key (type[T] | str): The service key to resolve.
        name (str | None): An optional name for the registration.

    Returns:
        T: The resolved service instance.

    Raises:
        KeyError: If the service is not registered.
        BijuxError: If the factory fails, returns None, or a circular
            dependency is detected.
    """
    result = self._resolve_common(key, name, async_mode=True)
    if asyncio.iscoroutine(result):
        return await cast(Awaitable[T], result)
    else:
        return cast(T, result)

def services(self) -> Sequence[tuple[type[Any] | str, str | None]]:
    """Returns a list of all resolved and cached service keys.

    Returns:
        Sequence[tuple[type[Any] | str, str | None]]: A sequence of
            (key, name) tuples for all currently resolved services.
    """
    with self._lock:
        return list(self._services.keys())

async def shutdown(self) -> None:
    """Shuts down all resolved services that have a cleanup method.

    Iterates through all cached services and calls a `shutdown()` or
    `close()` method if one exists, handling both sync and async methods.
    """
    services = []
    with self._lock:
        services = list(self._services.items())
        obs_ref = self._obs
        self._services.clear()
        self._store.clear()
        self._obs = None
    for key, instance in services:
        try:
            shutdown_func = getattr(instance, "shutdown", None)
            if shutdown_func and callable(shutdown_func):
                is_async_shutdown = asyncio.iscoroutinefunction(shutdown_func)
                if is_async_shutdown:
                    await asyncio.wait_for(shutdown_func(), timeout=5.0)
                else:
                    shutdown_func()
                self._log(
                    logging.DEBUG,
                    "Shutting down service",
                    extra={"service_name": _key_name(key[0]), "svc_alias": key[1]},
                )
            elif isinstance(instance, ObservabilityProtocol) and not isinstance(
                instance, type
            ):
                instance.close()
                self._log(
                    logging.DEBUG,
                    "Closing observability service",
                    extra={"service_name": _key_name(key[0]), "svc_alias": key[1]},
                )
        except (RuntimeError, TypeError, TimeoutError) as exc:
            self._log(
                logging.ERROR,
                f"Shutdown failed: {exc}",
                extra={"service_name": _key_name(key[0]), "svc_alias": key[1]},
            )
    if obs_ref and hasattr(obs_ref, "close"):
        with suppress(Exception):
            obs_ref.close()
    self._log(logging.INFO, "DIContainer shutdown", extra={})

def unregister(self, key: type[Any] | str, name: str | None = None) -> bool:
    """Unregisters a service factory and removes any cached instance.

    Args:
        key (type[Any] | str): The service key to unregister.
        name (str | None): An optional name for the registration.

    Returns:
        bool: True if a service was found and unregistered, otherwise False.
    """
    with self._lock:
        store_key = (key, name)
        removed = self._store.pop(store_key, None) is not None
        if store_key in self._services and isinstance(
            self._services[store_key], ObservabilityProtocol
        ):
            self._obs = None
        self._services.pop(store_key, None)
        if removed:
            self._log(
                logging.INFO,
                "Unregistered service",
                extra={"service_name": _key_name(key), "svc_alias": name},
            )
        return removed