API

doppy-di: minimal dependency injection container.

AsyncConfiguration

Bases: Configuration

Async variant of :class:Configuration resolved via aget().

Rules are marked async so they resolve through the async path; a sync get() on one raises the same error as any other async rule.

Source code in src/doppy_di/configuration.py

class AsyncConfiguration(Configuration):
    """Async variant of :class:`Configuration` resolved via ``aget()``.

    Rules are marked async so they resolve through the async path; a sync
    ``get()`` on one raises the same error as any other async rule.
    """

    def _rule(self, key: str, make: Any, *, is_async: bool = False) -> Rule:
        async def async_make() -> Any:
            return make()

        return Rule(key, async_make, "transient", ())

AsyncDependencyInSyncContextError

Bases: Exception

Raised when an async dependency is resolved via sync get().

Examples:

>>> raise AsyncDependencyInSyncContextError("a")
Traceback (most recent call last):
...
AsyncDependencyInSyncContextError: Async dependency 'a' cannot be resolved in sync context

Source code in src/doppy_di/container.py

class AsyncDependencyInSyncContextError(Exception):
    """Raised when an async dependency is resolved via sync ``get()``.

    Examples:
        >>> raise AsyncDependencyInSyncContextError("a")
        Traceback (most recent call last):
        ...
        AsyncDependencyInSyncContextError: Async dependency 'a' cannot be resolved in sync context
    """

    def __init__(self, key: Key) -> None:
        self.key = key
        super().__init__(f"Async dependency {key!r} cannot be resolved in sync context")

AsyncScope

Bases: Scope

Async scope-local cache with async yield provider support.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> c = builder.build()
>>> async def main():
...     async with c.ascope("s") as s:
...         return await s.get("x")
>>> import asyncio
>>> asyncio.run(main())
1

Source code in src/doppy_di/container.py

class AsyncScope(Scope):
    """Async scope-local cache with async yield provider support.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> c = builder.build()
        >>> async def main():
        ...     async with c.ascope("s") as s:
        ...         return await s.get("x")
        >>> import asyncio
        >>> asyncio.run(main())
        1
    """

    async def get(self, key: Key) -> Any:
        """Resolve key from scope cache or underlying container."""
        if key in self.cache:
            self.container._trace(key, 0.0, True, self.name)
            return self.cache[key]
        try:
            rule = self.container.config.ruleset.find(key)
        except ServiceNotFoundError:
            from .providers import implicit_collection_rule

            if implicit_collection_rule(key, self.container.config.ruleset) is None:
                raise
            obj = await self.container.aget(key, _scope_name=self.name)
            self.cache[key] = obj
            return obj
        if rule.async_yield_provider:
            started = self.container._tracer is not None
            start = time.perf_counter() if started else 0.0
            stack = AsyncExitStack()
            try:
                obj = await stack.enter_async_context(asynccontextmanager(rule.make)())
            except RuntimeError as exc:
                if "didn't yield" in str(exc):
                    raise YieldNotCalledError(key) from None
                raise
            if started:
                self.container._trace(key, time.perf_counter() - start, False, self.name)
            self._async_exit_stack.append((key, stack))
            self.cache[key] = obj
            return obj
        if rule.yield_provider:
            raise TypeError(f"Sync yield provider {key!r} cannot be resolved in async scope")
        if rule.is_async:
            obj = await self.container.aget(key, _scope_name=self.name)
        else:
            obj = self.container.get(key, _scope_name=self.name)
        self.cache[key] = obj
        return obj

    async def __aenter__(self) -> AsyncScope:
        self._depth += 1
        _previous = _ACTIVE_REQUEST_RESOLVER.get()
        _ACTIVE_REQUEST_RESOLVER.set(self)
        self._resolver_previous = _previous
        return self

    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        _ACTIVE_REQUEST_RESOLVER.set(self._resolver_previous)
        self._depth -= 1
        if self._depth == 0:
            self.cache.clear()
            self.request_context.clear()
            errors: List[Tuple[Key, Exception]] = []
            for key, stack in self._async_exit_stack:
                try:
                    await stack.aclose()
                except Exception as exc:
                    if self.container.config.finalization_errors:
                        errors.append((key, exc))
                    else:
                        logger.exception("Error finalizing yield provider %r", key)
            self._async_exit_stack.clear()
            if errors:
                raise ResourceFinalizationError(errors)

get(key) async

Resolve key from scope cache or underlying container.

Source code in src/doppy_di/container.py

async def get(self, key: Key) -> Any:
    """Resolve key from scope cache or underlying container."""
    if key in self.cache:
        self.container._trace(key, 0.0, True, self.name)
        return self.cache[key]
    try:
        rule = self.container.config.ruleset.find(key)
    except ServiceNotFoundError:
        from .providers import implicit_collection_rule

        if implicit_collection_rule(key, self.container.config.ruleset) is None:
            raise
        obj = await self.container.aget(key, _scope_name=self.name)
        self.cache[key] = obj
        return obj
    if rule.async_yield_provider:
        started = self.container._tracer is not None
        start = time.perf_counter() if started else 0.0
        stack = AsyncExitStack()
        try:
            obj = await stack.enter_async_context(asynccontextmanager(rule.make)())
        except RuntimeError as exc:
            if "didn't yield" in str(exc):
                raise YieldNotCalledError(key) from None
            raise
        if started:
            self.container._trace(key, time.perf_counter() - start, False, self.name)
        self._async_exit_stack.append((key, stack))
        self.cache[key] = obj
        return obj
    if rule.yield_provider:
        raise TypeError(f"Sync yield provider {key!r} cannot be resolved in async scope")
    if rule.is_async:
        obj = await self.container.aget(key, _scope_name=self.name)
    else:
        obj = self.container.get(key, _scope_name=self.name)
    self.cache[key] = obj
    return obj

ChildrenFirstPolicy dataclass

Resolve nested children before the parent.

Examples:

>>> policy = ChildrenFirstPolicy(nested={"service": ["repo"]})
>>> isinstance(policy, ChildrenFirstPolicy)
True
>>> policy.nested
{'service': ['repo']}

Source code in src/doppy_di/devkit/policy.py

@dataclass(frozen=True)
class ChildrenFirstPolicy:
    """Resolve nested children before the parent.

    Examples:
        >>> policy = ChildrenFirstPolicy(nested={"service": ["repo"]})
        >>> isinstance(policy, ChildrenFirstPolicy)
        True
        >>> policy.nested
        {'service': ['repo']}
    """

    nested: Dict[Key, List[str]]

    def __init__(self, nested: Optional[Dict[Key, List[str]]] = None) -> None:
        object.__setattr__(self, "nested", dict(nested or {}))

    def before_resolve(self, key: Key, ruleset: RuleSetProtocol, ctx: ResolveContext) -> None:
        for child_name in self.nested.get(key, []):
            child_key = (key, child_name)
            ctx.get(child_key)

    def after_resolve(
        self, key: Key, obj: Any, ruleset: RuleSetProtocol, ctx: ResolveContext
    ) -> None:
        return None

CompilePolicy

Bases: Enum

Strategy for resolving rules after a plan is compiled.

ALLOW_OVERRIDE (default): the compiled plan delegates to the live container, so later override() calls are honoured. STRICT: once compile() is called, any override() call on the container raises RuntimeError because the plan was snapshot immutable.

Source code in src/doppy_di/container.py

class CompilePolicy(Enum):
    """Strategy for resolving rules after a plan is compiled.

    ALLOW_OVERRIDE (default): the compiled plan delegates to the live container,
        so later ``override()`` calls are honoured.
    STRICT: once ``compile()`` is called, any ``override()`` call on the
        container raises ``RuntimeError`` because the plan was snapshot immutable.
    """

    ALLOW_OVERRIDE = "allow_override"
    STRICT = "strict"

CompositeRuleSet

Rule storage that layers local rules over a parent rule set.

Reads delegate to the parent when a key is not overridden locally, so rules added to the parent after the child was created stay visible. Writes go only to the local layer, leaving the parent untouched.

The merged view is cached and invalidated when the parent version changes, so repeated reads are cheap.

Examples:

>>> parent = RuleSet()
>>> parent.add("db", Rule("db", lambda: "base-db"))
>>> child = CompositeRuleSet(parent)
>>> child.add("extra", Rule("extra", lambda: 1))
>>> child.find("db").key
'db'
>>> child.find("extra").key
'extra'

Source code in src/doppy_di/container.py

class CompositeRuleSet:
    """Rule storage that layers local rules over a parent rule set.

    Reads delegate to the parent when a key is not overridden locally, so
    rules added to the parent after the child was created stay visible.
    Writes go only to the local layer, leaving the parent untouched.

    The merged view is cached and invalidated when the parent version
    changes, so repeated reads are cheap.

    Examples:
        >>> parent = RuleSet()
        >>> parent.add("db", Rule("db", lambda: "base-db"))
        >>> child = CompositeRuleSet(parent)
        >>> child.add("extra", Rule("extra", lambda: 1))
        >>> child.find("db").key
        'db'
        >>> child.find("extra").key
        'extra'
    """

    __slots__ = (
        "defer_cycle_check",
        "graph_cache",
        "map_cache",
        "own_graph",
        "own_map",
        "own_version",
        "parent",
        "parent_version_seen",
    )

    def __init__(
        self,
        parent: RuleSetProtocol,
        defer_cycle_check: bool = False,
    ) -> None:
        """Initialize a composite rule set over ``parent``."""
        self.parent = parent
        self.own_map: Dict[Key, Rule] = {}
        self.own_graph: Dict[Key, Tuple[Key, ...]] = {}
        self.defer_cycle_check = defer_cycle_check
        self.own_version = 0
        self.parent_version_seen: Optional[int] = None
        self.map_cache: Optional[Dict[Key, Rule]] = None
        self.graph_cache: Optional[Dict[Key, Tuple[Key, ...]]] = None

    @property
    def map(self) -> Dict[Key, Rule]:
        """Return the merged rule map (own rules win)."""
        if self.map_cache is None or self.parent_version_seen != self.parent.version:
            merged = dict(self.parent.map)
            merged.update(self.own_map)
            self.map_cache = merged
            self.parent_version_seen = self.parent.version
        return self.map_cache

    @property
    def graph(self) -> Dict[Key, Tuple[Key, ...]]:
        """Return the merged dependency graph (own edges win)."""
        if self.graph_cache is None or self.parent_version_seen != self.parent.version:
            merged = dict(self.parent.graph)
            merged.update(self.own_graph)
            self.graph_cache = merged
            self.parent_version_seen = self.parent.version
        return self.graph_cache

    @property
    def version(self) -> Tuple[int, int]:
        """Return ``(parent_version, own_version)`` for cache invalidation."""
        return (self.parent.version, self.own_version)

    def add(self, key: Key, rule: Rule) -> None:
        """Add a rule to the local layer and validate graph cycles."""
        old_map = dict(self.own_map)
        old_graph = dict(self.own_graph)
        self.own_map[key] = rule
        self.own_graph[key] = tuple(rule.deps)
        self.map_cache = None
        self.graph_cache = None
        if not self.defer_cycle_check:
            try:
                self._check_cycle(key)
            except CycleError:
                self.own_map = old_map
                self.own_graph = old_graph
                self.map_cache = None
                self.graph_cache = None
                raise
        self.own_version += 1

    def find(self, key: Key) -> Rule:
        """Return a rule by key, falling back to the parent."""
        try:
            return self.map[key]
        except KeyError:
            raise ServiceNotFoundError(key) from None

    def has(self, key: Key) -> bool:
        """Check whether a key is registered locally or in the parent."""
        return key in self.map

    def deps_of(self, key: Key) -> Tuple[Key, ...]:
        """Return direct dependencies for a key."""
        return self.graph.get(key, ())

    def keys(self) -> Tuple[Key, ...]:
        """Return all registered keys (parent and local)."""
        return tuple(self.map.keys())

    def _check_cycle(self, start: Key) -> None:
        """Check graph cycles from the given start node."""
        stack: List[Key] = []
        on_stack: set[Key] = set()
        visited: set[Key] = set()

        def dfs(node: Key) -> None:
            if node in on_stack:
                raise DependencyCycleError([*stack, node])
            if node in visited:
                return
            visited.add(node)
            on_stack.add(node)
            stack.append(node)
            for dep in self.graph.get(node, ()):
                if dep in self.map:
                    dfs(dep)
            stack.pop()
            on_stack.remove(node)

        dfs(start)

graph property

Return the merged dependency graph (own edges win).

map property

Return the merged rule map (own rules win).

version property

Return (parent_version, own_version) for cache invalidation.

__init__(parent, defer_cycle_check=False)

Initialize a composite rule set over parent.

Source code in src/doppy_di/container.py

def __init__(
    self,
    parent: RuleSetProtocol,
    defer_cycle_check: bool = False,
) -> None:
    """Initialize a composite rule set over ``parent``."""
    self.parent = parent
    self.own_map: Dict[Key, Rule] = {}
    self.own_graph: Dict[Key, Tuple[Key, ...]] = {}
    self.defer_cycle_check = defer_cycle_check
    self.own_version = 0
    self.parent_version_seen: Optional[int] = None
    self.map_cache: Optional[Dict[Key, Rule]] = None
    self.graph_cache: Optional[Dict[Key, Tuple[Key, ...]]] = None

add(key, rule)

Add a rule to the local layer and validate graph cycles.

Source code in src/doppy_di/container.py

def add(self, key: Key, rule: Rule) -> None:
    """Add a rule to the local layer and validate graph cycles."""
    old_map = dict(self.own_map)
    old_graph = dict(self.own_graph)
    self.own_map[key] = rule
    self.own_graph[key] = tuple(rule.deps)
    self.map_cache = None
    self.graph_cache = None
    if not self.defer_cycle_check:
        try:
            self._check_cycle(key)
        except CycleError:
            self.own_map = old_map
            self.own_graph = old_graph
            self.map_cache = None
            self.graph_cache = None
            raise
    self.own_version += 1

deps_of(key)

Return direct dependencies for a key.

Source code in src/doppy_di/container.py

def deps_of(self, key: Key) -> Tuple[Key, ...]:
    """Return direct dependencies for a key."""
    return self.graph.get(key, ())

find(key)

Return a rule by key, falling back to the parent.

Source code in src/doppy_di/container.py

def find(self, key: Key) -> Rule:
    """Return a rule by key, falling back to the parent."""
    try:
        return self.map[key]
    except KeyError:
        raise ServiceNotFoundError(key) from None

has(key)

Check whether a key is registered locally or in the parent.

Source code in src/doppy_di/container.py

def has(self, key: Key) -> bool:
    """Check whether a key is registered locally or in the parent."""
    return key in self.map

keys()

Return all registered keys (parent and local).

Source code in src/doppy_di/container.py

def keys(self) -> Tuple[Key, ...]:
    """Return all registered keys (parent and local)."""
    return tuple(self.map.keys())

Configuration

Bases: Provider

Declarative provider reading a configuration tree.

Sources are given as keyword arguments and merged in priority order: dictionary, ini_path, json_path, yaml_path, settings, env_prefix, then env (last wins).

The live flag controls env interpolation. With live=False (default) the config resolves once at assignment time and env values are cached, so later env changes are not observed until :meth:reload. With live=True environment variables are re-read on every get so dynamic changes are visible, at the cost of per-resolution closure reads.

Examples:

>>> import os
>>> from doppy_di import Container
>>> from doppy_di.providers import Configuration
>>> services = Container()
>>> services.config = Configuration(
...     dictionary={"db": {"host": "${DB_HOST}", "port": 5432}}
... )
>>> os.environ["DB_HOST"] = "localhost"
>>> services.get("config.db.host")
'localhost'
>>> services.get("config")
{'db': {'host': 'localhost', 'port': 5432}}

Source code in src/doppy_di/configuration.py

class Configuration(Provider):
    """Declarative provider reading a configuration tree.

    Sources are given as keyword arguments and merged in priority order:
    ``dictionary``, ``ini_path``, ``json_path``, ``yaml_path``, ``settings``,
    ``env_prefix``, then ``env`` (last wins).

    The ``live`` flag controls env interpolation. With ``live=False`` (default)
    the config resolves once at assignment time and env values are cached, so
    later env changes are not observed until :meth:`reload`. With
    ``live=True`` environment variables are re-read on every ``get`` so dynamic
    changes are visible, at the cost of per-resolution closure reads.

    Examples:
        >>> import os
        >>> from doppy_di import Container
        >>> from doppy_di.providers import Configuration
        >>> services = Container()
        >>> services.config = Configuration(
        ...     dictionary={"db": {"host": "${DB_HOST}", "port": 5432}}
        ... )
        >>> os.environ["DB_HOST"] = "localhost"
        >>> services.get("config.db.host")
        'localhost'
        >>> services.get("config")
        {'db': {'host': 'localhost', 'port': 5432}}
    """

    def __init__(
        self,
        *,
        dictionary: Optional[Mapping[str, Any]] = None,
        ini_path: Optional[str] = None,
        json_path: Optional[str] = None,
        yaml_path: Optional[str] = None,
        settings: Optional[Any] = None,
        env: Optional[Mapping[str, str]] = None,
        env_prefix: Optional[str] = None,
        live: bool = False,
    ) -> None:
        self.dictionary = dictionary
        self.ini_path = ini_path
        self.json_path = json_path
        self.yaml_path = yaml_path
        self.settings = settings
        self.env = env
        self.env_prefix = env_prefix
        self.live = live

        self._data: Dict[str, Any] = {}
        self._resolved: Dict[str, Any] = {}
        self._paths: List[Tuple[str, ...]] = []

    # -- source loading -------------------------------------------------

    def _load_json(self, path: str) -> Dict[str, Any]:
        with open(path, encoding="utf-8") as handle:
            data = json.load(handle)
        return data if isinstance(data, dict) else {}

    def _load_yaml(self, path: str) -> Dict[str, Any]:
        try:
            import yaml  # type: ignore[import-untyped]
        except ImportError as exc:  # pragma: no cover - depends on env
            raise ConfigurationError(
                "YAML support requires 'PyYAML'; install with 'pip install doppy-di[config]'"
            ) from exc
        with open(path, encoding="utf-8") as handle:
            data = yaml.safe_load(handle)
        return data if isinstance(data, dict) else {}

    def _load_ini(self, path: str) -> Dict[str, Any]:
        parser = configparser.ConfigParser()
        parser.read(path, encoding="utf-8")
        return {section: dict(parser.items(section)) for section in parser.sections()}

    def _load_settings(self, obj: Any) -> Dict[str, Any]:
        dump = getattr(obj, "model_dump", None)
        if callable(dump):
            return cast("Dict[str, Any]", dump())
        legacy = getattr(obj, "dict", None)
        if callable(legacy):
            return cast("Dict[str, Any]", legacy())
        return {key: value for key, value in vars(obj).items() if not key.startswith("_")}

    def _load_env_prefix(self, prefix: str) -> Dict[str, Any]:
        root: Dict[str, Any] = {}
        for key, value in os.environ.items():
            if not key.startswith(prefix):
                continue
            rest = key[len(prefix) :]
            if not rest:
                continue
            parts = rest.lstrip("_").split("__")
            node = root
            for part in parts[:-1]:
                node = node.setdefault(part.lower(), {})
            node[parts[-1].lower()] = value
        return root

    def _load_sources(self) -> Dict[str, Any]:
        sources: List[Dict[str, Any]] = []
        if self.dictionary is not None:
            sources.append(dict(self.dictionary))
        if self.ini_path is not None:
            sources.append(self._load_ini(self.ini_path))
        if self.json_path is not None:
            sources.append(self._load_json(self.json_path))
        if self.yaml_path is not None:
            sources.append(self._load_yaml(self.yaml_path))
        if self.settings is not None:
            sources.append(self._load_settings(self.settings))
        if self.env_prefix is not None:
            sources.append(self._load_env_prefix(self.env_prefix))
        if self.env is not None:
            sources.append(dict(self.env))
        merged: Dict[str, Any] = {}
        for source in sources:
            merged = _merge(merged, source)
        return merged

    # -- tree helpers ----------------------------------------------------

    def _flatten(self, node: Any, prefix: Tuple[str, ...] = ()) -> List[Tuple[str, ...]]:
        if not isinstance(node, dict):
            return []
        paths: List[Tuple[str, ...]] = []
        for key, value in node.items():
            path = (*prefix, str(key))
            paths.append(path)
            paths.extend(self._flatten(value, path))
        return paths

    def _lookup(self, tree: Dict[str, Any], path: Tuple[str, ...]) -> Any:
        node: Any = tree
        for part in path:
            node = node[part]
        return node

    def _resolve_full(self) -> Dict[str, Any]:
        if self.live:
            return cast("Dict[str, Any]", _interpolate(self._data, os.environ))
        return self._resolved

    def _resolve_path(self, path: Tuple[str, ...]) -> Any:
        if self.live:
            root = _interpolate(self._data, os.environ)
            return self._lookup(root, path)
        return self._lookup(self._resolved, path)

    def _read(self) -> None:
        self._data = self._load_sources()
        self._paths = self._flatten(self._data)
        self._resolved = _interpolate(self._data, os.environ)

    def _child_keys(self, name: str) -> Set[str]:
        return {".".join((name, *path)) for path in self._paths}

    # -- Provider interface ---------------------------------------------

    def pre_validate_registration(self, ruleset: RuleSetProtocol, name: str) -> None:
        """Raise :class:`DuplicateKeyError` if a namespaced child key is taken."""
        self._read()
        reserved = self._child_keys(name)
        if ruleset.has(name):
            raise DuplicateKeyError(name)
        for child in reserved:
            if ruleset.has(child):
                raise DuplicateKeyError(child)

    def _rule(self, key: str, make: Any, *, is_async: bool = False) -> Rule:
        return Rule(key, make, "transient", (), is_async=is_async)

    def to_rules(self, name: str) -> List[Rule]:
        """Register a parent rule plus one rule per dotted config path."""
        self.key = name
        self._read()

        def make_parent() -> Any:
            return self._resolve_full()

        rules: List[Rule] = [self._rule(name, make_parent)]

        def make_leaf(path: Tuple[str, ...]) -> Any:
            return lambda: self._resolve_path(path)

        for path in self._paths:
            child_key = ".".join((name, *path))
            rules.append(self._rule(child_key, make_leaf(path)))
        return rules

    def reload(self) -> None:
        """Re-read all sources and rebuild the resolved snapshot."""
        self._read()

pre_validate_registration(ruleset, name)

Raise :class:DuplicateKeyError if a namespaced child key is taken.

Source code in src/doppy_di/configuration.py

def pre_validate_registration(self, ruleset: RuleSetProtocol, name: str) -> None:
    """Raise :class:`DuplicateKeyError` if a namespaced child key is taken."""
    self._read()
    reserved = self._child_keys(name)
    if ruleset.has(name):
        raise DuplicateKeyError(name)
    for child in reserved:
        if ruleset.has(child):
            raise DuplicateKeyError(child)

reload()

Re-read all sources and rebuild the resolved snapshot.

Source code in src/doppy_di/configuration.py

def reload(self) -> None:
    """Re-read all sources and rebuild the resolved snapshot."""
    self._read()

to_rules(name)

Register a parent rule plus one rule per dotted config path.

Source code in src/doppy_di/configuration.py

def to_rules(self, name: str) -> List[Rule]:
    """Register a parent rule plus one rule per dotted config path."""
    self.key = name
    self._read()

    def make_parent() -> Any:
        return self._resolve_full()

    rules: List[Rule] = [self._rule(name, make_parent)]

    def make_leaf(path: Tuple[str, ...]) -> Any:
        return lambda: self._resolve_path(path)

    for path in self._paths:
        child_key = ".".join((name, *path))
        rules.append(self._rule(child_key, make_leaf(path)))
    return rules

ConfigurationError

Bases: Exception

Raised when a configuration source cannot be loaded.

Source code in src/doppy_di/configuration.py

class ConfigurationError(Exception):
    """Raised when a configuration source cannot be loaded."""

Container

Runtime container with singleton cache.

Thread-safe singleton resolution with double-checked locking.

Examples:

>>> builder = ContainerBuilder()
>>> builder.service("answer", lambda: 42, lifetime="singleton")
>>> container = builder.build()
>>> container.get("answer")
42

Source code in src/doppy_di/container.py

class Container:
    """Runtime container with singleton cache.

    Thread-safe singleton resolution with double-checked locking.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.service("answer", lambda: 42, lifetime="singleton")
        >>> container = builder.build()
        >>> container.get("answer")
        42
    """

    __slots__ = (
        "_compiled_plan",
        "_override_layers",
        "_policy",
        "_policy_depth",
        "_providers",
        "_tracer",
        "_visualize_cache",
        "_visualize_version",
        "config",
        "lock",
        "scope_policy",
        "scopes",
        "single",
    )

    def __init__(self, config: Optional[ContainerConfig] = None) -> None:
        if config is None:
            config = ContainerConfig(RuleSet())
        self.config = config
        self.single: Dict[Key, Any] = {}
        self.scopes: Dict[str, Scope] = {}
        self.scope_policy: ScopePolicy = config.scope_policy
        self.lock = threading.RLock()
        self._visualize_cache: Dict[str, Any] = {}
        self._visualize_version = -1
        self._providers: Dict[str, Any] = {}
        self._override_layers: List[OverrideLayer] = []
        self._compiled_plan: Optional["ExecutionPlan"] = None
        self._tracer: Optional[TracerFn] = None
        self._policy: Optional["ResolutionPolicy"] = config.policy
        self._policy_depth = 0

    def _resolve_with_policy(
        self,
        lookup: Key,
        policy: "ResolutionPolicy",
        _scope_name: Optional[str],
    ) -> Any:
        """Resolve ``lookup`` honouring the policy's key ordering."""
        from .resolution import ParallelPolicy

        order = list(policy.order(self.config.ruleset.map, lookup))
        self._policy_depth += 1
        cache: Dict[Key, Any] = {}
        try:
            if isinstance(policy, ParallelPolicy):
                for level in self._independent_levels(order):
                    for key in level:
                        if key not in self.single and key not in cache:
                            cache[key] = self.get(key, _scope_name=_scope_name)
            else:
                for key in order:
                    if key not in self.single and key not in cache:
                        cache[key] = self.get(key, _scope_name=_scope_name)
            if lookup in cache:
                return cache[lookup]
            return self.get(lookup, _scope_name=_scope_name)
        finally:
            self._policy_depth -= 1

    async def _resolve_with_policy_async(
        self,
        lookup: Key,
        policy: "ResolutionPolicy",
        _stacks: Optional[List[AsyncExitStack]],
        _scope_name: Optional[str],
        _path: Optional[List[Key]],
    ) -> Any:
        """Resolve ``lookup`` asynchronously honouring the policy ordering."""
        from .resolution import ParallelPolicy

        order = list(policy.order(self.config.ruleset.map, lookup))
        self._policy_depth += 1
        cache: Dict[Key, Any] = {}
        try:
            if isinstance(policy, ParallelPolicy):
                for level in self._independent_levels(order):
                    for key in level:
                        if key not in self.single and key not in cache:
                            cache[key] = await self.aget(
                                key,
                                _stacks=_stacks,
                                _scope_name=_scope_name,
                                _path=_path,
                            )
            else:
                for key in order:
                    if key not in self.single and key not in cache:
                        cache[key] = await self.aget(
                            key,
                            _stacks=_stacks,
                            _scope_name=_scope_name,
                            _path=_path,
                        )
            if lookup in cache:
                return cache[lookup]
            return await self.aget(
                lookup,
                _stacks=_stacks,
                _scope_name=_scope_name,
                _path=_path,
            )
        finally:
            self._policy_depth -= 1

    def __setattr__(self, name: str, value: Any) -> None:
        if hasattr(value, "to_rules"):
            pre_validate = getattr(value, "pre_validate_registration", None)
            if pre_validate is not None:
                pre_validate(self.config.ruleset, name)
            for rule in value.to_rules(name):
                self.config.ruleset.add(rule.key, rule)
            self._providers[name] = value
            return
        object.__setattr__(self, name, value)

    def __getattr__(self, name: str) -> Any:
        from .providers import UnboundProvider

        try:
            providers = object.__getattribute__(self, "_providers")
        except AttributeError:
            return UnboundProvider(name)
        if name in providers:
            return providers[name]
        return UnboundProvider(name)

    def _enter_path(self, key: Key, path: Optional[List[Key]] = None) -> List[Key]:
        current = path
        if current is None:
            current = _RESOLUTION_PATH.get()
        if current is None:
            current = []
            _RESOLUTION_PATH.set(current)
        current.append(key)
        return current

    def set_tracer(self, tracer_fn: Optional[TracerFn]) -> None:
        """Set a tracer callback or disable tracing with ``None``.

        The callback receives ``(key, duration, cache_hit, scope)`` after
        every successful resolution. When no tracer is set there is no
        timing and no dispatch, so overhead is zero.

        Args:
            tracer_fn: Callback receiving trace events, or ``None`` to
                disable tracing.

        Examples:
            >>> events = []
            >>> builder = ContainerBuilder()
            >>> builder.value("a", 1)
            >>> container = builder.build()
            >>> container.set_tracer(lambda *args: events.append(args))
            >>> container.get("a")
            1
            >>> len(events)
            1
            >>> container.set_tracer(None)
            >>> container.get("a")
            1
            >>> len(events)
            1
        """
        self._tracer = tracer_fn

    def _trace(self, key: Key, duration: float, cache_hit: bool, scope: Optional[str]) -> None:
        """Dispatch a trace event when a tracer is configured."""
        tracer = self._tracer
        if tracer is not None:
            tracer(key, duration, cache_hit, scope)

    def get(
        self,
        key: Key,
        qualifier: Optional[str] = None,
        _scope_name: Optional[str] = None,
        policy: Optional["ResolutionPolicy"] = None,
    ) -> Any:
        """Resolve a service by key.

        Returns the cached singleton if already resolved, otherwise resolves
        the rule from the config, applies the scope policy, and stores the
        result for singleton lifetimes.

        Double-checked locking provides thread safety.

        Args:
            key: Service key.
            qualifier: Optional named qualifier. When given, resolves the
                rule registered as ``(key, qualifier)``.
            policy: Optional per-call resolution policy. Overrides the
                container-wide policy for this call only.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.service("answer", lambda: 42)
            >>> container = builder.build()
            >>> container.get("answer")
            42
        """
        lookup = (key, qualifier) if qualifier is not None else key
        active = policy if policy is not None else self._policy
        if active is not None and self._policy_depth == 0:
            return self._resolve_with_policy(lookup, active, _scope_name)
        started = self._tracer is not None
        start = time.perf_counter() if started else 0.0
        if self._override_layers:
            overridden = self._resolve_override(lookup)
            if overridden is not _unset:
                if started:
                    self._trace(lookup, time.perf_counter() - start, False, _scope_name)
                return overridden
        if lookup in self.single:
            if started:
                self._trace(lookup, time.perf_counter() - start, True, _scope_name)
            return self.single[lookup]

        path = self._enter_path(lookup)
        try:
            if len(path) > 1 and lookup in path[:-1]:
                idx = path.index(lookup)
                raise DependencyCycleError(path[idx:])

            with self.lock:
                if lookup in self.single:
                    if started:
                        self._trace(lookup, time.perf_counter() - start, True, _scope_name)
                    return self.single[lookup]

                try:
                    rule = self.config.ruleset.find(lookup)
                except ServiceNotFoundError:
                    from .providers import implicit_collection_rule

                    collection = implicit_collection_rule(lookup, self.config.ruleset)
                    if collection is not None:
                        rule = collection
                    elif self._is_injectable_key(lookup):
                        from .auto_wiring import _rule_for

                        self.config.ruleset.add(lookup, _rule_for(lookup))
                        rule = self.config.ruleset.find(lookup)
                    elif qualifier is not None:
                        raise UnregisteredDependencyError(key, qualifier) from None
                    elif len(path) > 1:
                        src = self.config.ruleset.map.get(
                            lookup, Rule(lookup, lambda: None)
                        ).registration_source
                        raise MissingDependencyError(
                            lookup,
                            path.copy(),
                            scope=_scope_name,
                            registration_source=src,
                        ) from None
                    else:
                        raise
                if rule.async_yield_provider:
                    raise TypeError(f"Async yield provider {lookup!r} requires async scope")
                if rule.is_async:
                    raise AsyncDependencyInSyncContextError(lookup)
                ctx = ResolveContext(self)
                try:
                    args = [ctx.get(dep, _scope_name=_scope_name) for dep in rule.deps]
                except ServiceNotFoundError as exc:
                    if self._is_injectable_key(lookup):
                        from .auto_wiring import UnresolvableDependencyError

                        raise UnresolvableDependencyError(lookup, exc.key) from None
                    if isinstance(exc, MissingDependencyError):
                        if exc.registration_source is not None:
                            raise
                        src = self.config.ruleset.map.get(
                            lookup, Rule(lookup, lambda: None)
                        ).registration_source
                        if src is None:
                            raise
                        raise MissingDependencyError(
                            exc.key,
                            exc.resolution_path,
                            scope=exc.scope or _scope_name,
                            registration_source=src,
                        ) from None
                    if len(path) > 1:
                        src = self.config.ruleset.map.get(
                            lookup, Rule(lookup, lambda: None)
                        ).registration_source
                        raise MissingDependencyError(
                            exc.key,
                            path.copy(),
                            scope=_scope_name,
                            registration_source=src,
                        ) from None
                    raise
                try:
                    obj = rule.make(*args)
                except Exception as exc:
                    if self.config.wrap_factory_errors:
                        raise FactoryExecutionError(
                            lookup,
                            exc,
                            path.copy(),
                        ) from exc
                    raise
                if inspect.isawaitable(obj):
                    raise SyncFactoryReturningAwaitableError(lookup)

                if rule.lifetime == "singleton":
                    self.single[lookup] = obj
                self._cache_nested_aliases(lookup, obj)

                if started:
                    self._trace(lookup, time.perf_counter() - start, False, _scope_name)

                return obj
        finally:
            path.pop()

    @staticmethod
    def _is_injectable_key(key: Key) -> bool:
        """Return True when key is an injectable type or qualified type."""
        if isinstance(key, type):
            return bool(getattr(key, "__doppy_injectable__", False))
        if isinstance(key, tuple) and len(key) == 2 and isinstance(key[0], type):
            return bool(getattr(key[0], "__doppy_injectable__", False))
        return False

    def _cache_nested_aliases(self, key: Key, obj: Any) -> None:
        for alias, rule in self.config.ruleset.map.items():
            if not rule.nested:
                continue
            if not isinstance(alias, tuple) or len(alias) != 2 or alias[0] != key:
                continue
            child = alias[1]
            if isinstance(child, str) and hasattr(obj, child):
                self.single[alias] = getattr(obj, child)

    async def aget(
        self,
        key: Key,
        qualifier: Optional[str] = None,
        _stacks: Optional[List[AsyncExitStack]] = None,
        _scope_name: Optional[str] = None,
        _path: Optional[List[Key]] = None,
        policy: Optional["ResolutionPolicy"] = None,
    ) -> Any:
        """Resolve a service by key asynchronously.

        Resolves dependencies concurrently with ``asyncio.gather`` and awaits
        async factories. Sync factories are called directly, so there is no
        overhead for sync dependencies. Singleton results are cached.

        Async yield providers are supported; their resources are finalized
        when resolution is cancelled.

        Args:
            key: Service key.
            qualifier: Optional named qualifier. When given, the rule
                registered as ``(key, qualifier)`` is resolved.
            policy: Optional per-call resolution policy. Overrides the
                container-wide policy for this call only.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.service("answer", lambda: 42)
            >>> container = builder.build()
            >>> async def main():
            ...     return await container.aget("answer")
            >>> import asyncio
            >>> asyncio.run(main())
            42
        """
        lookup = (key, qualifier) if qualifier is not None else key
        active = policy if policy is not None else self._policy
        if active is not None and self._policy_depth == 0:
            return await self._resolve_with_policy_async(
                lookup,
                active,
                _stacks,
                _scope_name,
                _path,
            )
        started = self._tracer is not None
        start = time.perf_counter() if started else 0.0
        if self._override_layers:
            overridden = self._resolve_override(lookup)
            if overridden is not _unset:
                if inspect.isawaitable(overridden):
                    overridden = await overridden
                if started:
                    self._trace(lookup, time.perf_counter() - start, False, _scope_name)
                return overridden
        if lookup in self.single:
            if started:
                self._trace(lookup, time.perf_counter() - start, True, _scope_name)
            return self.single[lookup]

        path = self._enter_path(lookup, _path)
        try:
            if len(path) > 1 and lookup in path[:-1]:
                idx = path.index(lookup)
                raise DependencyCycleError(path[idx:])

            try:
                rule = self.config.ruleset.find(lookup)
            except ServiceNotFoundError:
                from .providers import implicit_collection_rule

                collection = implicit_collection_rule(lookup, self.config.ruleset)
                if collection is not None:
                    rule = collection
                elif self._is_injectable_key(lookup):
                    from .auto_wiring import _rule_for

                    self.config.ruleset.add(lookup, _rule_for(lookup))
                    rule = self.config.ruleset.find(lookup)
                elif qualifier is not None:
                    raise UnregisteredDependencyError(key, qualifier) from None
                elif len(path) > 1:
                    src = self.config.ruleset.map.get(
                        lookup, Rule(lookup, lambda: None)
                    ).registration_source
                    raise MissingDependencyError(
                        lookup,
                        path.copy(),
                        scope=_scope_name,
                        registration_source=src,
                    ) from None
                else:
                    raise
            stacks = _stacks if _stacks is not None else []
            try:
                if rule.async_yield_provider:
                    stack = AsyncExitStack()
                    stacks.append(stack)
                    try:
                        obj = await stack.enter_async_context(asynccontextmanager(rule.make)())
                    except RuntimeError as exc:
                        if "didn't yield" in str(exc):
                            raise YieldNotCalledError(lookup) from None
                        raise
                    if rule.lifetime == "singleton":
                        self.single[lookup] = obj
                    self._cache_nested_aliases(lookup, obj)
                    if started:
                        self._trace(lookup, time.perf_counter() - start, False, _scope_name)
                    return obj
                if rule.yield_provider:
                    raise TypeError(f"Sync yield provider {lookup!r} cannot be resolved via aget")
                levels = self._independent_levels(list(rule.deps))
                if rule.deps and not levels:
                    raise DependencyCycleError([lookup, *rule.deps])
                args_by_key: Dict[Key, Any] = {}
                for level in levels:
                    resolved = await asyncio.gather(
                        *(
                            self.aget(
                                dep,
                                _stacks=stacks,
                                _scope_name=_scope_name,
                                _path=path,
                            )
                            for dep in level
                        )
                    )
                    args_by_key.update(dict(zip(level, resolved)))
                args = [args_by_key[dep] for dep in rule.deps]
                try:
                    obj = rule.make(*args)
                except Exception as exc:
                    if self.config.wrap_factory_errors:
                        raise FactoryExecutionError(lookup, exc, path.copy()) from exc
                    raise
                if not rule.is_async and inspect.isawaitable(obj):
                    raise SyncFactoryReturningAwaitableError(lookup)
                if inspect.isawaitable(obj):
                    try:
                        obj = await obj
                    except Exception as exc:
                        if self.config.wrap_factory_errors:
                            raise FactoryExecutionError(lookup, exc, path.copy()) from exc
                        raise

                if rule.lifetime == "singleton":
                    self.single[lookup] = obj
                self._cache_nested_aliases(lookup, obj)

                if started:
                    self._trace(lookup, time.perf_counter() - start, False, _scope_name)

                return obj
            except asyncio.CancelledError:
                errors: List[Tuple[Key, Exception]] = []
                for stack in stacks:
                    try:
                        await stack.aclose()
                    except Exception as exc:
                        if self.config.finalization_errors:
                            errors.append((lookup, exc))
                        else:
                            logger.exception("Error finalizing yield provider %r", lookup)
                if errors:
                    raise ResourceFinalizationError(errors) from None
                raise ResolutionCancelledError(lookup) from None
        finally:
            path.pop()

    async def get_many(self, keys: List[Key], parallel: bool = False) -> List[Any]:
        """Resolve multiple services, optionally in parallel.

        Args:
            keys: Service keys to resolve.
            parallel: When True, resolve independent dependencies
                concurrently. Falls back to sequential resolution for small
                graphs (fewer than 5 nodes) where parallelism overhead
                outweighs the benefit.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("a", 1)
            >>> builder.value("b", 2)
            >>> container = builder.build()
            >>> async def main():
            ...     return await container.get_many(["a", "b"])
            >>> import asyncio
            >>> asyncio.run(main())
            [1, 2]
        """
        if not parallel:
            return [await self.aget(key) for key in keys]

        levels = self._independent_levels(keys)
        total = sum(len(level) for level in levels)
        if total < 5:
            return [await self.aget(key) for key in keys]

        results: Dict[Key, Any] = {}
        for level in levels:
            resolved = await asyncio.gather(*(self.aget(key) for key in level))
            results.update(dict(zip(level, resolved)))
        return [results[key] for key in keys]

    def _independent_levels(self, keys: List[Key]) -> List[List[Key]]:
        """Group keys and their transitive deps into dependency levels.

        Each level contains nodes whose dependencies all appear in earlier
        levels, so nodes within a level can be resolved concurrently.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("a", 1)
            >>> builder.service("b", lambda a: a + 1, deps=["a"])
            >>> container = builder.build()
            >>> container._independent_levels(["b"])
            [['a'], ['b']]
        """
        ruleset = self.config.ruleset
        needed: set[Key] = set()
        stack = list(keys)
        while stack:
            key = stack.pop()
            if key in needed:
                continue
            needed.add(key)
            stack.extend(ruleset.deps_of(key))

        indegree: Dict[Key, int] = dict.fromkeys(needed, 0)
        dependents: Dict[Key, List[Key]] = {key: [] for key in needed}
        for key in needed:
            for dep in ruleset.deps_of(key):
                if dep in needed:
                    indegree[key] += 1
                    dependents[dep].append(key)

        ready = [key for key in needed if indegree[key] == 0]
        levels: List[List[Key]] = []
        while ready:
            levels.append(ready)
            next_ready: List[Key] = []
            for key in ready:
                for dependent in dependents[key]:
                    indegree[dependent] -= 1
                    if indegree[dependent] == 0:
                        next_ready.append(dependent)
            ready = next_ready
        return levels

    def scan(
        self,
        *packages: Union[ModuleType, str],
        recursive: bool = True,
    ) -> None:
        """Register all injectable classes found in the given packages.

        Explicitly registered rules are never overridden.

        Examples:
            >>> builder = ContainerBuilder()
            >>> container = builder.build()
            >>> container.scan(__name__)
        """
        from .auto_wiring import scan_package

        for pkg in packages:
            scan_package(self, pkg, recursive)

    def has(self, key: Key, qualifier: Optional[str] = None) -> bool:
        """Return True if a rule for key is registered.

        Args:
            key: Service key.
            qualifier: Optional named qualifier.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.service("x", lambda: 1)
            >>> c = builder.build()
            >>> c.has("x")
            True
            >>> c.has("missing")
            False
        """
        lookup = (key, qualifier) if qualifier is not None else key
        return self.config.ruleset.has(lookup)

    def get_or_none(self, key: Key, qualifier: Optional[str] = None) -> Any:
        """Return resolved service or ``None`` if key not registered.

        Args:
            key: Service key.
            qualifier: Optional named qualifier.

        Examples:
            >>> builder = ContainerBuilder()
            >>> c = builder.build()
            >>> c.get_or_none("missing") is None
            True
        """
        try:
            return self.get(key, qualifier=qualifier)
        except (ServiceNotFoundError, UnregisteredDependencyError):
            return None

    def scope(self, name: str) -> Scope:
        """Return a named or unique scope according to the active policy.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("x", 1)
            >>> c = builder.build()
            >>> s = c.scope("req")
            >>> isinstance(s, Scope)
            True
        """
        if self.scope_policy == ScopePolicy.NAMED:
            if name in self.scopes:
                return self.scopes[name]
            scope = Scope(self, name)
            self.scopes[name] = scope
            return scope
        # UNIQUE: fresh Scope per call, stored under unique internal key
        internal = f"{name}#{uuid.uuid4().hex}"
        scope = Scope(self, name)
        self.scopes[internal] = scope
        return scope

    def ascope(self, name: str) -> AsyncScope:
        """Return a named or unique async scope.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("x", 1)
            >>> c = builder.build()
            >>> s = c.ascope("req")
            >>> isinstance(s, AsyncScope)
            True
        """
        if self.scope_policy == ScopePolicy.NAMED:
            if name in self.scopes:
                existing = self.scopes[name]
                if isinstance(existing, AsyncScope):
                    return existing
                raise TypeError(f"Scope {name!r} already exists as sync scope")
            scope = AsyncScope(self, name)
            self.scopes[name] = scope
            return scope
        # UNIQUE: fresh Scope per call, stored under unique internal key
        internal = f"{name}#{uuid.uuid4().hex}"
        scope = AsyncScope(self, name)
        self.scopes[internal] = scope
        return scope

    def _resolve_override(self, lookup: Key) -> Any:
        for layer in reversed(self._override_layers):
            if lookup in layer.values:
                return layer.resolve(lookup)
        return _unset

    def override(
        self,
        key: Union[Key, Dict[Key, Any]],
        value: Any = None,
        **overrides: Any,
    ) -> OverrideContext:
        """Create a temporary override context.

        Supports both a single ``key``/``value`` pair and a dictionary of
        overrides: ``container.override({"a": 1, "b": 2})``.

        Nested overrides stack LIFO: the last ``override()`` wins. On
        context exit the previous state is restored.

        Callable override values are treated as factories and invoked on
        every resolution.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("x", 1)
            >>> c = builder.build()
            >>> ctx = c.override("x", 2)
            >>> isinstance(ctx, OverrideContext)
            True

            >>> with c.override({"x": 3}):
            ...     c.get("x")
            3
            >>> c.get("x")
            1
        """
        if isinstance(key, dict):
            values: Dict[Key, Any] = dict(key)
        else:
            values = {key: value}
        for override_key, override_value in overrides.items():
            values[override_key] = override_value
        return OverrideContext(self, values)

    def value(self, key: Key, value: Any) -> Self:
        """Register a constant value on this container.

        Child containers inherit this rule; overriding it on a child does
        not touch the parent.

        Examples:
            >>> builder = ContainerBuilder()
            >>> container = builder.build()
            >>> container.value("env", "base")
            >>> container.get("env")
            'base'
        """

        def make_value() -> Any:
            return value

        self.config.ruleset.add(
            key,
            Rule(
                key=key,
                make=make_value,
                lifetime="singleton",
                deps=(),
            ),
        )
        return self

    def service(
        self,
        key: Key,
        make: Callable[..., Any],
        lifetime: Lifetime = "transient",
        deps: Optional[List[Key]] = None,
        qualifier: Optional[str] = None,
        scope: Optional[str] = None,
    ) -> Self:
        """Register a factory service on this container.

        Examples:
            >>> builder = ContainerBuilder()
            >>> container = builder.build()
            >>> container.service("a", lambda: 1)
            >>> container.get("a")
            1
        """
        lookup = (key, qualifier) if qualifier is not None else key
        rule = Rule(
            key=lookup,
            make=make,
            lifetime=lifetime,
            deps=tuple(deps or ()),
            scope=scope,
        )
        self.config.ruleset.add(lookup, rule)
        return self

    def child(self, name: Optional[str] = None) -> Container:
        """Return a new container layered over this one.

        The child inherits the parent rules and can add or override rules
        without mutating the parent. Rules added to the parent after the
        child was created stay visible to the child.

        Args:
            name: Optional profile name stored in ``config.profile``.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("db", "base-db")
            >>> parent = builder.build()
            >>> child = parent.child()
            >>> child.value("db", "child-db")
            >>> child.get("db")
            'child-db'
            >>> parent.get("db")
            'base-db'
        """
        composite = CompositeRuleSet(self.config.ruleset)
        child_container = Container(
            ContainerConfig(
                composite,
                scope_policy=self.config.scope_policy,
                track_sources=self.config.track_sources,
                wrap_factory_errors=self.config.wrap_factory_errors,
                finalization_errors=self.config.finalization_errors,
                profile=name,
            )
        )
        child_container._tracer = self._tracer
        child_container._policy = self._policy
        return child_container

    def with_profile(
        self,
        name: str,
        overrides: Optional[Dict[Key, Any]] = None,
    ) -> Container:
        """Return a derived child container with the given overrides applied.

        Each override is registered as a singleton value on the child. Keys
        must already exist in the effective rule set; unknown keys raise
        :class:`UnregisteredTypeError`.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("env", "base")
            >>> container = builder.build()
            >>> prod = container.with_profile("prod", {"env": "prod"})
            >>> container.get("env")
            'base'
            >>> prod.get("env")
            'prod'
            >>> prod.config.profile
            'prod'
        """
        derived = self.child(name)
        for key, item in (overrides or {}).items():
            if not derived.has(key):
                raise UnregisteredTypeError(key)
            derived.value(key, item)
        return derived

    def diff(self, other: Container) -> DiffReport:
        """Return rule differences between this container and ``other``.

        Added keys exist only in ``other``, removed keys exist only in this
        container, changed keys exist in both with different rule metadata
        (lifetime, deps, scope, resource flags).

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("a", 1)
            >>> base = builder.build()
            >>> modded = base.child()
            >>> modded.value("b", 2)
            >>> report = base.diff(modded)
            >>> "b" in report.added
            True
        """
        own = self.config.ruleset.map
        other_map = other.config.ruleset.map
        added: List[Key] = []
        removed: List[Key] = []
        changed: List[Key] = []
        for key in other_map:
            if key not in own:
                added.append(key)
            elif _rule_signature(own[key]) != _rule_signature(other_map[key]):
                changed.append(key)
        for key in own:
            if key not in other_map:
                removed.append(key)
        return DiffReport(
            added=tuple(added),
            removed=tuple(removed),
            changed=tuple(changed),
        )

    def export_config(self, format: str = "json") -> str:  # noqa: A002
        """Export the effective configuration as a JSON string.

        Includes the container profile and a rule table with lifetime, deps,
        scope, and resource flags. Non-string keys are rendered via ``repr``.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("a", 1)
            >>> container = builder.build()
            >>> out = container.export_config()
            >>> '"a"' in out
            True
        """
        if format != "json":
            raise ValueError(f"Unsupported export format: {format!r}")
        rules: Dict[str, Any] = {}
        for key, rule in self.config.ruleset.map.items():
            rules[repr(key)] = {
                "lifetime": rule.lifetime,
                "deps": [repr(dep) for dep in rule.deps],
                "scope": rule.scope,
                "yield": rule.yield_provider or rule.async_yield_provider,
                "nested": rule.nested,
            }
        payload = {
            "profile": self.config.profile,
            "rules": rules,
        }
        return json.dumps(payload, sort_keys=True, indent=2)

    def graph(self) -> DependencyGraph:
        """Return a DependencyGraph representation of this container."""
        from .graph import DependencyGraph

        return DependencyGraph(self.config.ruleset)

    def visualize(self, format: str = "mermaid") -> Any:  # noqa: A002
        """Return a textual representation of the dependency graph.

        Supported formats:
            - ``"mermaid"`` — mermaid ``graph TD`` for embedding in Markdown.
            - ``"graphviz"`` — Graphviz ``digraph`` for PNG/SVG generation.
            - ``"json"`` — structured dict for programmatic processing.

        Rendering applies only to registered rules. Lifetime and optional
        scope are encoded as node color and shape. Edges participating in a
        cycle are marked ``[CYCLE]``. The result is cached until the rule set
        changes, so repeated calls are free.

        Args:
            format: Output format among ``"mermaid"``, ``"graphviz"``,
                ``"json"``.

        Returns:
            str for ``"mermaid"``/``"graphviz"``, dict for ``"json"``.

        Raises:
            ValueError: If ``format`` is not supported.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("db", object())
            >>> builder.service("service", lambda db: db, deps=["db"])
            >>> c = builder.build()
            >>> out = c.visualize()
            >>> "graph TD" in out
            True
        """
        from .devkit.visualize import render

        if self.config.ruleset.version != self._visualize_version:
            self._visualize_cache = {}
            self._visualize_version = self.config.ruleset.version
        if format not in self._visualize_cache:
            self._visualize_cache[format] = render(self.config.ruleset, format)
        return self._visualize_cache[format]

    def validate(
        self,
        strict: bool = True,
    ) -> Optional[List[ValidationError]]:
        """Validate the whole dependency graph statically.

        Checks every registered rule for missing dependencies, dependency
        cycles, and factory arity mismatches. Validation is explicit and
        never runs automatically, so there is zero overhead unless called.

        Args:
            strict: When True, raise ValidationError on the first error.
                When False, collect all errors and return them as a list.

        Returns:
            None when strict=True and the graph is valid.
            List of ValidationError when strict=False.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("x", 1)
            >>> c = builder.build()
            >>> c.validate() is None
            True
        """
        errors: List[ValidationError] = []
        ruleset = self.config.ruleset

        for key, rule in ruleset.map.items():
            for dep in rule.deps:
                if dep not in ruleset.map:
                    errors.append(UnregisteredDependencyError(key, dep))

            try:
                sig = inspect.signature(rule.make)
            except (TypeError, ValueError):
                sig = None
            if sig is not None:
                positional = [
                    p
                    for p in sig.parameters.values()
                    if p.kind
                    in (
                        inspect.Parameter.POSITIONAL_ONLY,
                        inspect.Parameter.POSITIONAL_OR_KEYWORD,
                    )
                ]
                required = sum(1 for p in positional if p.default is inspect.Parameter.empty)
                total = len(positional)
                has_varargs = any(
                    p.kind == inspect.Parameter.VAR_POSITIONAL for p in sig.parameters.values()
                )
                if len(rule.deps) < required:
                    errors.append(
                        InvalidFactoryError(
                            key,
                            f"factory requires at least {required} args "
                            f"but only {len(rule.deps)} deps declared",
                        )
                    )
                elif len(rule.deps) > total and not has_varargs:
                    errors.append(
                        InvalidFactoryError(
                            key,
                            f"factory accepts at most {total} args "
                            f"but {len(rule.deps)} deps declared",
                        )
                    )

        for key in ruleset.map:
            try:
                ruleset._check_cycle(key)
            except CycleError as exc:
                errors.append(CyclicDependencyError(list(exc.path)))

        if strict:
            if errors:
                raise errors[0]
            return None
        return errors

    def compile(
        self,
        copy_parent_rules: bool = True,
        allow_post_compile_overrides: bool = True,
        guardless: bool = False,
    ) -> "ExecutionPlan":
        """Compile the dependency graph into an immutable execution plan.

        Computes a topological ordering of the registered rules, validates the
        full graph (missing dependencies raise :class:`MissingDependencyError`;
        cycles raise :class:`DependencyCycleError`) and returns an
        :class:`ExecutionPlan`. The plan delegates resolution to this container,
        so lifetimes, caches and scopes keep identical semantics.

        The plan is immutable and can be cached or serialized. If
        ``compile()`` is never called there is zero overhead.

        When ``copy_parent_rules`` is True (default) and this container is a
        child with a :class:`CompositeRuleSet`, a merged snapshot of the
        parent and local rules is taken so the plan is stable even if the
        parent is mutated later.

        When ``allow_post_compile_overrides`` is False, the plan freezes the
        graph at compile time: singletons are pre-resolved and the plan uses
        lockless resolvers. Any later ``override()`` on this container raises
        ``RuntimeError``. This is a breaking behavioral change for callers
        relying on override visibility through a compiled plan.

        Raises:
            MissingDependencyError: If a rule depends on an unregistered key.
            DependencyCycleError: If the graph contains a cycle.
            ContainerBuildError: If multiple missing dependencies are found.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.service("b", lambda a: a + 1, deps=["a"])
            >>> builder.value("a", 1)
            >>> container = builder.build()
            >>> plan = container.compile()
            >>> plan.get("b")
            2
        """
        from .plan import ExecutionPlan

        plan = ExecutionPlan.from_container(
            self,
            copy_parent_rules=copy_parent_rules,
            allow_post_compile_overrides=allow_post_compile_overrides,
            guardless=guardless,
        )
        if self.config.compile_policy == CompilePolicy.STRICT or plan.frozen:
            object.__setattr__(self, "_compiled_plan", plan)
        return plan

aget(key, qualifier=None, _stacks=None, _scope_name=None, _path=None, policy=None) async

Resolve a service by key asynchronously.

Resolves dependencies concurrently with asyncio.gather and awaits async factories. Sync factories are called directly, so there is no overhead for sync dependencies. Singleton results are cached.

Async yield providers are supported; their resources are finalized when resolution is cancelled.

Parameters:

Name	Type	Description	Default
key	Key	Service key.	required
qualifier	Optional[str]	Optional named qualifier. When given, the rule registered as (key, qualifier) is resolved.	None
policy	Optional['ResolutionPolicy']	Optional per-call resolution policy. Overrides the container-wide policy for this call only.	None

Examples:

>>> builder = ContainerBuilder()
>>> builder.service("answer", lambda: 42)
>>> container = builder.build()
>>> async def main():
...     return await container.aget("answer")
>>> import asyncio
>>> asyncio.run(main())
42

Source code in src/doppy_di/container.py

async def aget(
    self,
    key: Key,
    qualifier: Optional[str] = None,
    _stacks: Optional[List[AsyncExitStack]] = None,
    _scope_name: Optional[str] = None,
    _path: Optional[List[Key]] = None,
    policy: Optional["ResolutionPolicy"] = None,
) -> Any:
    """Resolve a service by key asynchronously.

    Resolves dependencies concurrently with ``asyncio.gather`` and awaits
    async factories. Sync factories are called directly, so there is no
    overhead for sync dependencies. Singleton results are cached.

    Async yield providers are supported; their resources are finalized
    when resolution is cancelled.

    Args:
        key: Service key.
        qualifier: Optional named qualifier. When given, the rule
            registered as ``(key, qualifier)`` is resolved.
        policy: Optional per-call resolution policy. Overrides the
            container-wide policy for this call only.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.service("answer", lambda: 42)
        >>> container = builder.build()
        >>> async def main():
        ...     return await container.aget("answer")
        >>> import asyncio
        >>> asyncio.run(main())
        42
    """
    lookup = (key, qualifier) if qualifier is not None else key
    active = policy if policy is not None else self._policy
    if active is not None and self._policy_depth == 0:
        return await self._resolve_with_policy_async(
            lookup,
            active,
            _stacks,
            _scope_name,
            _path,
        )
    started = self._tracer is not None
    start = time.perf_counter() if started else 0.0
    if self._override_layers:
        overridden = self._resolve_override(lookup)
        if overridden is not _unset:
            if inspect.isawaitable(overridden):
                overridden = await overridden
            if started:
                self._trace(lookup, time.perf_counter() - start, False, _scope_name)
            return overridden
    if lookup in self.single:
        if started:
            self._trace(lookup, time.perf_counter() - start, True, _scope_name)
        return self.single[lookup]

    path = self._enter_path(lookup, _path)
    try:
        if len(path) > 1 and lookup in path[:-1]:
            idx = path.index(lookup)
            raise DependencyCycleError(path[idx:])

        try:
            rule = self.config.ruleset.find(lookup)
        except ServiceNotFoundError:
            from .providers import implicit_collection_rule

            collection = implicit_collection_rule(lookup, self.config.ruleset)
            if collection is not None:
                rule = collection
            elif self._is_injectable_key(lookup):
                from .auto_wiring import _rule_for

                self.config.ruleset.add(lookup, _rule_for(lookup))
                rule = self.config.ruleset.find(lookup)
            elif qualifier is not None:
                raise UnregisteredDependencyError(key, qualifier) from None
            elif len(path) > 1:
                src = self.config.ruleset.map.get(
                    lookup, Rule(lookup, lambda: None)
                ).registration_source
                raise MissingDependencyError(
                    lookup,
                    path.copy(),
                    scope=_scope_name,
                    registration_source=src,
                ) from None
            else:
                raise
        stacks = _stacks if _stacks is not None else []
        try:
            if rule.async_yield_provider:
                stack = AsyncExitStack()
                stacks.append(stack)
                try:
                    obj = await stack.enter_async_context(asynccontextmanager(rule.make)())
                except RuntimeError as exc:
                    if "didn't yield" in str(exc):
                        raise YieldNotCalledError(lookup) from None
                    raise
                if rule.lifetime == "singleton":
                    self.single[lookup] = obj
                self._cache_nested_aliases(lookup, obj)
                if started:
                    self._trace(lookup, time.perf_counter() - start, False, _scope_name)
                return obj
            if rule.yield_provider:
                raise TypeError(f"Sync yield provider {lookup!r} cannot be resolved via aget")
            levels = self._independent_levels(list(rule.deps))
            if rule.deps and not levels:
                raise DependencyCycleError([lookup, *rule.deps])
            args_by_key: Dict[Key, Any] = {}
            for level in levels:
                resolved = await asyncio.gather(
                    *(
                        self.aget(
                            dep,
                            _stacks=stacks,
                            _scope_name=_scope_name,
                            _path=path,
                        )
                        for dep in level
                    )
                )
                args_by_key.update(dict(zip(level, resolved)))
            args = [args_by_key[dep] for dep in rule.deps]
            try:
                obj = rule.make(*args)
            except Exception as exc:
                if self.config.wrap_factory_errors:
                    raise FactoryExecutionError(lookup, exc, path.copy()) from exc
                raise
            if not rule.is_async and inspect.isawaitable(obj):
                raise SyncFactoryReturningAwaitableError(lookup)
            if inspect.isawaitable(obj):
                try:
                    obj = await obj
                except Exception as exc:
                    if self.config.wrap_factory_errors:
                        raise FactoryExecutionError(lookup, exc, path.copy()) from exc
                    raise

            if rule.lifetime == "singleton":
                self.single[lookup] = obj
            self._cache_nested_aliases(lookup, obj)

            if started:
                self._trace(lookup, time.perf_counter() - start, False, _scope_name)

            return obj
        except asyncio.CancelledError:
            errors: List[Tuple[Key, Exception]] = []
            for stack in stacks:
                try:
                    await stack.aclose()
                except Exception as exc:
                    if self.config.finalization_errors:
                        errors.append((lookup, exc))
                    else:
                        logger.exception("Error finalizing yield provider %r", lookup)
            if errors:
                raise ResourceFinalizationError(errors) from None
            raise ResolutionCancelledError(lookup) from None
    finally:
        path.pop()

ascope(name)

Return a named or unique async scope.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> c = builder.build()
>>> s = c.ascope("req")
>>> isinstance(s, AsyncScope)
True

Source code in src/doppy_di/container.py

def ascope(self, name: str) -> AsyncScope:
    """Return a named or unique async scope.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> c = builder.build()
        >>> s = c.ascope("req")
        >>> isinstance(s, AsyncScope)
        True
    """
    if self.scope_policy == ScopePolicy.NAMED:
        if name in self.scopes:
            existing = self.scopes[name]
            if isinstance(existing, AsyncScope):
                return existing
            raise TypeError(f"Scope {name!r} already exists as sync scope")
        scope = AsyncScope(self, name)
        self.scopes[name] = scope
        return scope
    # UNIQUE: fresh Scope per call, stored under unique internal key
    internal = f"{name}#{uuid.uuid4().hex}"
    scope = AsyncScope(self, name)
    self.scopes[internal] = scope
    return scope

child(name=None)

Return a new container layered over this one.

The child inherits the parent rules and can add or override rules without mutating the parent. Rules added to the parent after the child was created stay visible to the child.

Parameters:

Name	Type	Description	Default
name	Optional[str]	Optional profile name stored in config.profile.	None

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("db", "base-db")
>>> parent = builder.build()
>>> child = parent.child()
>>> child.value("db", "child-db")
>>> child.get("db")
'child-db'
>>> parent.get("db")
'base-db'

Source code in src/doppy_di/container.py

def child(self, name: Optional[str] = None) -> Container:
    """Return a new container layered over this one.

    The child inherits the parent rules and can add or override rules
    without mutating the parent. Rules added to the parent after the
    child was created stay visible to the child.

    Args:
        name: Optional profile name stored in ``config.profile``.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("db", "base-db")
        >>> parent = builder.build()
        >>> child = parent.child()
        >>> child.value("db", "child-db")
        >>> child.get("db")
        'child-db'
        >>> parent.get("db")
        'base-db'
    """
    composite = CompositeRuleSet(self.config.ruleset)
    child_container = Container(
        ContainerConfig(
            composite,
            scope_policy=self.config.scope_policy,
            track_sources=self.config.track_sources,
            wrap_factory_errors=self.config.wrap_factory_errors,
            finalization_errors=self.config.finalization_errors,
            profile=name,
        )
    )
    child_container._tracer = self._tracer
    child_container._policy = self._policy
    return child_container

compile(copy_parent_rules=True, allow_post_compile_overrides=True, guardless=False)

Compile the dependency graph into an immutable execution plan.

Computes a topological ordering of the registered rules, validates the full graph (missing dependencies raise :class:MissingDependencyError; cycles raise :class:DependencyCycleError) and returns an :class:ExecutionPlan. The plan delegates resolution to this container, so lifetimes, caches and scopes keep identical semantics.

The plan is immutable and can be cached or serialized. If compile() is never called there is zero overhead.

When copy_parent_rules is True (default) and this container is a child with a :class:CompositeRuleSet, a merged snapshot of the parent and local rules is taken so the plan is stable even if the parent is mutated later.

When allow_post_compile_overrides is False, the plan freezes the graph at compile time: singletons are pre-resolved and the plan uses lockless resolvers. Any later override() on this container raises RuntimeError. This is a breaking behavioral change for callers relying on override visibility through a compiled plan.

Raises:

Type	Description
MissingDependencyError	If a rule depends on an unregistered key.
DependencyCycleError	If the graph contains a cycle.
ContainerBuildError	If multiple missing dependencies are found.

Examples:

>>> builder = ContainerBuilder()
>>> builder.service("b", lambda a: a + 1, deps=["a"])
>>> builder.value("a", 1)
>>> container = builder.build()
>>> plan = container.compile()
>>> plan.get("b")
2

Source code in src/doppy_di/container.py

def compile(
    self,
    copy_parent_rules: bool = True,
    allow_post_compile_overrides: bool = True,
    guardless: bool = False,
) -> "ExecutionPlan":
    """Compile the dependency graph into an immutable execution plan.

    Computes a topological ordering of the registered rules, validates the
    full graph (missing dependencies raise :class:`MissingDependencyError`;
    cycles raise :class:`DependencyCycleError`) and returns an
    :class:`ExecutionPlan`. The plan delegates resolution to this container,
    so lifetimes, caches and scopes keep identical semantics.

    The plan is immutable and can be cached or serialized. If
    ``compile()`` is never called there is zero overhead.

    When ``copy_parent_rules`` is True (default) and this container is a
    child with a :class:`CompositeRuleSet`, a merged snapshot of the
    parent and local rules is taken so the plan is stable even if the
    parent is mutated later.

    When ``allow_post_compile_overrides`` is False, the plan freezes the
    graph at compile time: singletons are pre-resolved and the plan uses
    lockless resolvers. Any later ``override()`` on this container raises
    ``RuntimeError``. This is a breaking behavioral change for callers
    relying on override visibility through a compiled plan.

    Raises:
        MissingDependencyError: If a rule depends on an unregistered key.
        DependencyCycleError: If the graph contains a cycle.
        ContainerBuildError: If multiple missing dependencies are found.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.service("b", lambda a: a + 1, deps=["a"])
        >>> builder.value("a", 1)
        >>> container = builder.build()
        >>> plan = container.compile()
        >>> plan.get("b")
        2
    """
    from .plan import ExecutionPlan

    plan = ExecutionPlan.from_container(
        self,
        copy_parent_rules=copy_parent_rules,
        allow_post_compile_overrides=allow_post_compile_overrides,
        guardless=guardless,
    )
    if self.config.compile_policy == CompilePolicy.STRICT or plan.frozen:
        object.__setattr__(self, "_compiled_plan", plan)
    return plan

diff(other)

Return rule differences between this container and other.

Added keys exist only in other, removed keys exist only in this container, changed keys exist in both with different rule metadata (lifetime, deps, scope, resource flags).

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("a", 1)
>>> base = builder.build()
>>> modded = base.child()
>>> modded.value("b", 2)
>>> report = base.diff(modded)
>>> "b" in report.added
True

Source code in src/doppy_di/container.py

def diff(self, other: Container) -> DiffReport:
    """Return rule differences between this container and ``other``.

    Added keys exist only in ``other``, removed keys exist only in this
    container, changed keys exist in both with different rule metadata
    (lifetime, deps, scope, resource flags).

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("a", 1)
        >>> base = builder.build()
        >>> modded = base.child()
        >>> modded.value("b", 2)
        >>> report = base.diff(modded)
        >>> "b" in report.added
        True
    """
    own = self.config.ruleset.map
    other_map = other.config.ruleset.map
    added: List[Key] = []
    removed: List[Key] = []
    changed: List[Key] = []
    for key in other_map:
        if key not in own:
            added.append(key)
        elif _rule_signature(own[key]) != _rule_signature(other_map[key]):
            changed.append(key)
    for key in own:
        if key not in other_map:
            removed.append(key)
    return DiffReport(
        added=tuple(added),
        removed=tuple(removed),
        changed=tuple(changed),
    )

export_config(format='json')

Export the effective configuration as a JSON string.

Includes the container profile and a rule table with lifetime, deps, scope, and resource flags. Non-string keys are rendered via repr.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("a", 1)
>>> container = builder.build()
>>> out = container.export_config()
>>> '"a"' in out
True

Source code in src/doppy_di/container.py

def export_config(self, format: str = "json") -> str:  # noqa: A002
    """Export the effective configuration as a JSON string.

    Includes the container profile and a rule table with lifetime, deps,
    scope, and resource flags. Non-string keys are rendered via ``repr``.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("a", 1)
        >>> container = builder.build()
        >>> out = container.export_config()
        >>> '"a"' in out
        True
    """
    if format != "json":
        raise ValueError(f"Unsupported export format: {format!r}")
    rules: Dict[str, Any] = {}
    for key, rule in self.config.ruleset.map.items():
        rules[repr(key)] = {
            "lifetime": rule.lifetime,
            "deps": [repr(dep) for dep in rule.deps],
            "scope": rule.scope,
            "yield": rule.yield_provider or rule.async_yield_provider,
            "nested": rule.nested,
        }
    payload = {
        "profile": self.config.profile,
        "rules": rules,
    }
    return json.dumps(payload, sort_keys=True, indent=2)

get(key, qualifier=None, _scope_name=None, policy=None)

Resolve a service by key.

Returns the cached singleton if already resolved, otherwise resolves the rule from the config, applies the scope policy, and stores the result for singleton lifetimes.

Double-checked locking provides thread safety.

Parameters:

Name	Type	Description	Default
key	Key	Service key.	required
qualifier	Optional[str]	Optional named qualifier. When given, resolves the rule registered as (key, qualifier).	None
policy	Optional['ResolutionPolicy']	Optional per-call resolution policy. Overrides the container-wide policy for this call only.	None

Examples:

>>> builder = ContainerBuilder()
>>> builder.service("answer", lambda: 42)
>>> container = builder.build()
>>> container.get("answer")
42

Source code in src/doppy_di/container.py

def get(
    self,
    key: Key,
    qualifier: Optional[str] = None,
    _scope_name: Optional[str] = None,
    policy: Optional["ResolutionPolicy"] = None,
) -> Any:
    """Resolve a service by key.

    Returns the cached singleton if already resolved, otherwise resolves
    the rule from the config, applies the scope policy, and stores the
    result for singleton lifetimes.

    Double-checked locking provides thread safety.

    Args:
        key: Service key.
        qualifier: Optional named qualifier. When given, resolves the
            rule registered as ``(key, qualifier)``.
        policy: Optional per-call resolution policy. Overrides the
            container-wide policy for this call only.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.service("answer", lambda: 42)
        >>> container = builder.build()
        >>> container.get("answer")
        42
    """
    lookup = (key, qualifier) if qualifier is not None else key
    active = policy if policy is not None else self._policy
    if active is not None and self._policy_depth == 0:
        return self._resolve_with_policy(lookup, active, _scope_name)
    started = self._tracer is not None
    start = time.perf_counter() if started else 0.0
    if self._override_layers:
        overridden = self._resolve_override(lookup)
        if overridden is not _unset:
            if started:
                self._trace(lookup, time.perf_counter() - start, False, _scope_name)
            return overridden
    if lookup in self.single:
        if started:
            self._trace(lookup, time.perf_counter() - start, True, _scope_name)
        return self.single[lookup]

    path = self._enter_path(lookup)
    try:
        if len(path) > 1 and lookup in path[:-1]:
            idx = path.index(lookup)
            raise DependencyCycleError(path[idx:])

        with self.lock:
            if lookup in self.single:
                if started:
                    self._trace(lookup, time.perf_counter() - start, True, _scope_name)
                return self.single[lookup]

            try:
                rule = self.config.ruleset.find(lookup)
            except ServiceNotFoundError:
                from .providers import implicit_collection_rule

                collection = implicit_collection_rule(lookup, self.config.ruleset)
                if collection is not None:
                    rule = collection
                elif self._is_injectable_key(lookup):
                    from .auto_wiring import _rule_for

                    self.config.ruleset.add(lookup, _rule_for(lookup))
                    rule = self.config.ruleset.find(lookup)
                elif qualifier is not None:
                    raise UnregisteredDependencyError(key, qualifier) from None
                elif len(path) > 1:
                    src = self.config.ruleset.map.get(
                        lookup, Rule(lookup, lambda: None)
                    ).registration_source
                    raise MissingDependencyError(
                        lookup,
                        path.copy(),
                        scope=_scope_name,
                        registration_source=src,
                    ) from None
                else:
                    raise
            if rule.async_yield_provider:
                raise TypeError(f"Async yield provider {lookup!r} requires async scope")
            if rule.is_async:
                raise AsyncDependencyInSyncContextError(lookup)
            ctx = ResolveContext(self)
            try:
                args = [ctx.get(dep, _scope_name=_scope_name) for dep in rule.deps]
            except ServiceNotFoundError as exc:
                if self._is_injectable_key(lookup):
                    from .auto_wiring import UnresolvableDependencyError

                    raise UnresolvableDependencyError(lookup, exc.key) from None
                if isinstance(exc, MissingDependencyError):
                    if exc.registration_source is not None:
                        raise
                    src = self.config.ruleset.map.get(
                        lookup, Rule(lookup, lambda: None)
                    ).registration_source
                    if src is None:
                        raise
                    raise MissingDependencyError(
                        exc.key,
                        exc.resolution_path,
                        scope=exc.scope or _scope_name,
                        registration_source=src,
                    ) from None
                if len(path) > 1:
                    src = self.config.ruleset.map.get(
                        lookup, Rule(lookup, lambda: None)
                    ).registration_source
                    raise MissingDependencyError(
                        exc.key,
                        path.copy(),
                        scope=_scope_name,
                        registration_source=src,
                    ) from None
                raise
            try:
                obj = rule.make(*args)
            except Exception as exc:
                if self.config.wrap_factory_errors:
                    raise FactoryExecutionError(
                        lookup,
                        exc,
                        path.copy(),
                    ) from exc
                raise
            if inspect.isawaitable(obj):
                raise SyncFactoryReturningAwaitableError(lookup)

            if rule.lifetime == "singleton":
                self.single[lookup] = obj
            self._cache_nested_aliases(lookup, obj)

            if started:
                self._trace(lookup, time.perf_counter() - start, False, _scope_name)

            return obj
    finally:
        path.pop()

get_many(keys, parallel=False) async

Resolve multiple services, optionally in parallel.

Parameters:

Name	Type	Description	Default
keys	List[Key]	Service keys to resolve.	required
parallel	bool	When True, resolve independent dependencies concurrently. Falls back to sequential resolution for small graphs (fewer than 5 nodes) where parallelism overhead outweighs the benefit.	False

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("a", 1)
>>> builder.value("b", 2)
>>> container = builder.build()
>>> async def main():
...     return await container.get_many(["a", "b"])
>>> import asyncio
>>> asyncio.run(main())
[1, 2]

Source code in src/doppy_di/container.py

async def get_many(self, keys: List[Key], parallel: bool = False) -> List[Any]:
    """Resolve multiple services, optionally in parallel.

    Args:
        keys: Service keys to resolve.
        parallel: When True, resolve independent dependencies
            concurrently. Falls back to sequential resolution for small
            graphs (fewer than 5 nodes) where parallelism overhead
            outweighs the benefit.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("a", 1)
        >>> builder.value("b", 2)
        >>> container = builder.build()
        >>> async def main():
        ...     return await container.get_many(["a", "b"])
        >>> import asyncio
        >>> asyncio.run(main())
        [1, 2]
    """
    if not parallel:
        return [await self.aget(key) for key in keys]

    levels = self._independent_levels(keys)
    total = sum(len(level) for level in levels)
    if total < 5:
        return [await self.aget(key) for key in keys]

    results: Dict[Key, Any] = {}
    for level in levels:
        resolved = await asyncio.gather(*(self.aget(key) for key in level))
        results.update(dict(zip(level, resolved)))
    return [results[key] for key in keys]

get_or_none(key, qualifier=None)

Return resolved service or None if key not registered.

Parameters:

Name	Type	Description	Default
key	Key	Service key.	required
qualifier	Optional[str]	Optional named qualifier.	None

Examples:

>>> builder = ContainerBuilder()
>>> c = builder.build()
>>> c.get_or_none("missing") is None
True

Source code in src/doppy_di/container.py

def get_or_none(self, key: Key, qualifier: Optional[str] = None) -> Any:
    """Return resolved service or ``None`` if key not registered.

    Args:
        key: Service key.
        qualifier: Optional named qualifier.

    Examples:
        >>> builder = ContainerBuilder()
        >>> c = builder.build()
        >>> c.get_or_none("missing") is None
        True
    """
    try:
        return self.get(key, qualifier=qualifier)
    except (ServiceNotFoundError, UnregisteredDependencyError):
        return None

graph()

Return a DependencyGraph representation of this container.

Source code in src/doppy_di/container.py

def graph(self) -> DependencyGraph:
    """Return a DependencyGraph representation of this container."""
    from .graph import DependencyGraph

    return DependencyGraph(self.config.ruleset)

has(key, qualifier=None)

Return True if a rule for key is registered.

Parameters:

Name	Type	Description	Default
key	Key	Service key.	required
qualifier	Optional[str]	Optional named qualifier.	None

Examples:

>>> builder = ContainerBuilder()
>>> builder.service("x", lambda: 1)
>>> c = builder.build()
>>> c.has("x")
True
>>> c.has("missing")
False

Source code in src/doppy_di/container.py

def has(self, key: Key, qualifier: Optional[str] = None) -> bool:
    """Return True if a rule for key is registered.

    Args:
        key: Service key.
        qualifier: Optional named qualifier.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.service("x", lambda: 1)
        >>> c = builder.build()
        >>> c.has("x")
        True
        >>> c.has("missing")
        False
    """
    lookup = (key, qualifier) if qualifier is not None else key
    return self.config.ruleset.has(lookup)

override(key, value=None, **overrides)

Create a temporary override context.

Supports both a single key/value pair and a dictionary of overrides: container.override({"a": 1, "b": 2}).

Nested overrides stack LIFO: the last override() wins. On context exit the previous state is restored.

Callable override values are treated as factories and invoked on every resolution.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> c = builder.build()
>>> ctx = c.override("x", 2)
>>> isinstance(ctx, OverrideContext)
True

>>> with c.override({"x": 3}):
...     c.get("x")
3
>>> c.get("x")
1

Source code in src/doppy_di/container.py

def override(
    self,
    key: Union[Key, Dict[Key, Any]],
    value: Any = None,
    **overrides: Any,
) -> OverrideContext:
    """Create a temporary override context.

    Supports both a single ``key``/``value`` pair and a dictionary of
    overrides: ``container.override({"a": 1, "b": 2})``.

    Nested overrides stack LIFO: the last ``override()`` wins. On
    context exit the previous state is restored.

    Callable override values are treated as factories and invoked on
    every resolution.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> c = builder.build()
        >>> ctx = c.override("x", 2)
        >>> isinstance(ctx, OverrideContext)
        True

        >>> with c.override({"x": 3}):
        ...     c.get("x")
        3
        >>> c.get("x")
        1
    """
    if isinstance(key, dict):
        values: Dict[Key, Any] = dict(key)
    else:
        values = {key: value}
    for override_key, override_value in overrides.items():
        values[override_key] = override_value
    return OverrideContext(self, values)

scan(*packages, recursive=True)

Register all injectable classes found in the given packages.

Explicitly registered rules are never overridden.

Examples:

>>> builder = ContainerBuilder()
>>> container = builder.build()
>>> container.scan(__name__)

Source code in src/doppy_di/container.py

def scan(
    self,
    *packages: Union[ModuleType, str],
    recursive: bool = True,
) -> None:
    """Register all injectable classes found in the given packages.

    Explicitly registered rules are never overridden.

    Examples:
        >>> builder = ContainerBuilder()
        >>> container = builder.build()
        >>> container.scan(__name__)
    """
    from .auto_wiring import scan_package

    for pkg in packages:
        scan_package(self, pkg, recursive)

scope(name)

Return a named or unique scope according to the active policy.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> c = builder.build()
>>> s = c.scope("req")
>>> isinstance(s, Scope)
True

Source code in src/doppy_di/container.py

def scope(self, name: str) -> Scope:
    """Return a named or unique scope according to the active policy.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> c = builder.build()
        >>> s = c.scope("req")
        >>> isinstance(s, Scope)
        True
    """
    if self.scope_policy == ScopePolicy.NAMED:
        if name in self.scopes:
            return self.scopes[name]
        scope = Scope(self, name)
        self.scopes[name] = scope
        return scope
    # UNIQUE: fresh Scope per call, stored under unique internal key
    internal = f"{name}#{uuid.uuid4().hex}"
    scope = Scope(self, name)
    self.scopes[internal] = scope
    return scope

service(key, make, lifetime='transient', deps=None, qualifier=None, scope=None)

Register a factory service on this container.

Examples:

>>> builder = ContainerBuilder()
>>> container = builder.build()
>>> container.service("a", lambda: 1)
>>> container.get("a")
1

Source code in src/doppy_di/container.py

def service(
    self,
    key: Key,
    make: Callable[..., Any],
    lifetime: Lifetime = "transient",
    deps: Optional[List[Key]] = None,
    qualifier: Optional[str] = None,
    scope: Optional[str] = None,
) -> Self:
    """Register a factory service on this container.

    Examples:
        >>> builder = ContainerBuilder()
        >>> container = builder.build()
        >>> container.service("a", lambda: 1)
        >>> container.get("a")
        1
    """
    lookup = (key, qualifier) if qualifier is not None else key
    rule = Rule(
        key=lookup,
        make=make,
        lifetime=lifetime,
        deps=tuple(deps or ()),
        scope=scope,
    )
    self.config.ruleset.add(lookup, rule)
    return self

set_tracer(tracer_fn)

Set a tracer callback or disable tracing with None.

The callback receives (key, duration, cache_hit, scope) after every successful resolution. When no tracer is set there is no timing and no dispatch, so overhead is zero.

Parameters:

Name	Type	Description	Default
tracer_fn	Optional[TracerFn]	Callback receiving trace events, or None to disable tracing.	required

Examples:

>>> events = []
>>> builder = ContainerBuilder()
>>> builder.value("a", 1)
>>> container = builder.build()
>>> container.set_tracer(lambda *args: events.append(args))
>>> container.get("a")
1
>>> len(events)
1
>>> container.set_tracer(None)
>>> container.get("a")
1
>>> len(events)
1

Source code in src/doppy_di/container.py

def set_tracer(self, tracer_fn: Optional[TracerFn]) -> None:
    """Set a tracer callback or disable tracing with ``None``.

    The callback receives ``(key, duration, cache_hit, scope)`` after
    every successful resolution. When no tracer is set there is no
    timing and no dispatch, so overhead is zero.

    Args:
        tracer_fn: Callback receiving trace events, or ``None`` to
            disable tracing.

    Examples:
        >>> events = []
        >>> builder = ContainerBuilder()
        >>> builder.value("a", 1)
        >>> container = builder.build()
        >>> container.set_tracer(lambda *args: events.append(args))
        >>> container.get("a")
        1
        >>> len(events)
        1
        >>> container.set_tracer(None)
        >>> container.get("a")
        1
        >>> len(events)
        1
    """
    self._tracer = tracer_fn

validate(strict=True)

Validate the whole dependency graph statically.

Checks every registered rule for missing dependencies, dependency cycles, and factory arity mismatches. Validation is explicit and never runs automatically, so there is zero overhead unless called.

Parameters:

Name	Type	Description	Default
strict	bool	When True, raise ValidationError on the first error. When False, collect all errors and return them as a list.	True

Returns:

Type	Description
Optional[List[ValidationError]]	None when strict=True and the graph is valid.
Optional[List[ValidationError]]	List of ValidationError when strict=False.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> c = builder.build()
>>> c.validate() is None
True

Source code in src/doppy_di/container.py

def validate(
    self,
    strict: bool = True,
) -> Optional[List[ValidationError]]:
    """Validate the whole dependency graph statically.

    Checks every registered rule for missing dependencies, dependency
    cycles, and factory arity mismatches. Validation is explicit and
    never runs automatically, so there is zero overhead unless called.

    Args:
        strict: When True, raise ValidationError on the first error.
            When False, collect all errors and return them as a list.

    Returns:
        None when strict=True and the graph is valid.
        List of ValidationError when strict=False.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> c = builder.build()
        >>> c.validate() is None
        True
    """
    errors: List[ValidationError] = []
    ruleset = self.config.ruleset

    for key, rule in ruleset.map.items():
        for dep in rule.deps:
            if dep not in ruleset.map:
                errors.append(UnregisteredDependencyError(key, dep))

        try:
            sig = inspect.signature(rule.make)
        except (TypeError, ValueError):
            sig = None
        if sig is not None:
            positional = [
                p
                for p in sig.parameters.values()
                if p.kind
                in (
                    inspect.Parameter.POSITIONAL_ONLY,
                    inspect.Parameter.POSITIONAL_OR_KEYWORD,
                )
            ]
            required = sum(1 for p in positional if p.default is inspect.Parameter.empty)
            total = len(positional)
            has_varargs = any(
                p.kind == inspect.Parameter.VAR_POSITIONAL for p in sig.parameters.values()
            )
            if len(rule.deps) < required:
                errors.append(
                    InvalidFactoryError(
                        key,
                        f"factory requires at least {required} args "
                        f"but only {len(rule.deps)} deps declared",
                    )
                )
            elif len(rule.deps) > total and not has_varargs:
                errors.append(
                    InvalidFactoryError(
                        key,
                        f"factory accepts at most {total} args "
                        f"but {len(rule.deps)} deps declared",
                    )
                )

    for key in ruleset.map:
        try:
            ruleset._check_cycle(key)
        except CycleError as exc:
            errors.append(CyclicDependencyError(list(exc.path)))

    if strict:
        if errors:
            raise errors[0]
        return None
    return errors

value(key, value)

Register a constant value on this container.

Child containers inherit this rule; overriding it on a child does not touch the parent.

Examples:

>>> builder = ContainerBuilder()
>>> container = builder.build()
>>> container.value("env", "base")
>>> container.get("env")
'base'

Source code in src/doppy_di/container.py

def value(self, key: Key, value: Any) -> Self:
    """Register a constant value on this container.

    Child containers inherit this rule; overriding it on a child does
    not touch the parent.

    Examples:
        >>> builder = ContainerBuilder()
        >>> container = builder.build()
        >>> container.value("env", "base")
        >>> container.get("env")
        'base'
    """

    def make_value() -> Any:
        return value

    self.config.ruleset.add(
        key,
        Rule(
            key=key,
            make=make_value,
            lifetime="singleton",
            deps=(),
        ),
    )
    return self

visualize(format='mermaid')

Return a textual representation of the dependency graph.

Supported formats

• "mermaid" — mermaid graph TD for embedding in Markdown.
• "graphviz" — Graphviz digraph for PNG/SVG generation.
• "json" — structured dict for programmatic processing.

Rendering applies only to registered rules. Lifetime and optional scope are encoded as node color and shape. Edges participating in a cycle are marked [CYCLE]. The result is cached until the rule set changes, so repeated calls are free.

Parameters:

Name	Type	Description	Default
format	str	Output format among "mermaid", "graphviz", "json".	'mermaid'

Returns:

Type	Description
Any	str for "mermaid"/"graphviz", dict for "json".

Raises:

Type	Description
ValueError	If format is not supported.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("db", object())
>>> builder.service("service", lambda db: db, deps=["db"])
>>> c = builder.build()
>>> out = c.visualize()
>>> "graph TD" in out
True

Source code in src/doppy_di/container.py

def visualize(self, format: str = "mermaid") -> Any:  # noqa: A002
    """Return a textual representation of the dependency graph.

    Supported formats:
        - ``"mermaid"`` — mermaid ``graph TD`` for embedding in Markdown.
        - ``"graphviz"`` — Graphviz ``digraph`` for PNG/SVG generation.
        - ``"json"`` — structured dict for programmatic processing.

    Rendering applies only to registered rules. Lifetime and optional
    scope are encoded as node color and shape. Edges participating in a
    cycle are marked ``[CYCLE]``. The result is cached until the rule set
    changes, so repeated calls are free.

    Args:
        format: Output format among ``"mermaid"``, ``"graphviz"``,
            ``"json"``.

    Returns:
        str for ``"mermaid"``/``"graphviz"``, dict for ``"json"``.

    Raises:
        ValueError: If ``format`` is not supported.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("db", object())
        >>> builder.service("service", lambda db: db, deps=["db"])
        >>> c = builder.build()
        >>> out = c.visualize()
        >>> "graph TD" in out
        True
    """
    from .devkit.visualize import render

    if self.config.ruleset.version != self._visualize_version:
        self._visualize_cache = {}
        self._visualize_version = self.config.ruleset.version
    if format not in self._visualize_cache:
        self._visualize_cache[format] = render(self.config.ruleset, format)
    return self._visualize_cache[format]

with_profile(name, overrides=None)

Return a derived child container with the given overrides applied.

Each override is registered as a singleton value on the child. Keys must already exist in the effective rule set; unknown keys raise :class:UnregisteredTypeError.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("env", "base")
>>> container = builder.build()
>>> prod = container.with_profile("prod", {"env": "prod"})
>>> container.get("env")
'base'
>>> prod.get("env")
'prod'
>>> prod.config.profile
'prod'

Source code in src/doppy_di/container.py

def with_profile(
    self,
    name: str,
    overrides: Optional[Dict[Key, Any]] = None,
) -> Container:
    """Return a derived child container with the given overrides applied.

    Each override is registered as a singleton value on the child. Keys
    must already exist in the effective rule set; unknown keys raise
    :class:`UnregisteredTypeError`.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("env", "base")
        >>> container = builder.build()
        >>> prod = container.with_profile("prod", {"env": "prod"})
        >>> container.get("env")
        'base'
        >>> prod.get("env")
        'prod'
        >>> prod.config.profile
        'prod'
    """
    derived = self.child(name)
    for key, item in (overrides or {}).items():
        if not derived.has(key):
            raise UnregisteredTypeError(key)
        derived.value(key, item)
    return derived

ContainerBuilder

Builder for a container.

Provides methods to register services, values, and aliases, then produce a ready-to-use Container.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> builder.service("y", lambda x: x + 1, deps=["x"])
>>> c = builder.build()
>>> c.get("y")
2

Source code in src/doppy_di/container.py

class ContainerBuilder:
    """Builder for a container.


    Provides methods to register services, values, and aliases, then
    produce a ready-to-use Container.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> builder.service("y", lambda x: x + 1, deps=["x"])
        >>> c = builder.build()
        >>> c.get("y")
        2
    """

    __slots__ = (
        "check_cycles_on_register",
        "compile_policy",
        "duplicate_policy",
        "finalization_errors",
        "rules",
        "scope_policy",
        "track_sources",
        "wrap_factory_errors",
    )

    def __init__(
        self,
        duplicate_policy: DuplicateKeyPolicy = DuplicateKeyPolicy.OVERWRITE,
        scope_policy: ScopePolicy = ScopePolicy.NAMED,
        compile_policy: CompilePolicy = CompilePolicy.ALLOW_OVERRIDE,
        track_sources: bool = False,
        wrap_factory_errors: bool = False,
        finalization_errors: bool = False,
        check_cycles_on_register: bool = True,
    ) -> None:
        """Initialize builder with optional policies.

        Examples:
            >>> b = ContainerBuilder(
            ...     DuplicateKeyPolicy.FAIL, ScopePolicy.UNIQUE
            ... )
            >>> b.duplicate_policy
            <DuplicateKeyPolicy.FAIL: 'fail'>
        """
        self.rules = RuleSet(defer_cycle_check=not check_cycles_on_register)
        self.duplicate_policy = duplicate_policy
        self.scope_policy = scope_policy
        self.compile_policy = compile_policy
        self.track_sources = track_sources
        self.wrap_factory_errors = wrap_factory_errors
        self.finalization_errors = finalization_errors
        self.check_cycles_on_register = check_cycles_on_register

    def _capture_source(self) -> Optional[RegistrationSource]:
        if not self.track_sources:
            return None
        frame = inspect.currentframe()
        try:
            # _capture_source -> _register -> service/value/alias -> user
            frame = frame.f_back if frame is not None else None
            frame = frame.f_back if frame is not None else None
            frame = frame.f_back if frame is not None else None
            if frame is None:
                return None
            return RegistrationSource(
                filename=frame.f_code.co_filename,
                lineno=frame.f_lineno,
                function_name=frame.f_code.co_name,
            )
        finally:
            del frame

    def _register(self, key: Key, rule: Rule) -> None:
        """Add rule honoring the active duplicate policy."""
        source = self._capture_source()
        existing = self.rules.map.get(key)
        if existing is not None and (self.duplicate_policy != DuplicateKeyPolicy.OVERWRITE):
            if self.duplicate_policy == DuplicateKeyPolicy.FAIL:
                raise DuplicateRegistrationError(
                    key,
                    existing_source=existing.registration_source,
                    new_source=source,
                )
            # WARN
            logger.warning("Duplicate key %r registered; overwriting", key)
        if source is not None:
            object.__setattr__(rule, "registration_source", source)
        self.rules.add(key, rule)

    def service(
        self,
        key: Key,
        make: Callable[..., Any],
        lifetime: Lifetime = "transient",
        deps: Optional[List[Key]] = None,
        qualifier: Optional[str] = None,
        scope: Optional[str] = None,
    ) -> Self:
        """Register a factory service.

        Args:
            key: Registration key.
            make: Factory callable.
            lifetime: Service lifetime.
            deps: Dependency keys.
            qualifier: Optional named qualifier. When given, the rule is
                stored under the key ``(key, qualifier)``.

        Examples:
            >>> b = ContainerBuilder()
            >>> b.service("greet", lambda name: f"Hello {name}", deps=["name"])
            >>> b.value("name", "World")
            >>> c = b.build()
            >>> c.get("greet")
            'Hello World'
        """
        lookup = (key, qualifier) if qualifier is not None else key
        rule = Rule(
            key=lookup,
            make=make,
            lifetime=lifetime,
            deps=tuple(deps or ()),
            scope=scope,
        )
        self._register(lookup, rule)
        return self

    def value(self, key: Key, value: Any) -> Self:
        """Register a constant value as a singleton.

        Examples:
            >>> b = ContainerBuilder()
            >>> b.value("pi", 3.14)
            >>> c = b.build()
            >>> c.get("pi")
            3.14
        """

        def make_value() -> Any:
            return value

        self._register(
            key,
            Rule(
                key=key,
                make=make_value,
                lifetime="singleton",
                deps=(),
            ),
        )
        return self

    def alias(self, key: Key, target: Key) -> Self:
        """Register an alias pointing to another key.

        Examples:
            >>> b = ContainerBuilder()
            >>> b.value("x", 42)
            >>> b.alias("answer", "x")
            >>> c = b.build()
            >>> c.get("answer")
            42
        """

        def make_alias(value: Any) -> Any:
            return value

        self._register(
            key,
            Rule(
                key=key,
                make=make_alias,
                lifetime="transient",
                deps=(target,),
            ),
        )
        return self

    def build(
        self,
        validate: bool = False,
        policy: Optional["ResolutionPolicy"] = None,
    ) -> Container:
        """Build and return a Container.

        Args:
            validate: When True, raises ContainerBuildError if any
                dependency key is not registered.
            policy: Optional resolution policy applied to the container.

        Examples:
            >>> b = ContainerBuilder()
            >>> b.value("x", 1)
            >>> c = b.build()
            >>> c.get("x")
            1

        Example with validation:
            >>> b = ContainerBuilder()
            >>> b.service("a", lambda b: b, deps=["b"])
            >>> b.build(validate=True)  # doctest: +IGNORE_EXCEPTION_DETAIL
            Traceback (most recent call last):
            ...
            ContainerBuildError
        """
        if validate:
            missing: List[Tuple[Key, Key]] = []
            for key, rule in self.rules.map.items():
                for dep in rule.deps:
                    if dep not in self.rules.map:
                        missing.append((key, dep))
            if missing:
                raise ContainerBuildError(missing)
        container = Container(
            ContainerConfig(
                self.rules,
                scope_policy=self.scope_policy,
                compile_policy=self.compile_policy,
                track_sources=self.track_sources,
                wrap_factory_errors=self.wrap_factory_errors,
                finalization_errors=self.finalization_errors,
                policy=policy,
            )
        )
        if policy is not None:
            from .resolution import EagerPolicy

            if isinstance(policy, EagerPolicy):
                first = next(iter(self.rules.map), None)
                if first is not None:
                    for key in policy.order(self.rules.map, first):
                        if key in self.rules.map:
                            container.get(key)
        return container

__init__(duplicate_policy=DuplicateKeyPolicy.OVERWRITE, scope_policy=ScopePolicy.NAMED, compile_policy=CompilePolicy.ALLOW_OVERRIDE, track_sources=False, wrap_factory_errors=False, finalization_errors=False, check_cycles_on_register=True)

Initialize builder with optional policies.

Examples:

>>> b = ContainerBuilder(
...     DuplicateKeyPolicy.FAIL, ScopePolicy.UNIQUE
... )
>>> b.duplicate_policy
<DuplicateKeyPolicy.FAIL: 'fail'>

Source code in src/doppy_di/container.py

def __init__(
    self,
    duplicate_policy: DuplicateKeyPolicy = DuplicateKeyPolicy.OVERWRITE,
    scope_policy: ScopePolicy = ScopePolicy.NAMED,
    compile_policy: CompilePolicy = CompilePolicy.ALLOW_OVERRIDE,
    track_sources: bool = False,
    wrap_factory_errors: bool = False,
    finalization_errors: bool = False,
    check_cycles_on_register: bool = True,
) -> None:
    """Initialize builder with optional policies.

    Examples:
        >>> b = ContainerBuilder(
        ...     DuplicateKeyPolicy.FAIL, ScopePolicy.UNIQUE
        ... )
        >>> b.duplicate_policy
        <DuplicateKeyPolicy.FAIL: 'fail'>
    """
    self.rules = RuleSet(defer_cycle_check=not check_cycles_on_register)
    self.duplicate_policy = duplicate_policy
    self.scope_policy = scope_policy
    self.compile_policy = compile_policy
    self.track_sources = track_sources
    self.wrap_factory_errors = wrap_factory_errors
    self.finalization_errors = finalization_errors
    self.check_cycles_on_register = check_cycles_on_register

alias(key, target)

Register an alias pointing to another key.

Examples:

>>> b = ContainerBuilder()
>>> b.value("x", 42)
>>> b.alias("answer", "x")
>>> c = b.build()
>>> c.get("answer")
42

Source code in src/doppy_di/container.py

def alias(self, key: Key, target: Key) -> Self:
    """Register an alias pointing to another key.

    Examples:
        >>> b = ContainerBuilder()
        >>> b.value("x", 42)
        >>> b.alias("answer", "x")
        >>> c = b.build()
        >>> c.get("answer")
        42
    """

    def make_alias(value: Any) -> Any:
        return value

    self._register(
        key,
        Rule(
            key=key,
            make=make_alias,
            lifetime="transient",
            deps=(target,),
        ),
    )
    return self

build(validate=False, policy=None)

Build and return a Container.

Parameters:

Name	Type	Description	Default
validate	bool	When True, raises ContainerBuildError if any dependency key is not registered.	False
policy	Optional['ResolutionPolicy']	Optional resolution policy applied to the container.	None

Examples:

>>> b = ContainerBuilder()
>>> b.value("x", 1)
>>> c = b.build()
>>> c.get("x")
1

Example with validation

b = ContainerBuilder() b.service("a", lambda b: b, deps=["b"]) b.build(validate=True) # doctest: +IGNORE_EXCEPTION_DETAIL Traceback (most recent call last): ... ContainerBuildError

Source code in src/doppy_di/container.py

def build(
    self,
    validate: bool = False,
    policy: Optional["ResolutionPolicy"] = None,
) -> Container:
    """Build and return a Container.

    Args:
        validate: When True, raises ContainerBuildError if any
            dependency key is not registered.
        policy: Optional resolution policy applied to the container.

    Examples:
        >>> b = ContainerBuilder()
        >>> b.value("x", 1)
        >>> c = b.build()
        >>> c.get("x")
        1

    Example with validation:
        >>> b = ContainerBuilder()
        >>> b.service("a", lambda b: b, deps=["b"])
        >>> b.build(validate=True)  # doctest: +IGNORE_EXCEPTION_DETAIL
        Traceback (most recent call last):
        ...
        ContainerBuildError
    """
    if validate:
        missing: List[Tuple[Key, Key]] = []
        for key, rule in self.rules.map.items():
            for dep in rule.deps:
                if dep not in self.rules.map:
                    missing.append((key, dep))
        if missing:
            raise ContainerBuildError(missing)
    container = Container(
        ContainerConfig(
            self.rules,
            scope_policy=self.scope_policy,
            compile_policy=self.compile_policy,
            track_sources=self.track_sources,
            wrap_factory_errors=self.wrap_factory_errors,
            finalization_errors=self.finalization_errors,
            policy=policy,
        )
    )
    if policy is not None:
        from .resolution import EagerPolicy

        if isinstance(policy, EagerPolicy):
            first = next(iter(self.rules.map), None)
            if first is not None:
                for key in policy.order(self.rules.map, first):
                    if key in self.rules.map:
                        container.get(key)
    return container

service(key, make, lifetime='transient', deps=None, qualifier=None, scope=None)

Register a factory service.

Parameters:

Name	Type	Description	Default
key	Key	Registration key.	required
make	Callable[..., Any]	Factory callable.	required
lifetime	Lifetime	Service lifetime.	'transient'
deps	Optional[List[Key]]	Dependency keys.	None
qualifier	Optional[str]	Optional named qualifier. When given, the rule is stored under the key (key, qualifier).	None

Examples:

>>> b = ContainerBuilder()
>>> b.service("greet", lambda name: f"Hello {name}", deps=["name"])
>>> b.value("name", "World")
>>> c = b.build()
>>> c.get("greet")
'Hello World'

Source code in src/doppy_di/container.py

def service(
    self,
    key: Key,
    make: Callable[..., Any],
    lifetime: Lifetime = "transient",
    deps: Optional[List[Key]] = None,
    qualifier: Optional[str] = None,
    scope: Optional[str] = None,
) -> Self:
    """Register a factory service.

    Args:
        key: Registration key.
        make: Factory callable.
        lifetime: Service lifetime.
        deps: Dependency keys.
        qualifier: Optional named qualifier. When given, the rule is
            stored under the key ``(key, qualifier)``.

    Examples:
        >>> b = ContainerBuilder()
        >>> b.service("greet", lambda name: f"Hello {name}", deps=["name"])
        >>> b.value("name", "World")
        >>> c = b.build()
        >>> c.get("greet")
        'Hello World'
    """
    lookup = (key, qualifier) if qualifier is not None else key
    rule = Rule(
        key=lookup,
        make=make,
        lifetime=lifetime,
        deps=tuple(deps or ()),
        scope=scope,
    )
    self._register(lookup, rule)
    return self

value(key, value)

Register a constant value as a singleton.

Examples:

>>> b = ContainerBuilder()
>>> b.value("pi", 3.14)
>>> c = b.build()
>>> c.get("pi")
3.14

Source code in src/doppy_di/container.py

def value(self, key: Key, value: Any) -> Self:
    """Register a constant value as a singleton.

    Examples:
        >>> b = ContainerBuilder()
        >>> b.value("pi", 3.14)
        >>> c = b.build()
        >>> c.get("pi")
        3.14
    """

    def make_value() -> Any:
        return value

    self._register(
        key,
        Rule(
            key=key,
            make=make_value,
            lifetime="singleton",
            deps=(),
        ),
    )
    return self

ContainerConfig dataclass

Immutable container configuration.

Examples:

>>> rules = RuleSet()
>>> config = ContainerConfig(rules)
>>> isinstance(config.ruleset, RuleSet)
True
>>> config.scope_policy
<ScopePolicy.NAMED: 'named'>

Source code in src/doppy_di/container.py

@dataclass(frozen=True)
class ContainerConfig:
    """Immutable container configuration.

    Examples:
        >>> rules = RuleSet()
        >>> config = ContainerConfig(rules)
        >>> isinstance(config.ruleset, RuleSet)
        True
        >>> config.scope_policy
        <ScopePolicy.NAMED: 'named'>
    """

    ruleset: RuleSetProtocol
    scope_policy: ScopePolicy = ScopePolicy.NAMED
    track_sources: bool = False
    wrap_factory_errors: bool = False
    finalization_errors: bool = False
    profile: Optional[str] = None
    compile_policy: CompilePolicy = CompilePolicy.ALLOW_OVERRIDE
    policy: Optional["ResolutionPolicy"] = None

ContextValueMissingError

Bases: Exception

Raised when a from_context key is absent in the active scope.

Examples:

>>> from doppy_di import Container, Scope
>>> from doppy_di.providers import from_context
>>> services = Container()
>>> services.user = from_context("user")
>>> with services.scope("req") as s:
...     try:
...         s.get("user")
...     except ContextValueMissingError as exc:
...         str(exc)
"Context value 'user' missing in 'request' scope"

Source code in src/doppy_di/container.py

class ContextValueMissingError(Exception):
    """Raised when a ``from_context`` key is absent in the active scope.

    Examples:
        >>> from doppy_di import Container, Scope
        >>> from doppy_di.providers import from_context
        >>> services = Container()
        >>> services.user = from_context("user")
        >>> with services.scope("req") as s:
        ...     try:
        ...         s.get("user")
        ...     except ContextValueMissingError as exc:
        ...         str(exc)
        "Context value 'user' missing in 'request' scope"
    """

    def __init__(self, key: Key, scope: str) -> None:
        self.key = key
        self.scope = scope
        super().__init__(f"Context value {key!r} missing in {scope!r} scope")

CycleError

Bases: Exception

Raised when the rule graph contains a dependency cycle.

Examples:

>>> raise CycleError(["a", "b", "a"])
Traceback (most recent call last):
...
CycleError: Cycle detected: 'a' -> 'b' -> 'a'

Source code in src/doppy_di/container.py

class CycleError(Exception):
    """Raised when the rule graph contains a dependency cycle.

    Examples:
        >>> raise CycleError(["a", "b", "a"])
        Traceback (most recent call last):
        ...
        CycleError: Cycle detected: 'a' -> 'b' -> 'a'
    """

    def __init__(self, path: List[Key]) -> None:
        self.path = tuple(path)
        super().__init__(f"Cycle detected: {' -> '.join(map(repr, path))}")

CyclicDependencyError

Bases: ValidationError

Raised when the rule graph contains a dependency cycle.

Examples:

>>> raise CyclicDependencyError(["a", "b", "a"])
Traceback (most recent call last):
...
CyclicDependencyError: Cycle detected: 'a' -> 'b' -> 'a'

Source code in src/doppy_di/container.py

class CyclicDependencyError(ValidationError):
    """Raised when the rule graph contains a dependency cycle.

    Examples:
        >>> raise CyclicDependencyError(["a", "b", "a"])
        Traceback (most recent call last):
        ...
        CyclicDependencyError: Cycle detected: 'a' -> 'b' -> 'a'
    """

    def __init__(self, path: List[Key]) -> None:
        self.path = tuple(path)
        super().__init__(f"Cycle detected: {' -> '.join(map(repr, path))}")

DefaultResolutionPolicy dataclass

Default resolution order: resolve only the requested key.

Matches the historical container behaviour exactly. Dependencies are resolved recursively on demand; nothing is pre-resolved.

Examples:

>>> policy = DefaultResolutionPolicy()
>>> list(policy.order({}, "a"))
['a']

Source code in src/doppy_di/resolution.py

@dataclass(frozen=True)
class DefaultResolutionPolicy:
    """Default resolution order: resolve only the requested key.

    Matches the historical container behaviour exactly. Dependencies are
    resolved recursively on demand; nothing is pre-resolved.

    Examples:
        >>> policy = DefaultResolutionPolicy()
        >>> list(policy.order({}, "a"))
        ['a']
    """

    def order(
        self,
        graph: Mapping[Key, Rule],
        root: Key,
    ) -> Iterable[Key]:
        return (root,)

DependencyCycleError

Bases: CycleError

Raised when a dependency cycle is detected.

Subclasses CycleError for backward compatibility.

Examples:

>>> err = DependencyCycleError(["a", "b", "a"])
>>> "a" in err.cycle
True

Source code in src/doppy_di/container.py

class DependencyCycleError(CycleError):
    """Raised when a dependency cycle is detected.

    Subclasses ``CycleError`` for backward compatibility.

    Examples:
        >>> err = DependencyCycleError(["a", "b", "a"])
        >>> "a" in err.cycle
        True
    """

    def __init__(self, path: List[Key]) -> None:
        super().__init__(path)
        self.cycle = list(path)

    def __str__(self) -> str:
        return "Cycle detected: " + " → ".join(map(repr, self.cycle))

DependencyGraph

DependencyGraph for container introspection.

Source code in src/doppy_di/graph.py

class DependencyGraph:
    """DependencyGraph for container introspection."""

    def __init__(self, ruleset: RuleSetProtocol) -> None:
        self._ruleset = ruleset

    def nodes(self) -> Tuple[Key, ...]:
        """Return all registered keys in the graph."""
        return tuple(self._ruleset.map.keys())

    def edges(self) -> Tuple[Tuple[Key, Key], ...]:
        """Return all directed edges (key, dependency).

        Only returns edges pointing to registered dependencies.
        """
        res: List[Tuple[Key, Key]] = []
        for key, rule in self._ruleset.map.items():
            for dep in rule.deps:
                if dep in self._ruleset.map:
                    res.append((key, dep))
        return tuple(res)

    def dependencies_of(self, key: Key) -> Tuple[Key, ...]:
        """Return registered direct dependencies of key."""
        if key not in self._ruleset.map:
            raise ServiceNotFoundError(key)
        rule = self._ruleset.map[key]
        return tuple(d for d in rule.deps if d in self._ruleset.map)

    def dependents_of(self, key: Key) -> Tuple[Key, ...]:
        """Return registered direct dependents of key."""
        if key not in self._ruleset.map:
            raise ServiceNotFoundError(key)
        res: List[Key] = []
        for k, rule in self._ruleset.map.items():
            if key in rule.deps:
                res.append(k)
        return tuple(res)

    def to_mermaid(self) -> str:
        """Render graph as mermaid graph."""
        return render_mermaid(self._ruleset)

    def to_dot(self) -> str:
        """Render graph as graphviz dot string."""
        return render_graphviz(self._ruleset)

    def to_json(self) -> Dict[str, Any]:
        """Render graph as JSON dictionary."""
        return render_json(self._ruleset)

    def to_text(self) -> str:
        """Render graph as text tree."""
        lines: List[str] = []
        visited: Set[Key] = set()

        def draw(node: Key, prefix: str = "", is_last: bool = True) -> None:
            visited.add(node)
            lines.append(f"{prefix}{'└── ' if is_last else '├── '}{node!r}")
            deps = self.dependencies_of(node)
            for i, dep in enumerate(deps):
                new_prefix = prefix + ("    " if is_last else "│   ")
                if dep not in visited:
                    draw(dep, new_prefix, i == len(deps) - 1)
                else:
                    is_last_dep = i == len(deps) - 1
                    marker = "└── " if is_last_dep else "├── "
                    lines.append(f"{new_prefix}{marker}{dep!r} (cycle)")

        roots = [k for k in self.nodes() if not self.dependents_of(k)]
        if not roots:
            roots = list(self.nodes())

        for root in roots:
            if root not in visited:
                lines.append(f"{root!r}")
                deps = self.dependencies_of(root)
                for i, dep in enumerate(deps):
                    draw(dep, "", i == len(deps) - 1)
        return "\n".join(lines)

dependencies_of(key)

Return registered direct dependencies of key.

Source code in src/doppy_di/graph.py

def dependencies_of(self, key: Key) -> Tuple[Key, ...]:
    """Return registered direct dependencies of key."""
    if key not in self._ruleset.map:
        raise ServiceNotFoundError(key)
    rule = self._ruleset.map[key]
    return tuple(d for d in rule.deps if d in self._ruleset.map)

dependents_of(key)

Return registered direct dependents of key.

Source code in src/doppy_di/graph.py

def dependents_of(self, key: Key) -> Tuple[Key, ...]:
    """Return registered direct dependents of key."""
    if key not in self._ruleset.map:
        raise ServiceNotFoundError(key)
    res: List[Key] = []
    for k, rule in self._ruleset.map.items():
        if key in rule.deps:
            res.append(k)
    return tuple(res)

edges()

Return all directed edges (key, dependency).

Only returns edges pointing to registered dependencies.

Source code in src/doppy_di/graph.py

def edges(self) -> Tuple[Tuple[Key, Key], ...]:
    """Return all directed edges (key, dependency).

    Only returns edges pointing to registered dependencies.
    """
    res: List[Tuple[Key, Key]] = []
    for key, rule in self._ruleset.map.items():
        for dep in rule.deps:
            if dep in self._ruleset.map:
                res.append((key, dep))
    return tuple(res)

nodes()

Return all registered keys in the graph.

Source code in src/doppy_di/graph.py

def nodes(self) -> Tuple[Key, ...]:
    """Return all registered keys in the graph."""
    return tuple(self._ruleset.map.keys())

to_dot()

Render graph as graphviz dot string.

Source code in src/doppy_di/graph.py

def to_dot(self) -> str:
    """Render graph as graphviz dot string."""
    return render_graphviz(self._ruleset)

to_json()

Render graph as JSON dictionary.

Source code in src/doppy_di/graph.py

def to_json(self) -> Dict[str, Any]:
    """Render graph as JSON dictionary."""
    return render_json(self._ruleset)

to_mermaid()

Render graph as mermaid graph.

Source code in src/doppy_di/graph.py

def to_mermaid(self) -> str:
    """Render graph as mermaid graph."""
    return render_mermaid(self._ruleset)

to_text()

Render graph as text tree.

Source code in src/doppy_di/graph.py

def to_text(self) -> str:
    """Render graph as text tree."""
    lines: List[str] = []
    visited: Set[Key] = set()

    def draw(node: Key, prefix: str = "", is_last: bool = True) -> None:
        visited.add(node)
        lines.append(f"{prefix}{'└── ' if is_last else '├── '}{node!r}")
        deps = self.dependencies_of(node)
        for i, dep in enumerate(deps):
            new_prefix = prefix + ("    " if is_last else "│   ")
            if dep not in visited:
                draw(dep, new_prefix, i == len(deps) - 1)
            else:
                is_last_dep = i == len(deps) - 1
                marker = "└── " if is_last_dep else "├── "
                lines.append(f"{new_prefix}{marker}{dep!r} (cycle)")

    roots = [k for k in self.nodes() if not self.dependents_of(k)]
    if not roots:
        roots = list(self.nodes())

    for root in roots:
        if root not in visited:
            lines.append(f"{root!r}")
            deps = self.dependencies_of(root)
            for i, dep in enumerate(deps):
                draw(dep, "", i == len(deps) - 1)
    return "\n".join(lines)

DiffReport dataclass

Differences between two containers' effective rule sets.

Attributes:

Name	Type	Description
added	Tuple[Key, ...]	Keys present in other but not in self.
removed	Tuple[Key, ...]	Keys present in self but not in other.
changed	Tuple[Key, ...]	Keys present in both with different rules.

Examples:

>>> report = DiffReport(added=("c",), changed=("a",))
>>> "a" in report
True
>>> "c" in report
True
>>> "b" in report
False

Source code in src/doppy_di/container.py

@dataclass(frozen=True)
class DiffReport:
    """Differences between two containers' effective rule sets.

    Attributes:
        added: Keys present in ``other`` but not in ``self``.
        removed: Keys present in ``self`` but not in ``other``.
        changed: Keys present in both with different rules.

    Examples:
        >>> report = DiffReport(added=("c",), changed=("a",))
        >>> "a" in report
        True
        >>> "c" in report
        True
        >>> "b" in report
        False
    """

    added: Tuple[Key, ...] = ()
    removed: Tuple[Key, ...] = ()
    changed: Tuple[Key, ...] = ()

    def __contains__(self, key: object) -> bool:
        return key in self.added or key in self.removed or key in self.changed

    def __str__(self) -> str:
        lines: List[str] = []
        for key in self.added:
            lines.append(f"+ {key!r}")
        for key in self.removed:
            lines.append(f"- {key!r}")
        for key in self.changed:
            lines.append(f"~ {key!r}")
        return "\n".join(lines)

DuplicateKeyError

Bases: KeyError

Raised when a duplicate key is registered under the FAIL policy.

Examples:

>>> raise DuplicateKeyError("x")
Traceback (most recent call last):
...
DuplicateKeyError: Duplicate key: 'x'

Source code in src/doppy_di/container.py

class DuplicateKeyError(KeyError):
    """Raised when a duplicate key is registered under the FAIL policy.

    Examples:
        >>> raise DuplicateKeyError("x")
        Traceback (most recent call last):
        ...
        DuplicateKeyError: Duplicate key: 'x'
    """

    def __init__(self, key: Key) -> None:
        self.key = key
        super().__init__(f"Duplicate key: {key!r}")

DuplicateKeyPolicy

Bases: Enum

Strategy for handling duplicate key registration.

OVERWRITE: replace existing rule (current default behavior). FAIL: raise DuplicateKeyError on duplicate. WARN: log a warning and overwrite.

Examples:

>>> DuplicateKeyPolicy.FAIL.name
'FAIL'
>>> DuplicateKeyPolicy.OVERWRITE.value
'overwrite'

Source code in src/doppy_di/container.py

class DuplicateKeyPolicy(Enum):
    """Strategy for handling duplicate key registration.

    OVERWRITE: replace existing rule (current default behavior).
    FAIL: raise DuplicateKeyError on duplicate.
    WARN: log a warning and overwrite.

    Examples:
        >>> DuplicateKeyPolicy.FAIL.name
        'FAIL'
        >>> DuplicateKeyPolicy.OVERWRITE.value
        'overwrite'
    """

    OVERWRITE = "overwrite"
    FAIL = "fail"
    WARN = "warn"

DuplicateRegistrationError

Bases: DuplicateKeyError

Raised when a duplicate key is registered with source information.

Subclasses DuplicateKeyError for backward compatibility.

Source code in src/doppy_di/container.py

class DuplicateRegistrationError(DuplicateKeyError):
    """Raised when a duplicate key is registered with source information.

    Subclasses ``DuplicateKeyError`` for backward compatibility.
    """

    def __init__(
        self,
        key: Key,
        existing_source: Optional[RegistrationSource] = None,
        new_source: Optional[RegistrationSource] = None,
    ) -> None:
        self.key = key
        self.existing_source = existing_source
        self.new_source = new_source
        super().__init__(key)

    def __str__(self) -> str:
        parts = [f"Duplicate registration for {self.key!r}:"]
        if self.existing_source is not None:
            parts.append(f"  existing: {self.existing_source}")
        if self.new_source is not None:
            parts.append(f"  new: {self.new_source}")
        return "\n".join(parts)

EagerPolicy dataclass

Resolve the entire graph eagerly.

All registered keys are resolved up front. When used with build(), every sync rule is resolved at build time. Async rules cannot be resolved in a sync context and raise :class:AsyncDependencyInSyncContextError.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("a", 1)
>>> container = builder.build(policy=EagerPolicy())
>>> container.get("a")
1

Source code in src/doppy_di/resolution.py

@dataclass(frozen=True)
class EagerPolicy:
    """Resolve the entire graph eagerly.

    All registered keys are resolved up front. When used with ``build()``,
    every sync rule is resolved at build time. Async rules cannot be resolved
    in a sync context and raise :class:`AsyncDependencyInSyncContextError`.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("a", 1)
        >>> container = builder.build(policy=EagerPolicy())
        >>> container.get("a")
        1
    """

    def order(
        self,
        graph: Mapping[Key, Rule],
        root: Key,
    ) -> Iterable[Key]:
        return _topological(graph, graph.keys())

ExecutionPlan dataclass

Immutable, pre-compiled execution plan for a container.

The plan holds a topological ordering of the registered rules plus the resolved dependency edges. get walks the precomputed order and resolves directly without re-entering Container.get, so lifetimes, caches and scopes keep the same semantics as Container.get.

The plan is immutable: once built it cannot be changed. It may be serialized for caching or cross-process reuse via :meth:serialize.

Source code in src/doppy_di/plan.py

@dataclass(frozen=True, slots=True)
class ExecutionPlan:
    """Immutable, pre-compiled execution plan for a container.

    The plan holds a topological ordering of the registered rules plus the
    resolved dependency edges. ``get`` walks the precomputed order and
    resolves directly without re-entering ``Container.get``, so lifetimes,
    caches and scopes keep the same semantics as ``Container.get``.

    The plan is immutable: once built it cannot be changed. It may be
    serialized for caching or cross-process reuse via :meth:`serialize`.
    """

    container: Optional[Container]
    order: Tuple[str, ...]
    edges: Dict[str, Tuple[str, ...]]
    rules: Dict[str, Dict[str, Any]]
    keys: Dict[str, Key]
    singletons: Dict[str, Any] = field(default_factory=dict)
    compile_policy: str = "allow_override"
    node_index: Dict[Key, int] = field(default_factory=dict)
    nodes: Tuple[_NodeSpec, ...] = field(default_factory=tuple)
    resolvers: Dict[Key, Callable[[], Any]] = field(default_factory=dict)
    resolver_kinds: Dict[Key, str] = field(default_factory=dict)
    _frozen: Dict[Key, Any] = field(default_factory=dict)
    frozen: bool = False
    guardless: bool = False

    def bind(self, key: Key, qualifier: Optional[str] = None) -> BoundResolver:
        """Return a bound resolver for ``key`` without repeated root lookup.

        The returned callable validates the key at bind time and preserves
        the same semantics as :meth:`get`. It exposes the selected
        execution mode through :attr:`BoundResolver.kind`.
        """
        lookup = (key, qualifier) if qualifier is not None else key
        known = lookup in self.node_index or lookup in self.resolver_kinds
        if self.nodes and not known:
            if self.container is not None and not self.container.has(key, qualifier):
                raise ServiceNotFoundError(key)
        elif self.container is None and not known and _key_repr(lookup) not in self.singletons:
            raise ServiceNotFoundError(key)
        direct: Optional[Callable[[], Any]] = None
        needs_guard = False
        container = self.container
        resolver = self.resolvers.get(lookup)
        if resolver is not None:
            if self.guardless:
                direct = resolver
                needs_guard = False
            elif container is not None and (
                self.frozen or (not container._override_layers and container._tracer is None)
            ):
                direct = resolver
                needs_guard = not self.frozen
        return BoundResolver(
            plan=self,
            key=key,
            qualifier=qualifier,
            _direct=direct,
            _needs_guard=needs_guard,
        )

    def _resolve_fast(self, lookup: Key) -> Any:
        """Resolve ``lookup`` using the precomputed node graph."""
        container = self.container
        idx = self.node_index.get(lookup)
        if idx is None:
            if container is not None:
                return container.get(lookup)
            idx = self.node_index.get(_key_repr(lookup))
            if idx is None:
                raise ServiceNotFoundError(lookup)

        nodes = self.nodes
        if container is None:
            resolved = [None] * (idx + 1)
            for i in range(idx + 1):
                spec = nodes[i]
                if spec.lifetime == "singleton":
                    cached = self.singletons.get(self.order[i], _MISSING)
                    if cached is _MISSING:
                        cached = self.singletons.get(_key_repr(spec.key), _MISSING)
                    if cached is _MISSING and isinstance(spec.key, str):
                        cached = self.singletons.get(spec.key, _MISSING)
                    if cached is not _MISSING:
                        resolved[i] = cached
                        continue
                make = spec.make
                if make is None:
                    raise ServiceNotFoundError(spec.key)
                deps = spec.deps_idx
                dep_objs = make(*[resolved[j] for j in deps]) if deps else make()
                resolved[i] = dep_objs
            return resolved[idx]

        single = container.single
        lock = container.lock
        override_layers = container._override_layers
        tracer = container._tracer
        started = tracer is not None
        start = 0.0
        if started:
            start = time.perf_counter()

        resolved = [None] * (idx + 1)
        for i in range(idx + 1):
            spec = nodes[i]
            if self.frozen:
                if spec.lifetime == "singleton":
                    resolved[i] = self._frozen[spec.key]
                    continue
            else:
                if override_layers:
                    overridden = container._resolve_override(spec.key)
                    if overridden is not _unset:
                        resolved[i] = overridden
                        continue
                if spec.lifetime == "singleton":
                    cached = single.get(spec.key, _MISSING)
                    if cached is not _MISSING:
                        resolved[i] = cached
                        continue

            if spec.yield_provider or spec.async_yield_provider or spec.is_async:
                return container.get(spec.key)

            make = spec.make
            if make is None:
                return container.get(spec.key)

            deps = spec.deps_idx
            obj = make(*[resolved[j] for j in deps]) if deps else make()

            if not self.frozen and spec.lifetime == "singleton":
                with lock:
                    existing = single.get(spec.key, _MISSING)
                    if existing is _MISSING:
                        single[spec.key] = obj
                    else:
                        obj = existing
            if spec.nested:
                container._cache_nested_aliases(spec.key, obj)
            if started:
                container._trace(spec.key, time.perf_counter() - start, False, None)
            resolved[i] = obj

        return resolved[idx]

    def get(self, key: Key, qualifier: Optional[str] = None) -> Any:
        """Resolve ``key`` using the precomputed order."""
        lookup = (key, qualifier) if qualifier is not None else key
        if self.nodes:
            if self.guardless:
                resolver = self.resolvers.get(lookup)
                if resolver is not None:
                    return resolver()
                return self._resolve_fast(lookup)
            container = self.container
            if container is not None and (
                self.frozen or (not container._override_layers and container._tracer is None)
            ):
                resolvers = self.resolvers
                try:
                    resolver = resolvers[lookup]
                except KeyError:
                    return self._resolve_fast(lookup)
                return resolver()
            return self._resolve_fast(lookup)
        container = self.container
        if container is not None:
            return container.get(lookup)
        lookup_repr = _key_repr(lookup)
        if lookup_repr in self.singletons:
            return self.singletons[lookup_repr]
        raise ServiceNotFoundError(lookup)

    def aget(self, key: Key, qualifier: Optional[str] = None) -> Any:
        """Async resolution using the precomputed order."""
        lookup = (key, qualifier) if qualifier is not None else key
        container = self.container
        if container is None:
            raise ServiceNotFoundError(lookup)
        return container.aget(lookup)

    @classmethod
    def from_container(
        cls,
        container: Container,
        copy_parent_rules: bool = True,
        allow_post_compile_overrides: bool = True,
        guardless: bool = False,
    ) -> "ExecutionPlan":
        """Build an :class:`ExecutionPlan` from a container."""
        if not allow_post_compile_overrides and container._override_layers:
            raise RuntimeError(
                "Cannot compile with allow_post_compile_overrides=False "
                "while an override layer is active"
            )
        ruleset = container.config.ruleset

        from .providers import implicit_collection_rule

        for _key, rule in list(ruleset.map.items()):
            for dep in rule.deps:
                if dep not in ruleset.map:
                    collection = implicit_collection_rule(dep, ruleset)
                    if collection is not None:
                        ruleset.add(dep, collection)

        errors: List[Tuple[Key, Key]] = []
        for key, rule in ruleset.map.items():
            for dep in rule.deps:
                if dep not in ruleset.map:
                    errors.append((key, dep))
        if errors:
            raise MissingDependencyError(
                errors[0][0],
                resolution_path=[errors[0][0], errors[0][1]],
            ) from None

        for key, rule in ruleset.map.items():
            try:
                sig = inspect.signature(rule.make)
            except (TypeError, ValueError):
                sig = None
            if sig is not None:
                positional = [
                    p
                    for p in sig.parameters.values()
                    if p.kind
                    in (
                        inspect.Parameter.POSITIONAL_ONLY,
                        inspect.Parameter.POSITIONAL_OR_KEYWORD,
                    )
                ]
                required = sum(1 for p in positional if p.default is inspect.Parameter.empty)
                total = len(positional)
                has_varargs = any(
                    p.kind == inspect.Parameter.VAR_POSITIONAL for p in sig.parameters.values()
                )
                if len(rule.deps) < required:
                    raise InvalidFactoryError(
                        key,
                        f"factory requires at least {required} args "
                        f"but only {len(rule.deps)} deps declared",
                    ) from None
                if len(rule.deps) > total and not has_varargs:
                    raise InvalidFactoryError(
                        key,
                        f"factory accepts at most {total} args but {len(rule.deps)} deps declared",
                    ) from None

        for key in ruleset.map:
            try:
                ruleset._check_cycle(key)
            except DependencyCycleError:
                raise
            except Exception as exc:  # pragma: no cover - defensive
                raise DependencyCycleError([key]) from exc

        if copy_parent_rules and isinstance(ruleset, CompositeRuleSet):
            rules_map: Dict[Key, Rule] = dict(ruleset.map)
        else:
            rules_map = ruleset.map

        order, edges = _topological_order(ruleset, rules_map)
        meta: Dict[str, Dict[str, Any]] = {}
        keys: Dict[str, Key] = {}
        for key in rules_map:
            repr_key = _key_repr(key)
            meta[repr_key] = _rule_meta(rules_map[key])
            keys[repr_key] = key

        key_to_idx: Dict[Key, int] = {}
        for i, repr_key in enumerate(order):
            key_to_idx[keys[repr_key]] = i

        nodes: List[_NodeSpec] = []
        for repr_key in order:
            key = keys[repr_key]
            rule = rules_map[key]
            deps_idx = tuple(key_to_idx[d] for d in rule.deps if d in key_to_idx)
            nodes.append(
                _NodeSpec(
                    key=key,
                    make=rule.make,
                    deps_idx=deps_idx,
                    lifetime=rule.lifetime,
                    yield_provider=rule.yield_provider,
                    async_yield_provider=rule.async_yield_provider,
                    is_async=rule.is_async,
                    nested=rule.nested,
                )
            )

        frozen: Optional[Dict[Key, Any]] = None
        if not allow_post_compile_overrides or guardless:
            frozen = {}
            for _i, spec in enumerate(nodes):
                if spec.lifetime != "singleton":
                    continue
                if spec.make is None:
                    continue
                deps = spec.deps_idx
                args = [frozen[nodes[j].key] for j in deps]
                frozen[spec.key] = spec.make(*args) if args else spec.make()
            container.single.update(frozen)
            object.__setattr__(container, "_compiled_plan", None)

        makers: List[Optional[Callable[[], Any]]] = [None] * len(nodes)
        resolvers: Dict[Key, Callable[[], Any]] = {}
        for i, spec in enumerate(nodes):
            if (
                spec.make is None
                or spec.yield_provider
                or spec.async_yield_provider
                or spec.is_async
                or spec.nested
            ):
                continue
            if not all(makers[j] is not None for j in spec.deps_idx):
                continue
            dep_makers = tuple(cast(Callable[[], Any], makers[j]) for j in spec.deps_idx)
            maker = _build_node_maker(spec, dep_makers, container, frozen)
            makers[i] = maker
            resolvers[spec.key] = maker

        nodes_tuple = tuple(nodes)
        resolver_kinds: Dict[Key, str] = dict.fromkeys(resolvers, "composed")
        for i, spec in enumerate(nodes):
            if spec.key not in resolvers:
                continue
            if frozen is not None and spec.lifetime == "singleton":
                resolver_kinds[spec.key] = "frozen"
                continue
            if frozen is not None:
                fr = _build_frozen_resolver(i, nodes_tuple, frozen)

                if fr is not None:
                    kind, fn = fr
                    resolvers[spec.key] = fn
                    resolver_kinds[spec.key] = kind
                    continue
            flat = _build_flat_resolver(i, nodes_tuple, makers, container, frozen)
            if flat is None:
                continue
            kind, fn = flat
            resolvers[spec.key] = fn
            resolver_kinds[spec.key] = kind

        policy = container.config.compile_policy.value
        return cls(
            container=container,
            order=tuple(order),
            edges=edges,
            rules=meta,
            keys=keys,
            singletons={},
            compile_policy=policy,
            node_index=key_to_idx,
            nodes=nodes_tuple,
            resolvers=resolvers,
            resolver_kinds=resolver_kinds,
            _frozen=frozen or {},
            frozen=frozen is not None,
            guardless=guardless,
        )

    def _singleton_snapshot(self) -> Dict[str, Any]:
        """Capture resolved singletons (+ unresolved singleton constants)."""
        container = self.container
        if container is None:
            return dict(self.singletons)
        snapshot: Dict[str, Any] = {_key_repr(k): v for k, v in container.single.items()}
        if self.frozen:
            snapshot.update({_key_repr(k): v for k, v in self._frozen.items()})
        for repr_key, meta in self.rules.items():
            if meta.get("lifetime") != "singleton":
                continue
            if repr_key in snapshot:
                continue
            key = self.keys.get(repr_key)
            if key is None:
                continue
            try:
                snapshot[repr_key] = container.get(key)
            except Exception as exc:
                logger.debug("singleton %r not resolvable for snapshot: %s", repr_key, exc)
        return snapshot

    def serialize(self, format: str = "json") -> str:  # noqa: A002
        """Serialize the plan to a string for caching or cross-process use."""
        if format != "json":
            raise ValueError(f"Unsupported serialize format: {format!r}")
        payload = {
            "order": list(self.order),
            "edges": self.edges,
            "rules": self.rules,
            "keys": {rk: _key_to_serializable(k) for rk, k in self.keys.items()},
            "singletons": {
                rk: _value_to_serializable(v) for rk, v in self._singleton_snapshot().items()
            },
            "policy": self.compile_policy,
            "frozen": self.frozen,
        }
        return json.dumps(payload, sort_keys=True, indent=2)

    @classmethod
    def deserialize(cls, data: str) -> "ExecutionPlan":
        """Rebuild an :class:`ExecutionPlan` from serialized data."""
        payload = json.loads(data)
        keys: Dict[str, Key] = {
            rk: cast(Key, _key_from_serializable(v)) for rk, v in payload["keys"].items()
        }
        singletons: Dict[str, Any] = {
            rk: _value_from_serializable(v) for rk, v in payload.get("singletons", {}).items()
        }
        order = tuple(payload["order"])
        repr_to_idx = {rk: i for i, rk in enumerate(order)}

        key_to_idx: Dict[Key, int] = {}
        for i, rk in enumerate(order):
            if rk not in keys:
                continue
            key = keys[rk]
            key_to_idx[key] = i
            key_to_idx[rk] = i

        nodes: List[_NodeSpec] = []
        for rk in order:
            meta = payload.get("rules", {}).get(rk)
            if meta is None or rk not in keys:
                continue
            key = keys[rk]
            deps: List[Any] = meta.get("deps", [])
            deps_idx = tuple(repr_to_idx[d] for d in deps if d in repr_to_idx)
            nodes.append(
                _NodeSpec(
                    key=key,
                    make=None,
                    deps_idx=deps_idx,
                    lifetime=str(meta.get("lifetime", "transient")),
                    yield_provider=bool(meta.get("yield")),
                    async_yield_provider=bool(meta.get("async")),
                    is_async=bool(meta.get("is_async")),
                    nested=bool(meta.get("nested")),
                )
            )

        frozen = bool(payload.get("frozen", False))
        frozen_map: Dict[Key, Any] = {}
        if frozen:
            for rk, v in payload.get("singletons", {}).items():
                if rk in keys:
                    frozen_map[keys[rk]] = _value_from_serializable(v)
        return cls(
            container=None,
            order=order,
            edges=payload["edges"],
            rules=payload["rules"],
            keys=keys,
            singletons=singletons,
            compile_policy=payload.get("policy", "allow_override"),
            node_index=key_to_idx,
            nodes=tuple(nodes),
            _frozen=frozen_map,
            frozen=frozen,
        )

aget(key, qualifier=None)

Async resolution using the precomputed order.

Source code in src/doppy_di/plan.py

def aget(self, key: Key, qualifier: Optional[str] = None) -> Any:
    """Async resolution using the precomputed order."""
    lookup = (key, qualifier) if qualifier is not None else key
    container = self.container
    if container is None:
        raise ServiceNotFoundError(lookup)
    return container.aget(lookup)

bind(key, qualifier=None)

Return a bound resolver for key without repeated root lookup.

The returned callable validates the key at bind time and preserves the same semantics as :meth:get. It exposes the selected execution mode through :attr:BoundResolver.kind.

Source code in src/doppy_di/plan.py

def bind(self, key: Key, qualifier: Optional[str] = None) -> BoundResolver:
    """Return a bound resolver for ``key`` without repeated root lookup.

    The returned callable validates the key at bind time and preserves
    the same semantics as :meth:`get`. It exposes the selected
    execution mode through :attr:`BoundResolver.kind`.
    """
    lookup = (key, qualifier) if qualifier is not None else key
    known = lookup in self.node_index or lookup in self.resolver_kinds
    if self.nodes and not known:
        if self.container is not None and not self.container.has(key, qualifier):
            raise ServiceNotFoundError(key)
    elif self.container is None and not known and _key_repr(lookup) not in self.singletons:
        raise ServiceNotFoundError(key)
    direct: Optional[Callable[[], Any]] = None
    needs_guard = False
    container = self.container
    resolver = self.resolvers.get(lookup)
    if resolver is not None:
        if self.guardless:
            direct = resolver
            needs_guard = False
        elif container is not None and (
            self.frozen or (not container._override_layers and container._tracer is None)
        ):
            direct = resolver
            needs_guard = not self.frozen
    return BoundResolver(
        plan=self,
        key=key,
        qualifier=qualifier,
        _direct=direct,
        _needs_guard=needs_guard,
    )

deserialize(data) classmethod

Rebuild an :class:ExecutionPlan from serialized data.

Source code in src/doppy_di/plan.py

@classmethod
def deserialize(cls, data: str) -> "ExecutionPlan":
    """Rebuild an :class:`ExecutionPlan` from serialized data."""
    payload = json.loads(data)
    keys: Dict[str, Key] = {
        rk: cast(Key, _key_from_serializable(v)) for rk, v in payload["keys"].items()
    }
    singletons: Dict[str, Any] = {
        rk: _value_from_serializable(v) for rk, v in payload.get("singletons", {}).items()
    }
    order = tuple(payload["order"])
    repr_to_idx = {rk: i for i, rk in enumerate(order)}

    key_to_idx: Dict[Key, int] = {}
    for i, rk in enumerate(order):
        if rk not in keys:
            continue
        key = keys[rk]
        key_to_idx[key] = i
        key_to_idx[rk] = i

    nodes: List[_NodeSpec] = []
    for rk in order:
        meta = payload.get("rules", {}).get(rk)
        if meta is None or rk not in keys:
            continue
        key = keys[rk]
        deps: List[Any] = meta.get("deps", [])
        deps_idx = tuple(repr_to_idx[d] for d in deps if d in repr_to_idx)
        nodes.append(
            _NodeSpec(
                key=key,
                make=None,
                deps_idx=deps_idx,
                lifetime=str(meta.get("lifetime", "transient")),
                yield_provider=bool(meta.get("yield")),
                async_yield_provider=bool(meta.get("async")),
                is_async=bool(meta.get("is_async")),
                nested=bool(meta.get("nested")),
            )
        )

    frozen = bool(payload.get("frozen", False))
    frozen_map: Dict[Key, Any] = {}
    if frozen:
        for rk, v in payload.get("singletons", {}).items():
            if rk in keys:
                frozen_map[keys[rk]] = _value_from_serializable(v)
    return cls(
        container=None,
        order=order,
        edges=payload["edges"],
        rules=payload["rules"],
        keys=keys,
        singletons=singletons,
        compile_policy=payload.get("policy", "allow_override"),
        node_index=key_to_idx,
        nodes=tuple(nodes),
        _frozen=frozen_map,
        frozen=frozen,
    )

from_container(container, copy_parent_rules=True, allow_post_compile_overrides=True, guardless=False) classmethod

Build an :class:ExecutionPlan from a container.

Source code in src/doppy_di/plan.py

@classmethod
def from_container(
    cls,
    container: Container,
    copy_parent_rules: bool = True,
    allow_post_compile_overrides: bool = True,
    guardless: bool = False,
) -> "ExecutionPlan":
    """Build an :class:`ExecutionPlan` from a container."""
    if not allow_post_compile_overrides and container._override_layers:
        raise RuntimeError(
            "Cannot compile with allow_post_compile_overrides=False "
            "while an override layer is active"
        )
    ruleset = container.config.ruleset

    from .providers import implicit_collection_rule

    for _key, rule in list(ruleset.map.items()):
        for dep in rule.deps:
            if dep not in ruleset.map:
                collection = implicit_collection_rule(dep, ruleset)
                if collection is not None:
                    ruleset.add(dep, collection)

    errors: List[Tuple[Key, Key]] = []
    for key, rule in ruleset.map.items():
        for dep in rule.deps:
            if dep not in ruleset.map:
                errors.append((key, dep))
    if errors:
        raise MissingDependencyError(
            errors[0][0],
            resolution_path=[errors[0][0], errors[0][1]],
        ) from None

    for key, rule in ruleset.map.items():
        try:
            sig = inspect.signature(rule.make)
        except (TypeError, ValueError):
            sig = None
        if sig is not None:
            positional = [
                p
                for p in sig.parameters.values()
                if p.kind
                in (
                    inspect.Parameter.POSITIONAL_ONLY,
                    inspect.Parameter.POSITIONAL_OR_KEYWORD,
                )
            ]
            required = sum(1 for p in positional if p.default is inspect.Parameter.empty)
            total = len(positional)
            has_varargs = any(
                p.kind == inspect.Parameter.VAR_POSITIONAL for p in sig.parameters.values()
            )
            if len(rule.deps) < required:
                raise InvalidFactoryError(
                    key,
                    f"factory requires at least {required} args "
                    f"but only {len(rule.deps)} deps declared",
                ) from None
            if len(rule.deps) > total and not has_varargs:
                raise InvalidFactoryError(
                    key,
                    f"factory accepts at most {total} args but {len(rule.deps)} deps declared",
                ) from None

    for key in ruleset.map:
        try:
            ruleset._check_cycle(key)
        except DependencyCycleError:
            raise
        except Exception as exc:  # pragma: no cover - defensive
            raise DependencyCycleError([key]) from exc

    if copy_parent_rules and isinstance(ruleset, CompositeRuleSet):
        rules_map: Dict[Key, Rule] = dict(ruleset.map)
    else:
        rules_map = ruleset.map

    order, edges = _topological_order(ruleset, rules_map)
    meta: Dict[str, Dict[str, Any]] = {}
    keys: Dict[str, Key] = {}
    for key in rules_map:
        repr_key = _key_repr(key)
        meta[repr_key] = _rule_meta(rules_map[key])
        keys[repr_key] = key

    key_to_idx: Dict[Key, int] = {}
    for i, repr_key in enumerate(order):
        key_to_idx[keys[repr_key]] = i

    nodes: List[_NodeSpec] = []
    for repr_key in order:
        key = keys[repr_key]
        rule = rules_map[key]
        deps_idx = tuple(key_to_idx[d] for d in rule.deps if d in key_to_idx)
        nodes.append(
            _NodeSpec(
                key=key,
                make=rule.make,
                deps_idx=deps_idx,
                lifetime=rule.lifetime,
                yield_provider=rule.yield_provider,
                async_yield_provider=rule.async_yield_provider,
                is_async=rule.is_async,
                nested=rule.nested,
            )
        )

    frozen: Optional[Dict[Key, Any]] = None
    if not allow_post_compile_overrides or guardless:
        frozen = {}
        for _i, spec in enumerate(nodes):
            if spec.lifetime != "singleton":
                continue
            if spec.make is None:
                continue
            deps = spec.deps_idx
            args = [frozen[nodes[j].key] for j in deps]
            frozen[spec.key] = spec.make(*args) if args else spec.make()
        container.single.update(frozen)
        object.__setattr__(container, "_compiled_plan", None)

    makers: List[Optional[Callable[[], Any]]] = [None] * len(nodes)
    resolvers: Dict[Key, Callable[[], Any]] = {}
    for i, spec in enumerate(nodes):
        if (
            spec.make is None
            or spec.yield_provider
            or spec.async_yield_provider
            or spec.is_async
            or spec.nested
        ):
            continue
        if not all(makers[j] is not None for j in spec.deps_idx):
            continue
        dep_makers = tuple(cast(Callable[[], Any], makers[j]) for j in spec.deps_idx)
        maker = _build_node_maker(spec, dep_makers, container, frozen)
        makers[i] = maker
        resolvers[spec.key] = maker

    nodes_tuple = tuple(nodes)
    resolver_kinds: Dict[Key, str] = dict.fromkeys(resolvers, "composed")
    for i, spec in enumerate(nodes):
        if spec.key not in resolvers:
            continue
        if frozen is not None and spec.lifetime == "singleton":
            resolver_kinds[spec.key] = "frozen"
            continue
        if frozen is not None:
            fr = _build_frozen_resolver(i, nodes_tuple, frozen)

            if fr is not None:
                kind, fn = fr
                resolvers[spec.key] = fn
                resolver_kinds[spec.key] = kind
                continue
        flat = _build_flat_resolver(i, nodes_tuple, makers, container, frozen)
        if flat is None:
            continue
        kind, fn = flat
        resolvers[spec.key] = fn
        resolver_kinds[spec.key] = kind

    policy = container.config.compile_policy.value
    return cls(
        container=container,
        order=tuple(order),
        edges=edges,
        rules=meta,
        keys=keys,
        singletons={},
        compile_policy=policy,
        node_index=key_to_idx,
        nodes=nodes_tuple,
        resolvers=resolvers,
        resolver_kinds=resolver_kinds,
        _frozen=frozen or {},
        frozen=frozen is not None,
        guardless=guardless,
    )

get(key, qualifier=None)

Resolve key using the precomputed order.

Source code in src/doppy_di/plan.py

def get(self, key: Key, qualifier: Optional[str] = None) -> Any:
    """Resolve ``key`` using the precomputed order."""
    lookup = (key, qualifier) if qualifier is not None else key
    if self.nodes:
        if self.guardless:
            resolver = self.resolvers.get(lookup)
            if resolver is not None:
                return resolver()
            return self._resolve_fast(lookup)
        container = self.container
        if container is not None and (
            self.frozen or (not container._override_layers and container._tracer is None)
        ):
            resolvers = self.resolvers
            try:
                resolver = resolvers[lookup]
            except KeyError:
                return self._resolve_fast(lookup)
            return resolver()
        return self._resolve_fast(lookup)
    container = self.container
    if container is not None:
        return container.get(lookup)
    lookup_repr = _key_repr(lookup)
    if lookup_repr in self.singletons:
        return self.singletons[lookup_repr]
    raise ServiceNotFoundError(lookup)

serialize(format='json')

Serialize the plan to a string for caching or cross-process use.

Source code in src/doppy_di/plan.py

def serialize(self, format: str = "json") -> str:  # noqa: A002
    """Serialize the plan to a string for caching or cross-process use."""
    if format != "json":
        raise ValueError(f"Unsupported serialize format: {format!r}")
    payload = {
        "order": list(self.order),
        "edges": self.edges,
        "rules": self.rules,
        "keys": {rk: _key_to_serializable(k) for rk, k in self.keys.items()},
        "singletons": {
            rk: _value_to_serializable(v) for rk, v in self._singleton_snapshot().items()
        },
        "policy": self.compile_policy,
        "frozen": self.frozen,
    }
    return json.dumps(payload, sort_keys=True, indent=2)

Factory

Bases: Protocol[P, T]

A factory callable with parameter specification.

Examples:

>>> def make(host: str) -> Database:
...     return Database(host)
>>> isinstance(make, Factory)
True

Source code in src/doppy_di/container.py

@runtime_checkable
class Factory(Protocol[P, T]):
    """A factory callable with parameter specification.

    Examples:
        >>> def make(host: str) -> Database:
        ...     return Database(host)
        >>> isinstance(make, Factory)
        True
    """

    def __call__(self, *args: P.args, **kwargs: P.kwargs) -> T: ...

FactoryExecutionError

Bases: Exception

Wraps an exception raised by a factory body.

Only raised when wrap_factory_errors=True is set on the builder.

Source code in src/doppy_di/container.py

class FactoryExecutionError(Exception):
    """Wraps an exception raised by a factory body.

    Only raised when ``wrap_factory_errors=True`` is set on the builder.
    """

    def __init__(
        self,
        key: Key,
        original_exception: Exception,
        resolution_path: Optional[List[Key]] = None,
    ) -> None:
        self.key = key
        self.original_exception = original_exception
        self.resolution_path = list(resolution_path or [])
        super().__init__(f"Factory for {key!r} raised {original_exception!r}")

    def __str__(self) -> str:
        parts = [f"Factory for {self.key!r} raised:"]
        parts.append(f"  {self.original_exception!r}")
        if self.resolution_path:
            parts.append("")
            parts.append("Resolution path: " + " → ".join(map(repr, self.resolution_path)))
        return "\n".join(parts)

InvalidFactoryError

Bases: ValidationError

Raised when a factory is incompatible with its declared deps.

Examples:

>>> raise InvalidFactoryError("a", "arity mismatch")
Traceback (most recent call last):
...
InvalidFactoryError: Invalid factory for 'a': arity mismatch

Source code in src/doppy_di/container.py

class InvalidFactoryError(ValidationError):
    """Raised when a factory is incompatible with its declared deps.

    Examples:
        >>> raise InvalidFactoryError("a", "arity mismatch")
        Traceback (most recent call last):
        ...
        InvalidFactoryError: Invalid factory for 'a': arity mismatch
    """

    def __init__(self, key: Key, reason: str) -> None:
        self.key = key
        self.reason = reason
        super().__init__(f"Invalid factory for {key!r}: {reason}")

InvalidLifetimeError

Bases: ValueError

Raised when an unknown lifetime is used.

Subclasses ValueError for backward compatibility.

Examples:

>>> raise InvalidLifetimeError("per_request")
Traceback (most recent call last):
...
InvalidLifetimeError: Unknown lifetime: 'per_request'

Source code in src/doppy_di/container.py

class InvalidLifetimeError(ValueError):
    """Raised when an unknown lifetime is used.

    Subclasses ``ValueError`` for backward compatibility.

    Examples:
        >>> raise InvalidLifetimeError("per_request")
        Traceback (most recent call last):
        ...
        InvalidLifetimeError: Unknown lifetime: 'per_request'
    """

    def __init__(self, lifetime: str) -> None:
        self.lifetime = lifetime
        super().__init__(f"Unknown lifetime: {lifetime!r}")

KeyProtocol

Bases: Protocol

A protocol for custom hashable keys.

Examples:

>>> class MyKey:
...     def __hash__(self): return 1
...     def __eq__(self, other): return isinstance(other, MyKey)
>>> isinstance(MyKey(), KeyProtocol)
True

Source code in src/doppy_di/container.py

class KeyProtocol(Protocol):
    """A protocol for custom hashable keys.

    Examples:
        >>> class MyKey:
        ...     def __hash__(self): return 1
        ...     def __eq__(self, other): return isinstance(other, MyKey)
        >>> isinstance(MyKey(), KeyProtocol)
        True
    """

    def __hash__(self) -> int: ...

    def __eq__(self, other: object) -> bool: ...

LazyPolicy dataclass

Resolve only the requested key on demand.

This is the same semantics as the default behaviour: nothing is resolved until get() is called, and only the requested key plus its transitive dependencies are resolved.

Examples:

>>> policy = LazyPolicy()
>>> list(policy.order({}, "a"))
['a']

Source code in src/doppy_di/resolution.py

@dataclass(frozen=True)
class LazyPolicy:
    """Resolve only the requested key on demand.

    This is the same semantics as the default behaviour: nothing is resolved
    until ``get()`` is called, and only the requested key plus its transitive
    dependencies are resolved.

    Examples:
        >>> policy = LazyPolicy()
        >>> list(policy.order({}, "a"))
        ['a']
    """

    def order(
        self,
        graph: Mapping[Key, Rule],
        root: Key,
    ) -> Iterable[Key]:
        return (root,)

LifetimePolicy

Validation policy for service lifetimes.

Centralizes the set of known lifetime identifiers and provides an extension point for registering custom lifetimes.

Examples:

>>> LifetimePolicy.validate("singleton")
>>> LifetimePolicy.validate("transient")
>>> LifetimePolicy.validate("per_request")  # raises ValueError
Traceback (most recent call last):
...
ValueError: Unknown lifetime: 'per_request'

Source code in src/doppy_di/container.py

class LifetimePolicy:
    """Validation policy for service lifetimes.

    Centralizes the set of known lifetime identifiers and provides an
    extension point for registering custom lifetimes.

    Examples:
        >>> LifetimePolicy.validate("singleton")
        >>> LifetimePolicy.validate("transient")
        >>> LifetimePolicy.validate("per_request")  # raises ValueError
        Traceback (most recent call last):
        ...
        ValueError: Unknown lifetime: 'per_request'
    """

    known: ClassVar[set[str]] = {"transient", "singleton"}

    @classmethod
    def validate(cls, lifetime: str) -> None:
        if lifetime not in cls.known:
            raise InvalidLifetimeError(lifetime)

LoggingContainer dataclass

Log container operations while preserving the same API.

Wraps a Container and calls a log function on every operation.

Examples:

>>> events = []
>>> def log(msg):
...     events.append(msg)
>>> from doppy_di.container import ContainerBuilder
>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> base = builder.build()
>>> wrapped = LoggingContainer(base, log)
>>> wrapped.get("x")
1
>>> events
["get('x')", "get('x') -> ok"]

Source code in src/doppy_di/devkit/logging.py

@dataclass(frozen=True)
class LoggingContainer:
    """Log container operations while preserving the same API.

    Wraps a Container and calls a log function on every operation.

    Examples:
        >>> events = []
        >>> def log(msg):
        ...     events.append(msg)
        >>> from doppy_di.container import ContainerBuilder
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> base = builder.build()
        >>> wrapped = LoggingContainer(base, log)
        >>> wrapped.get("x")
        1
        >>> events
        ["get('x')", "get('x') -> ok"]
    """

    wrapped: Container
    log: Callable[[str], None]

    def get(self, key: Key) -> Any:
        """Resolve key and log the operation.

        Examples:
            >>> wrapped.get("x")
            1
        """
        self.log(f"get({key!r})")
        try:
            obj = self.wrapped.get(key)
            self.log(f"get({key!r}) -> ok")
            return obj
        except BaseException as exc:
            self.log(f"get({key!r}) -> error: {exc.__class__.__name__}")
            raise

    def has(self, key: Key) -> bool:
        """Log presence check and delegate.

        Examples:
            >>> wrapped.has("x")
            True
        """
        self.log(f"has({key!r})")
        return self.wrapped.has(key)

    def scope(self, name: str) -> Scope:
        """Log scope creation and delegate.

        Examples:
            >>> s = wrapped.scope("req")
            >>> isinstance(s, Scope)
            True
        """
        self.log(f"scope({name!r})")
        return self.wrapped.scope(name)

    def override(self, key: Key, value: Any) -> OverrideContext:
        """Log override creation and delegate.

        Examples:
            >>> ctx = wrapped.override("x", 2)
            >>> isinstance(ctx, OverrideContext)
            True
        """
        self.log(f"override({key!r})")
        return self.wrapped.override(key, value)

get(key)

Resolve key and log the operation.

Examples:

>>> wrapped.get("x")
1

Source code in src/doppy_di/devkit/logging.py

def get(self, key: Key) -> Any:
    """Resolve key and log the operation.

    Examples:
        >>> wrapped.get("x")
        1
    """
    self.log(f"get({key!r})")
    try:
        obj = self.wrapped.get(key)
        self.log(f"get({key!r}) -> ok")
        return obj
    except BaseException as exc:
        self.log(f"get({key!r}) -> error: {exc.__class__.__name__}")
        raise

has(key)

Log presence check and delegate.

Examples:

>>> wrapped.has("x")
True

Source code in src/doppy_di/devkit/logging.py

def has(self, key: Key) -> bool:
    """Log presence check and delegate.

    Examples:
        >>> wrapped.has("x")
        True
    """
    self.log(f"has({key!r})")
    return self.wrapped.has(key)

override(key, value)

Log override creation and delegate.

Examples:

>>> ctx = wrapped.override("x", 2)
>>> isinstance(ctx, OverrideContext)
True

Source code in src/doppy_di/devkit/logging.py

def override(self, key: Key, value: Any) -> OverrideContext:
    """Log override creation and delegate.

    Examples:
        >>> ctx = wrapped.override("x", 2)
        >>> isinstance(ctx, OverrideContext)
        True
    """
    self.log(f"override({key!r})")
    return self.wrapped.override(key, value)

scope(name)

Log scope creation and delegate.

Examples:

>>> s = wrapped.scope("req")
>>> isinstance(s, Scope)
True

Source code in src/doppy_di/devkit/logging.py

def scope(self, name: str) -> Scope:
    """Log scope creation and delegate.

    Examples:
        >>> s = wrapped.scope("req")
        >>> isinstance(s, Scope)
        True
    """
    self.log(f"scope({name!r})")
    return self.wrapped.scope(name)

MissingAnnotationError

Bases: TypeError

Raised when an injectable class has an unannotated dependency.

Source code in src/doppy_di/auto_wiring.py

class MissingAnnotationError(TypeError):
    """Raised when an injectable class has an unannotated dependency."""

    def __init__(self, cls: type, param: str) -> None:
        self.cls = cls
        self.param = param
        super().__init__(f"Missing annotation for {param!r} in {cls!r}")

MissingDependencyError

Bases: ServiceNotFoundError

Raised when a deep dependency chain fails to resolve.

Subclasses ServiceNotFoundError so existing KeyError handling keeps working.

Examples:

>>> err = MissingDependencyError("c", ["a", "b", "c"])
>>> err.key
'c'
>>> err.resolution_path
['a', 'b', 'c']

Source code in src/doppy_di/container.py

class MissingDependencyError(ServiceNotFoundError):
    """Raised when a deep dependency chain fails to resolve.

    Subclasses ``ServiceNotFoundError`` so existing ``KeyError`` handling
    keeps working.

    Examples:
        >>> err = MissingDependencyError("c", ["a", "b", "c"])
        >>> err.key
        'c'
        >>> err.resolution_path
        ['a', 'b', 'c']
    """

    def __init__(
        self,
        key: Key,
        resolution_path: Optional[List[Key]] = None,
        scope: Optional[str] = None,
        registration_source: Optional[RegistrationSource] = None,
    ) -> None:
        self.key = key
        self.resolution_path = list(resolution_path or [])
        self.scope = scope
        self.registration_source = registration_source
        super().__init__(key)

    def __str__(self) -> str:
        parts = [f"Cannot resolve {self.key!r}:"]
        if self.resolution_path:
            parts.append("")
            parts.append(_format_tree(self.resolution_path, self.key))
        if self.scope is not None:
            parts.append("")
            parts.append(f"Requested scope: {self.scope}")
        if self.registration_source is not None:
            parts.append("")
            parts.append(f"Registration source: {self.registration_source}")
        if self.resolution_path:
            parts.append("")
            parts.append("Resolution path: " + " → ".join(map(repr, self.resolution_path)))
        return "\n".join(parts)

MissingExternalArgumentError

Bases: TypeError

Raised when a declared External() argument is not supplied.

Source code in src/doppy_di/inject.py

class MissingExternalArgumentError(TypeError):
    """Raised when a declared ``External()`` argument is not supplied."""

    def __init__(self, func: Callable[..., Any], name: str) -> None:
        self.func = func
        self.name = name
        super().__init__(f"Missing external argument {name!r} for {func!r}")

NestedPolicy

Bases: Protocol

Policy used to compare nested objects.

Implementations compare the resolved nested attribute against the object stored under the nested rule key.

Examples:

>>> isinstance(SameObjectPolicy(), NestedPolicy)
True

Source code in src/doppy_di/devkit/nested.py

class NestedPolicy(Protocol):
    """Policy used to compare nested objects.

    Implementations compare the resolved nested attribute against the object
    stored under the nested rule key.

    Examples:
        >>> isinstance(SameObjectPolicy(), NestedPolicy)
        True
    """

    def check(self, nested: Any, resolved: Any) -> bool:
        """Return True if nested object is valid."""

check(nested, resolved)

Return True if nested object is valid.

Source code in src/doppy_di/devkit/nested.py

def check(self, nested: Any, resolved: Any) -> bool:
    """Return True if nested object is valid."""

NestedRules

Track nested relations and validate resolved objects.

Source code in src/doppy_di/devkit/nested.py

class NestedRules:
    """Track nested relations and validate resolved objects."""

    __slots__ = ("map", "same_policy")

    def __init__(self) -> None:
        self.map: Dict[Key, List[str]] = {}
        self.same_policy: NestedPolicy = SameValuePolicy()

    def add_nested(self, parent: Key, child: str, rule: Rule, ruleset: RuleSetProtocol) -> None:
        """Register a nested dependency for a parent key.

        The nested key ``(parent, child)`` is added to the shared ruleset and
        tracked for validation.

        Examples:
            >>> nested = NestedRules()
            >>> rule = Rule(("db", "conn"), lambda: object())
            >>> rs = RuleSet()
            >>> nested.add_nested("db", "conn", rule, rs)
            >>> rs.has(("db", "conn"))
            True
        """
        nested_key = (parent, child)
        ruleset.add(nested_key, replace(rule, nested=True))
        self.map.setdefault(parent, [])
        if child not in self.map[parent]:
            self.map[parent].append(child)

    def children_of(self, parent: Key) -> List[str]:
        """Return the registered nested child names for a parent."""
        return list(self.map.get(parent, []))

    def validate_nested(self, parent: Key, container: Container, parent_obj: Any = None) -> None:
        """Validate nested rules for a resolved parent object."""
        children = self.children_of(parent)
        if not children:
            return

        if parent_obj is None:
            parent_obj = container.get(parent)
        for child in children:
            nested_key = (parent, child)
            nested_obj = container.get(nested_key)

            if not hasattr(parent_obj, child):
                raise NestedRuleError(parent, child, f"field {child!r} not found")

            resolved = getattr(parent_obj, child)
            if not self.same_policy.check(nested_obj, resolved):
                raise NestedRuleError(parent, child, "nested validation failed")

add_nested(parent, child, rule, ruleset)

Register a nested dependency for a parent key.

The nested key (parent, child) is added to the shared ruleset and tracked for validation.

Examples:

>>> nested = NestedRules()
>>> rule = Rule(("db", "conn"), lambda: object())
>>> rs = RuleSet()
>>> nested.add_nested("db", "conn", rule, rs)
>>> rs.has(("db", "conn"))
True

Source code in src/doppy_di/devkit/nested.py

def add_nested(self, parent: Key, child: str, rule: Rule, ruleset: RuleSetProtocol) -> None:
    """Register a nested dependency for a parent key.

    The nested key ``(parent, child)`` is added to the shared ruleset and
    tracked for validation.

    Examples:
        >>> nested = NestedRules()
        >>> rule = Rule(("db", "conn"), lambda: object())
        >>> rs = RuleSet()
        >>> nested.add_nested("db", "conn", rule, rs)
        >>> rs.has(("db", "conn"))
        True
    """
    nested_key = (parent, child)
    ruleset.add(nested_key, replace(rule, nested=True))
    self.map.setdefault(parent, [])
    if child not in self.map[parent]:
        self.map[parent].append(child)

children_of(parent)

Return the registered nested child names for a parent.

Source code in src/doppy_di/devkit/nested.py

def children_of(self, parent: Key) -> List[str]:
    """Return the registered nested child names for a parent."""
    return list(self.map.get(parent, []))

validate_nested(parent, container, parent_obj=None)

Validate nested rules for a resolved parent object.

Source code in src/doppy_di/devkit/nested.py

def validate_nested(self, parent: Key, container: Container, parent_obj: Any = None) -> None:
    """Validate nested rules for a resolved parent object."""
    children = self.children_of(parent)
    if not children:
        return

    if parent_obj is None:
        parent_obj = container.get(parent)
    for child in children:
        nested_key = (parent, child)
        nested_obj = container.get(nested_key)

        if not hasattr(parent_obj, child):
            raise NestedRuleError(parent, child, f"field {child!r} not found")

        resolved = getattr(parent_obj, child)
        if not self.same_policy.check(nested_obj, resolved):
            raise NestedRuleError(parent, child, "nested validation failed")

OrderPolicy

Bases: Protocol

A strategy for controlling resolution order.

Examples:

>>> isinstance(UnorderedPolicy(), OrderPolicy)
True

Source code in src/doppy_di/devkit/policy.py

class OrderPolicy(Protocol):
    """A strategy for controlling resolution order.

    Examples:
        >>> isinstance(UnorderedPolicy(), OrderPolicy)
        True
    """

    def before_resolve(self, key: Key, ruleset: RuleSetProtocol, ctx: ResolveContext) -> None:
        """Run before object resolution."""

    def after_resolve(
        self, key: Key, obj: Any, ruleset: RuleSetProtocol, ctx: ResolveContext
    ) -> None:
        """Run after object resolution."""

after_resolve(key, obj, ruleset, ctx)

Run after object resolution.

Source code in src/doppy_di/devkit/policy.py

def after_resolve(
    self, key: Key, obj: Any, ruleset: RuleSetProtocol, ctx: ResolveContext
) -> None:
    """Run after object resolution."""

before_resolve(key, ruleset, ctx)

Run before object resolution.

Source code in src/doppy_di/devkit/policy.py

def before_resolve(self, key: Key, ruleset: RuleSetProtocol, ctx: ResolveContext) -> None:
    """Run before object resolution."""

OverrideContext

Context manager for temporary stack-based overrides.

Entering pushes a validated :class:OverrideLayer onto the container's layer stack. Lookups walk the stack LIFO, so the last override() wins. Exiting pops the layer and restores the original rules — even when the with block raises.

Callable override values are treated as factories and invoked on every resolution, mirroring a transient service.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> c = builder.build()
>>> with c.override("x", 2):
...     c.get("x")
2
>>> c.get("x")
1

>>> with c.override({"x": 3}):
...     c.get("x")
3
>>> c.get("x")
1

Source code in src/doppy_di/container.py

class OverrideContext:
    """Context manager for temporary stack-based overrides.

    Entering pushes a validated :class:`OverrideLayer` onto the container's
    layer stack. Lookups walk the stack LIFO, so the last ``override()``
    wins. Exiting pops the layer and restores the original rules — even when
    the ``with`` block raises.

    Callable override values are treated as factories and invoked on every
    resolution, mirroring a transient service.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> c = builder.build()
        >>> with c.override("x", 2):
        ...     c.get("x")
        2
        >>> c.get("x")
        1

        >>> with c.override({"x": 3}):
        ...     c.get("x")
        3
        >>> c.get("x")
        1
    """

    __slots__ = ("container", "values")

    def __init__(self, container: Container, values: Dict[Key, Any]) -> None:
        self.container = container
        self.values = dict(values)

    def __enter__(self) -> OverrideContext:
        layer = OverrideLayer(self.container, self.values)
        self.container._override_layers.append(layer)
        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        if self.container._override_layers:
            self.container._override_layers.pop()

ParallelPolicy dataclass

Resolve independent branches concurrently (async only).

The policy orders keys by dependency level. In aget() each level is resolved concurrently with asyncio.gather. In sync get() the order is honoured sequentially, because there is no event loop.

Examples:

>>> policy = ParallelPolicy()
>>> list(policy.order({}, "a"))
['a']

Source code in src/doppy_di/resolution.py

@dataclass(frozen=True)
class ParallelPolicy:
    """Resolve independent branches concurrently (async only).

    The policy orders keys by dependency level. In ``aget()`` each level is
    resolved concurrently with ``asyncio.gather``. In sync ``get()`` the
    order is honoured sequentially, because there is no event loop.

    Examples:
        >>> policy = ParallelPolicy()
        >>> list(policy.order({}, "a"))
        ['a']
    """

    def order(
        self,
        graph: Mapping[Key, Rule],
        root: Key,
    ) -> Iterable[Key]:
        return _topological(graph, _reachable(graph, root))

ParentFirstPolicy dataclass

Resolve parent first, then optionally inspect children.

Examples:

>>> policy = ParentFirstPolicy(nested={"service": ["repo"]})
>>> isinstance(policy, ParentFirstPolicy)
True

Source code in src/doppy_di/devkit/policy.py

@dataclass(frozen=True)
class ParentFirstPolicy:
    """Resolve parent first, then optionally inspect children.

    Examples:
        >>> policy = ParentFirstPolicy(nested={"service": ["repo"]})
        >>> isinstance(policy, ParentFirstPolicy)
        True
    """

    nested: Dict[Key, List[str]]

    def __init__(self, nested: Optional[Dict[Key, List[str]]] = None) -> None:
        object.__setattr__(self, "nested", dict(nested or {}))

    def before_resolve(self, key: Key, ruleset: RuleSetProtocol, ctx: ResolveContext) -> None:
        return None

    def after_resolve(
        self, key: Key, obj: Any, ruleset: RuleSetProtocol, ctx: ResolveContext
    ) -> None:
        for child_name in self.nested.get(key, []):
            child_key = (key, child_name)
            ctx.get(child_key)

Qualifier dataclass

Marker for named dependencies via typing.Annotated.

Examples:

>>> Qualifier("read")
Qualifier(name='read')

Source code in src/doppy_di/container.py

@dataclass(frozen=True)
class Qualifier:
    """Marker for named dependencies via ``typing.Annotated``.

    Examples:
        >>> Qualifier("read")
        Qualifier(name='read')
    """

    name: str

RegistrationSource dataclass

Source location where a rule was registered.

Examples:

>>> src = RegistrationSource("app/container.py", 42, "setup")
>>> str(src)
'app/container.py:42'

Source code in src/doppy_di/container.py

@dataclass(frozen=True)
class RegistrationSource:
    """Source location where a rule was registered.

    Examples:
        >>> src = RegistrationSource("app/container.py", 42, "setup")
        >>> str(src)
        'app/container.py:42'
    """

    filename: str
    lineno: int
    function_name: str

    def __str__(self) -> str:
        return f"{self.filename}:{self.lineno}"

ResolutionCancelledError

Bases: CancelledError

Raised when aget() is cancelled after partially creating resources.

Source code in src/doppy_di/container.py

class ResolutionCancelledError(asyncio.CancelledError):
    """Raised when ``aget()`` is cancelled after partially creating resources."""

    def __init__(self, key: Key) -> None:
        self.key = key
        super().__init__(f"Resolution of {key!r} was cancelled")

ResolutionChildrenFirstPolicy dataclass

Resolve children before parents.

Transitive dependencies (children) are resolved before the requested root (parent).

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("b", 1)
>>> builder.service("a", lambda b: b, deps=["b"])
>>> container = builder.build(policy=ChildrenFirstPolicy())
>>> container.get("a")
1

Source code in src/doppy_di/resolution.py

@dataclass(frozen=True)
class ChildrenFirstPolicy:
    """Resolve children before parents.

    Transitive dependencies (children) are resolved before the requested
    root (parent).

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("b", 1)
        >>> builder.service("a", lambda b: b, deps=["b"])
        >>> container = builder.build(policy=ChildrenFirstPolicy())
        >>> container.get("a")
        1
    """

    def order(
        self,
        graph: Mapping[Key, Rule],
        root: Key,
    ) -> Iterable[Key]:
        return _topological(graph, _reachable(graph, root))

ResolutionParentFirstPolicy dataclass

Resolve parents before children.

The requested root (parent) is attempted first; its transitive dependencies (children) follow.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("b", 1)
>>> builder.service("a", lambda b: b, deps=["b"])
>>> container = builder.build(policy=ParentFirstPolicy())
>>> container.get("a")
1

Source code in src/doppy_di/resolution.py

@dataclass(frozen=True)
class ParentFirstPolicy:
    """Resolve parents before children.

    The requested root (parent) is attempted first; its transitive
    dependencies (children) follow.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("b", 1)
        >>> builder.service("a", lambda b: b, deps=["b"])
        >>> container = builder.build(policy=ParentFirstPolicy())
        >>> container.get("a")
        1
    """

    def order(
        self,
        graph: Mapping[Key, Rule],
        root: Key,
    ) -> Iterable[Key]:
        return reversed(_topological(graph, _reachable(graph, root)))

ResolutionPolicy

Bases: Protocol

A strategy for ordering dependency resolution.

Implementations receive the rule graph and the requested root key and return an iteration order of keys. The container resolves keys in that order.

Examples:

>>> class ReversePolicy:
...     def order(self, graph, root):
...         return reversed(list(graph.keys()))
>>> isinstance(ReversePolicy(), ResolutionPolicy)
True

Source code in src/doppy_di/resolution.py

@runtime_checkable
class ResolutionPolicy(Protocol):
    """A strategy for ordering dependency resolution.

    Implementations receive the rule graph and the requested root key and
    return an iteration order of keys. The container resolves keys in that
    order.

    Examples:
        >>> class ReversePolicy:
        ...     def order(self, graph, root):
        ...         return reversed(list(graph.keys()))
        >>> isinstance(ReversePolicy(), ResolutionPolicy)
        True
    """

    def order(
        self,
        graph: Mapping[Key, Rule],
        root: Key,
    ) -> Iterable[Key]:
        """Return the resolution order of keys for ``root``."""
        ...

order(graph, root)

Return the resolution order of keys for root.

Source code in src/doppy_di/resolution.py

def order(
    self,
    graph: Mapping[Key, Rule],
    root: Key,
) -> Iterable[Key]:
    """Return the resolution order of keys for ``root``."""
    ...

ResolveContext

Resolution context used during object creation.

Provides access to the container and scope for dependency resolution.

Examples:

>>> builder = ContainerBuilder()
>>> builder.service("x", lambda: 1)
>>> c = builder.build()
>>> ctx = ResolveContext(c)
>>> ctx.get("x")
1

Source code in src/doppy_di/container.py

class ResolveContext:
    """Resolution context used during object creation.

    Provides access to the container and scope for dependency resolution.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.service("x", lambda: 1)
        >>> c = builder.build()
        >>> ctx = ResolveContext(c)
        >>> ctx.get("x")
        1
    """

    __slots__ = ("container", "scope")

    def __init__(
        self,
        container: Container,
        scope: Optional[Scope] = None,
    ) -> None:
        self.container = container
        self.scope = scope or container

    def get(self, key: Key, _scope_name: Optional[str] = None) -> Any:
        if isinstance(self.scope, Scope):
            return self.scope.get(key)
        return self.container.get(key, _scope_name=_scope_name)

ResourceFinalizationError

Bases: Exception

Raised when one or more yield providers fail to finalize.

Only raised when finalization_errors=True is set on the builder.

Source code in src/doppy_di/container.py

class ResourceFinalizationError(Exception):
    """Raised when one or more yield providers fail to finalize.

    Only raised when ``finalization_errors=True`` is set on the builder.
    """

    def __init__(self, errors: List[Tuple[Key, Exception]]) -> None:
        self.errors = list(errors)
        super().__init__(f"{len(self.errors)} resource(s) failed to finalize")

    def __str__(self) -> str:
        parts = ["Resource finalization failed:"]
        for key, exc in self.errors:
            parts.append(f"  {key!r}: {exc!r}")
        return "\n".join(parts)

Rule dataclass

Immutable service rule.

Parameters:

Name	Type	Description	Default
key	Key	Registration key.	required
make	Callable[..., Any]	Factory callable.	required
lifetime	Lifetime	Service lifetime.	'transient'
deps	Tuple[Key, ...]	Dependency keys.	()

Examples:

>>> rule = Rule("answer", lambda: 42, "singleton", ())
>>> rule.key
'answer'
>>> rule.lifetime
'singleton'

Source code in src/doppy_di/container.py

@dataclass(frozen=True)
class Rule:
    """Immutable service rule.

    Args:
        key: Registration key.
        make: Factory callable.
        lifetime: Service lifetime.
        deps: Dependency keys.

    Examples:
        >>> rule = Rule("answer", lambda: 42, "singleton", ())
        >>> rule.key
        'answer'
        >>> rule.lifetime
        'singleton'
    """

    key: Key
    make: Callable[..., Any]
    lifetime: Lifetime = "transient"
    deps: Tuple[Key, ...] = ()
    yield_provider: bool = False
    async_yield_provider: bool = False
    nested: bool = False
    scope: Optional[str] = None
    registration_source: Optional[RegistrationSource] = None
    is_async: bool = False

    def __post_init__(self) -> None:
        LifetimePolicy.validate(self.lifetime)
        if inspect.isasyncgenfunction(self.make):
            object.__setattr__(self, "async_yield_provider", True)
        elif inspect.isgeneratorfunction(self.make):
            object.__setattr__(self, "yield_provider", True)
        object.__setattr__(
            self,
            "is_async",
            inspect.iscoroutinefunction(self.make) or self.async_yield_provider,
        )

RuleSet

Immutable-by-convention rule storage and dependency graph.

Examples:

>>> rules = RuleSet()
>>> rules.add("x", Rule("x", lambda: 1))
>>> rules.find("x").key
'x'
>>> rules.has("x")
True

Source code in src/doppy_di/container.py

class RuleSet:
    """Immutable-by-convention rule storage and dependency graph.

    Examples:
        >>> rules = RuleSet()
        >>> rules.add("x", Rule("x", lambda: 1))
        >>> rules.find("x").key
        'x'
        >>> rules.has("x")
        True
    """

    __slots__ = ("defer_cycle_check", "graph", "map", "version")

    def __init__(
        self,
        rules_map: Optional[Dict[Key, Rule]] = None,
        graph: Optional[Dict[Key, Tuple[Key, ...]]] = None,
        defer_cycle_check: bool = False,
    ) -> None:
        """Initialize storage from optional existing map and graph."""
        self.map = dict(rules_map or {})
        self.graph = dict(graph or {})
        self.defer_cycle_check = defer_cycle_check
        self.version = 0

    def copy(self) -> RuleSet:
        """Return a deep copy of this RuleSet."""
        return RuleSet(
            rules_map=dict(self.map),
            graph=dict(self.graph),
            defer_cycle_check=self.defer_cycle_check,
        )

    def add(self, key: Key, rule: Rule) -> None:
        """Add a rule and validate graph cycles.

        Raises:
            CycleError: If adding the rule creates a dependency cycle.

        Examples:
            >>> rules = RuleSet()
            >>> rules.add("a", Rule("a", lambda: 1, deps=("b",)))
            >>> rules.has("a")
            True
        """
        old_map = dict(self.map)
        old_graph = dict(self.graph)
        self.map[key] = rule
        self.graph[key] = tuple(rule.deps)
        if not self.defer_cycle_check:
            try:
                self._check_cycle(key)
            except CycleError:
                self.map = old_map
                self.graph = old_graph
                raise
        self.version += 1

    def find(self, key: Key) -> Rule:
        """Return a rule by key.

        Raises:
            ServiceNotFoundError: If key is not registered.

        Examples:
            >>> rules = RuleSet()
            >>> rules.add("x", Rule("x", lambda: 1))
            >>> rules.find("x")
            Rule(key='x', make=..., lifetime='transient', deps=())
            >>> rules.find("missing")  # doctest: +IGNORE_EXCEPTION_DETAIL
            Traceback (most recent call last):
            ...
            ServiceNotFoundError
        """
        try:
            return self.map[key]
        except KeyError:
            raise ServiceNotFoundError(key) from None

    def has(self, key: Key) -> bool:
        """Check whether a key is registered.

        Examples:
            >>> rules = RuleSet()
            >>> rules.add("x", Rule("x", lambda: 1))
            >>> rules.has("x")
            True
            >>> rules.has("y")
            False
        """
        return key in self.map

    def deps_of(self, key: Key) -> Tuple[Key, ...]:
        """Return direct dependencies for a key.

        Examples:
            >>> rules = RuleSet()
            >>> rules.add("a", Rule("a", lambda: 1, deps=("b", "c")))
            >>> rules.deps_of("a")
            ('b', 'c')
        """
        return self.graph.get(key, ())

    def keys(self) -> Tuple[Key, ...]:
        """Return registered keys.

        Examples:
            >>> rules = RuleSet()
            >>> rules.add("x", Rule("x", lambda: 1))
            >>> rules.add("y", Rule("y", lambda: 2))
            >>> sorted(rules.keys())
            ['x', 'y']
        """
        return tuple(self.map.keys())

    def _check_cycle(self, start: Key) -> None:
        """Check graph cycles from the given start node."""
        stack: List[Key] = []
        on_stack: set[Key] = set()
        visited: set[Key] = set()

        def dfs(node: Key) -> None:
            if node in on_stack:
                raise DependencyCycleError([*stack, node])
            if node in visited:
                return
            visited.add(node)
            on_stack.add(node)
            stack.append(node)
            for dep in self.graph.get(node, ()):
                if dep in self.map:
                    dfs(dep)
            stack.pop()
            on_stack.remove(node)

        dfs(start)

__init__(rules_map=None, graph=None, defer_cycle_check=False)

Initialize storage from optional existing map and graph.

Source code in src/doppy_di/container.py

def __init__(
    self,
    rules_map: Optional[Dict[Key, Rule]] = None,
    graph: Optional[Dict[Key, Tuple[Key, ...]]] = None,
    defer_cycle_check: bool = False,
) -> None:
    """Initialize storage from optional existing map and graph."""
    self.map = dict(rules_map or {})
    self.graph = dict(graph or {})
    self.defer_cycle_check = defer_cycle_check
    self.version = 0

add(key, rule)

Add a rule and validate graph cycles.

Raises:

Type	Description
CycleError	If adding the rule creates a dependency cycle.

Examples:

>>> rules = RuleSet()
>>> rules.add("a", Rule("a", lambda: 1, deps=("b",)))
>>> rules.has("a")
True

Source code in src/doppy_di/container.py

def add(self, key: Key, rule: Rule) -> None:
    """Add a rule and validate graph cycles.

    Raises:
        CycleError: If adding the rule creates a dependency cycle.

    Examples:
        >>> rules = RuleSet()
        >>> rules.add("a", Rule("a", lambda: 1, deps=("b",)))
        >>> rules.has("a")
        True
    """
    old_map = dict(self.map)
    old_graph = dict(self.graph)
    self.map[key] = rule
    self.graph[key] = tuple(rule.deps)
    if not self.defer_cycle_check:
        try:
            self._check_cycle(key)
        except CycleError:
            self.map = old_map
            self.graph = old_graph
            raise
    self.version += 1

copy()

Return a deep copy of this RuleSet.

Source code in src/doppy_di/container.py

def copy(self) -> RuleSet:
    """Return a deep copy of this RuleSet."""
    return RuleSet(
        rules_map=dict(self.map),
        graph=dict(self.graph),
        defer_cycle_check=self.defer_cycle_check,
    )

deps_of(key)

Return direct dependencies for a key.

Examples:

>>> rules = RuleSet()
>>> rules.add("a", Rule("a", lambda: 1, deps=("b", "c")))
>>> rules.deps_of("a")
('b', 'c')

Source code in src/doppy_di/container.py

def deps_of(self, key: Key) -> Tuple[Key, ...]:
    """Return direct dependencies for a key.

    Examples:
        >>> rules = RuleSet()
        >>> rules.add("a", Rule("a", lambda: 1, deps=("b", "c")))
        >>> rules.deps_of("a")
        ('b', 'c')
    """
    return self.graph.get(key, ())

find(key)

Return a rule by key.

Raises:

Type	Description
ServiceNotFoundError	If key is not registered.

Examples:

>>> rules = RuleSet()
>>> rules.add("x", Rule("x", lambda: 1))
>>> rules.find("x")
Rule(key='x', make=..., lifetime='transient', deps=())
>>> rules.find("missing")
Traceback (most recent call last):
...
ServiceNotFoundError

Source code in src/doppy_di/container.py

def find(self, key: Key) -> Rule:
    """Return a rule by key.

    Raises:
        ServiceNotFoundError: If key is not registered.

    Examples:
        >>> rules = RuleSet()
        >>> rules.add("x", Rule("x", lambda: 1))
        >>> rules.find("x")
        Rule(key='x', make=..., lifetime='transient', deps=())
        >>> rules.find("missing")  # doctest: +IGNORE_EXCEPTION_DETAIL
        Traceback (most recent call last):
        ...
        ServiceNotFoundError
    """
    try:
        return self.map[key]
    except KeyError:
        raise ServiceNotFoundError(key) from None

has(key)

Check whether a key is registered.

Examples:

>>> rules = RuleSet()
>>> rules.add("x", Rule("x", lambda: 1))
>>> rules.has("x")
True
>>> rules.has("y")
False

Source code in src/doppy_di/container.py

def has(self, key: Key) -> bool:
    """Check whether a key is registered.

    Examples:
        >>> rules = RuleSet()
        >>> rules.add("x", Rule("x", lambda: 1))
        >>> rules.has("x")
        True
        >>> rules.has("y")
        False
    """
    return key in self.map

keys()

Return registered keys.

Examples:

>>> rules = RuleSet()
>>> rules.add("x", Rule("x", lambda: 1))
>>> rules.add("y", Rule("y", lambda: 2))
>>> sorted(rules.keys())
['x', 'y']

Source code in src/doppy_di/container.py

def keys(self) -> Tuple[Key, ...]:
    """Return registered keys.

    Examples:
        >>> rules = RuleSet()
        >>> rules.add("x", Rule("x", lambda: 1))
        >>> rules.add("y", Rule("y", lambda: 2))
        >>> sorted(rules.keys())
        ['x', 'y']
    """
    return tuple(self.map.keys())

SameObjectPolicy dataclass

Check object identity for nested values.

Examples:

>>> policy = SameObjectPolicy()
>>> obj = object()
>>> policy.check(obj, obj)
True

Source code in src/doppy_di/devkit/nested.py

@dataclass(frozen=True)
class SameObjectPolicy:
    """Check object identity for nested values.

    Examples:
        >>> policy = SameObjectPolicy()
        >>> obj = object()
        >>> policy.check(obj, obj)
        True
    """

    def check(self, nested: Any, resolved: Any) -> bool:
        """Return True only when both references are the same object."""
        return nested is resolved

check(nested, resolved)

Return True only when both references are the same object.

Source code in src/doppy_di/devkit/nested.py

def check(self, nested: Any, resolved: Any) -> bool:
    """Return True only when both references are the same object."""
    return nested is resolved

SameValuePolicy dataclass

Check value equality for nested values.

Attributes:

Name	Type	Description
strict	bool	When True, propagate comparison exceptions instead of returning False.

Examples:

>>> policy = SameValuePolicy()
>>> policy.check(1, 1)
True
>>> policy.check(1, 2)
False

Source code in src/doppy_di/devkit/nested.py

@dataclass(frozen=True)
class SameValuePolicy:
    """Check value equality for nested values.

    Attributes:
        strict: When True, propagate comparison exceptions instead of
            returning False.

    Examples:
        >>> policy = SameValuePolicy()
        >>> policy.check(1, 1)
        True
        >>> policy.check(1, 2)
        False
    """

    strict: bool = False

    def check(self, nested: Any, resolved: Any) -> bool:
        """Return True when both values compare equal.

        When ``strict`` is False, exceptions raised during comparison are
        logged and treated as inequality.
        """
        try:
            return bool(nested == resolved)
        except Exception as exc:
            if self.strict:
                raise
            logger.warning(
                "SameValuePolicy check failed for %r == %r: %s",
                nested,
                resolved,
                exc,
            )
            return False

check(nested, resolved)

Return True when both values compare equal.

When strict is False, exceptions raised during comparison are logged and treated as inequality.

Source code in src/doppy_di/devkit/nested.py

def check(self, nested: Any, resolved: Any) -> bool:
    """Return True when both values compare equal.

    When ``strict`` is False, exceptions raised during comparison are
    logged and treated as inequality.
    """
    try:
        return bool(nested == resolved)
    except Exception as exc:
        if self.strict:
            raise
        logger.warning(
            "SameValuePolicy check failed for %r == %r: %s",
            nested,
            resolved,
            exc,
        )
        return False

Scope

Scope-local cache over a container.

Scopes allow caching within a with block. All resolved values are cached until the scope exits.

Attributes:

Name	Type	Description
APP		Application-wide scope name.
REQUEST		Per-request scope name.
SESSION		Per-session scope name.

Examples:

>>> builder = ContainerBuilder()
>>> builder.service("x", lambda: object(), lifetime="transient")
>>> c = builder.build()
>>> with c.scope("req") as s:
...     a = s.get("x")
...     b = s.get("x")
...     a is b
True

Source code in src/doppy_di/container.py

class Scope:
    """Scope-local cache over a container.

    Scopes allow caching within a ``with`` block. All resolved values are
    cached until the scope exits.

    Attributes:
        APP: Application-wide scope name.
        REQUEST: Per-request scope name.
        SESSION: Per-session scope name.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.service("x", lambda: object(), lifetime="transient")
        >>> c = builder.build()
        >>> with c.scope("req") as s:
        ...     a = s.get("x")
        ...     b = s.get("x")
        ...     a is b
        True
    """

    APP = "app"
    REQUEST = "request"
    SESSION = "session"

    __slots__ = (
        "_async_exit_stack",
        "_depth",
        "_exit_stack",
        "_resolver_previous",
        "cache",
        "container",
        "name",
        "request_context",
        "session_context",
    )

    def __init__(self, container: Container, name: str) -> None:
        """Create a named scope with an empty local cache."""
        self.container = container
        self.name = name
        self.cache: Dict[Key, Any] = {}
        self.request_context: Dict[Key, Any] = {}
        self.session_context: Dict[Key, Any] = {}
        self._exit_stack: List[Tuple[Key, ExitStack]] = []
        self._async_exit_stack: List[Tuple[Key, AsyncExitStack]] = []
        self._depth = 0
        self._resolver_previous: Optional[object] = None

    def set_context(
        self,
        key: Key,
        value: Any,
        scope: Optional[Union[str, "Scope"]] = None,
    ) -> None:
        """Store a context value available to ``from_context`` providers.

        Request-scope values are cleared when the scope exits; session-scope
        values persist for the scope lifetime.

        Examples:
            >>> from doppy_di import Container
            >>> from doppy_di.providers import from_context
            >>> services = Container()
            >>> services.user = from_context("user")
            >>> with services.scope("req") as s:
            ...     s.set_context("user", "alice")
            ...     s.get("user")
            'alice'
        """
        from .providers import _scope_value

        is_session = scope is not None and _scope_value(scope) == _scope_value(Scope.SESSION)
        if is_session:
            self.session_context[key] = value
        else:
            self.request_context[key] = value

    def get_context(
        self,
        key: Key,
        scope: Optional[Union[str, "Scope"]] = None,
    ) -> Any:
        """Read a context value stored via :meth:`set_context`.

        Raises:
            ContextValueMissingError: When the key is absent in the scope.
        """
        from .providers import _scope_value

        is_session = scope is not None and _scope_value(scope) == _scope_value(Scope.SESSION)
        store = self.session_context if is_session else self.request_context
        try:
            return store[key]
        except KeyError:
            raise ContextValueMissingError(
                key,
                _scope_value(scope) if scope is not None else _scope_value(Scope.REQUEST),
            ) from None

    def get(self, key: Key) -> Any:
        """Resolve key from scope cache or underlying container.

        Examples:
            >>> builder = ContainerBuilder()
            >>> builder.value("x", 1)
            >>> c = builder.build()
            >>> with c.scope("s") as s:
            ...     s.get("x")
            1
        """
        if key in self.cache:
            self.container._trace(key, 0.0, True, self.name)
            return self.cache[key]
        try:
            rule = self.container.config.ruleset.find(key)
        except ServiceNotFoundError:
            from .providers import implicit_collection_rule

            if implicit_collection_rule(key, self.container.config.ruleset) is None:
                raise
            obj = self.container.get(key, _scope_name=self.name)
            self.cache[key] = obj
            return obj
        if rule.yield_provider:
            started = self.container._tracer is not None
            start = time.perf_counter() if started else 0.0
            stack = ExitStack()
            try:
                obj = stack.enter_context(contextmanager(rule.make)())
            except RuntimeError as exc:
                if "didn't yield" in str(exc):
                    raise YieldNotCalledError(key) from None
                raise
            if started:
                self.container._trace(key, time.perf_counter() - start, False, self.name)
            self._exit_stack.append((key, stack))
            self.cache[key] = obj
            return obj
        obj = self.container.get(key, _scope_name=self.name)
        self.cache[key] = obj
        return obj

    def __enter__(self) -> Scope:
        self._depth += 1
        _previous = _ACTIVE_REQUEST_RESOLVER.get()
        _ACTIVE_REQUEST_RESOLVER.set(self)
        self._resolver_previous = _previous
        return self

    def __exit__(
        self,
        exc_type: Optional[Type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        _ACTIVE_REQUEST_RESOLVER.set(self._resolver_previous)
        self._depth -= 1
        if self._depth == 0:
            self.cache.clear()
            self.request_context.clear()
            errors: List[Tuple[Key, Exception]] = []
            for key, stack in self._exit_stack:
                try:
                    stack.close()
                except Exception as exc:
                    if self.container.config.finalization_errors:
                        errors.append((key, exc))
                    else:
                        logger.exception("Error finalizing yield provider %r", key)
            self._exit_stack.clear()
            if errors:
                raise ResourceFinalizationError(errors)

__init__(container, name)

Create a named scope with an empty local cache.

Source code in src/doppy_di/container.py

def __init__(self, container: Container, name: str) -> None:
    """Create a named scope with an empty local cache."""
    self.container = container
    self.name = name
    self.cache: Dict[Key, Any] = {}
    self.request_context: Dict[Key, Any] = {}
    self.session_context: Dict[Key, Any] = {}
    self._exit_stack: List[Tuple[Key, ExitStack]] = []
    self._async_exit_stack: List[Tuple[Key, AsyncExitStack]] = []
    self._depth = 0
    self._resolver_previous: Optional[object] = None

get(key)

Resolve key from scope cache or underlying container.

Examples:

>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> c = builder.build()
>>> with c.scope("s") as s:
...     s.get("x")
1

Source code in src/doppy_di/container.py

def get(self, key: Key) -> Any:
    """Resolve key from scope cache or underlying container.

    Examples:
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> c = builder.build()
        >>> with c.scope("s") as s:
        ...     s.get("x")
        1
    """
    if key in self.cache:
        self.container._trace(key, 0.0, True, self.name)
        return self.cache[key]
    try:
        rule = self.container.config.ruleset.find(key)
    except ServiceNotFoundError:
        from .providers import implicit_collection_rule

        if implicit_collection_rule(key, self.container.config.ruleset) is None:
            raise
        obj = self.container.get(key, _scope_name=self.name)
        self.cache[key] = obj
        return obj
    if rule.yield_provider:
        started = self.container._tracer is not None
        start = time.perf_counter() if started else 0.0
        stack = ExitStack()
        try:
            obj = stack.enter_context(contextmanager(rule.make)())
        except RuntimeError as exc:
            if "didn't yield" in str(exc):
                raise YieldNotCalledError(key) from None
            raise
        if started:
            self.container._trace(key, time.perf_counter() - start, False, self.name)
        self._exit_stack.append((key, stack))
        self.cache[key] = obj
        return obj
    obj = self.container.get(key, _scope_name=self.name)
    self.cache[key] = obj
    return obj

get_context(key, scope=None)

Read a context value stored via :meth:set_context.

Raises:

Type	Description
ContextValueMissingError	When the key is absent in the scope.

Source code in src/doppy_di/container.py

def get_context(
    self,
    key: Key,
    scope: Optional[Union[str, "Scope"]] = None,
) -> Any:
    """Read a context value stored via :meth:`set_context`.

    Raises:
        ContextValueMissingError: When the key is absent in the scope.
    """
    from .providers import _scope_value

    is_session = scope is not None and _scope_value(scope) == _scope_value(Scope.SESSION)
    store = self.session_context if is_session else self.request_context
    try:
        return store[key]
    except KeyError:
        raise ContextValueMissingError(
            key,
            _scope_value(scope) if scope is not None else _scope_value(Scope.REQUEST),
        ) from None

set_context(key, value, scope=None)

Store a context value available to from_context providers.

Request-scope values are cleared when the scope exits; session-scope values persist for the scope lifetime.

Examples:

>>> from doppy_di import Container
>>> from doppy_di.providers import from_context
>>> services = Container()
>>> services.user = from_context("user")
>>> with services.scope("req") as s:
...     s.set_context("user", "alice")
...     s.get("user")
'alice'

Source code in src/doppy_di/container.py

def set_context(
    self,
    key: Key,
    value: Any,
    scope: Optional[Union[str, "Scope"]] = None,
) -> None:
    """Store a context value available to ``from_context`` providers.

    Request-scope values are cleared when the scope exits; session-scope
    values persist for the scope lifetime.

    Examples:
        >>> from doppy_di import Container
        >>> from doppy_di.providers import from_context
        >>> services = Container()
        >>> services.user = from_context("user")
        >>> with services.scope("req") as s:
        ...     s.set_context("user", "alice")
        ...     s.get("user")
        'alice'
    """
    from .providers import _scope_value

    is_session = scope is not None and _scope_value(scope) == _scope_value(Scope.SESSION)
    if is_session:
        self.session_context[key] = value
    else:
        self.request_context[key] = value

ScopePolicy

Bases: Enum

Strategy for resolving scopes by name.

NAMED: reuse the same Scope object for the same name (current default).

return a fresh Scope per call, stored under a unique internal

key so its cache never leaks across calls even if exit is forgotten.

Examples:

>>> ScopePolicy.NAMED.value
'named'
>>> ScopePolicy.UNIQUE.name
'UNIQUE'

Source code in src/doppy_di/container.py

class ScopePolicy(Enum):
    """Strategy for resolving scopes by name.

    NAMED: reuse the same Scope object for the same name (current default).

    UNIQUE: return a fresh Scope per call, stored under a unique internal
            key so its cache never leaks across calls even if __exit__
            is forgotten.

    Examples:
        >>> ScopePolicy.NAMED.value
        'named'
        >>> ScopePolicy.UNIQUE.name
        'UNIQUE'
    """

    NAMED = "named"
    UNIQUE = "unique"

ScopeViolationError

Bases: Exception

Raised when a scope/lifetime rule is violated.

Defined for API completeness; not raised by the current container.

Source code in src/doppy_di/container.py

class ScopeViolationError(Exception):
    """Raised when a scope/lifetime rule is violated.

    Defined for API completeness; not raised by the current container.
    """

    def __init__(self, key: Key, scope: str, violation_type: str) -> None:
        self.key = key
        self.scope = scope
        self.violation_type = violation_type
        super().__init__(f"Scope violation for {key!r} in {scope!r}: {violation_type}")

ServiceNotFoundError

Bases: KeyError

Raised when a service key is not registered.

Examples:

>>> raise ServiceNotFoundError("missing")
Traceback (most recent call last):
...
ServiceNotFoundError: Service not found: 'missing'

Source code in src/doppy_di/container.py

class ServiceNotFoundError(KeyError):
    """Raised when a service key is not registered.

    Examples:
        >>> raise ServiceNotFoundError("missing")
        Traceback (most recent call last):
        ...
        ServiceNotFoundError: Service not found: 'missing'
    """

    def __init__(self, key: Key) -> None:
        self.key = key
        super().__init__(f"Service not found: {key!r}")

SyncFactoryReturningAwaitableError

Bases: Exception

Raised when a sync factory returns an awaitable.

Examples:

>>> raise SyncFactoryReturningAwaitableError("a")
Traceback (most recent call last):
...
SyncFactoryReturningAwaitableError: Sync factory 'a' returned an awaitable

Source code in src/doppy_di/container.py

class SyncFactoryReturningAwaitableError(Exception):
    """Raised when a sync factory returns an awaitable.

    Examples:
        >>> raise SyncFactoryReturningAwaitableError("a")
        Traceback (most recent call last):
        ...
        SyncFactoryReturningAwaitableError: Sync factory 'a' returned an awaitable
    """

    def __init__(self, key: Key) -> None:
        self.key = key
        super().__init__(f"Sync factory {key!r} returned an awaitable")

UnorderedPolicy dataclass

Policy with no extra resolution ordering.

Examples:

>>> policy = UnorderedPolicy()
>>> isinstance(policy, UnorderedPolicy)
True
>>> policy.before_resolve("x", None, ResolveContext(container))

Source code in src/doppy_di/devkit/policy.py

@dataclass(frozen=True)
class UnorderedPolicy:
    """Policy with no extra resolution ordering.

    Examples:
        >>> policy = UnorderedPolicy()
        >>> isinstance(policy, UnorderedPolicy)
        True
        >>> policy.before_resolve("x", None, ResolveContext(container))
    """

    def before_resolve(self, key: Key, ruleset: RuleSetProtocol, ctx: ResolveContext) -> None:
        return None

    def after_resolve(
        self, key: Key, obj: Any, ruleset: RuleSetProtocol, ctx: ResolveContext
    ) -> None:
        return None

UnregisteredDependencyError

Bases: ValidationError

Raised when a rule depends on an unregistered key.

Examples:

>>> raise UnregisteredDependencyError("a", "b")
Traceback (most recent call last):
...
UnregisteredDependencyError: Unregistered dependency: 'a' -> 'b'

Source code in src/doppy_di/container.py

class UnregisteredDependencyError(ValidationError):
    """Raised when a rule depends on an unregistered key.

    Examples:
        >>> raise UnregisteredDependencyError("a", "b")
        Traceback (most recent call last):
        ...
        UnregisteredDependencyError: Unregistered dependency: 'a' -> 'b'
    """

    def __init__(self, key: Key, dependency: Key) -> None:
        self.key = key
        self.dependency = dependency
        super().__init__(f"Unregistered dependency: {key!r} -> {dependency!r}")

UnregisteredTypeError

Bases: KeyError

Raised when an override targets an unregistered key.

Examples:

>>> raise UnregisteredTypeError("missing")
Traceback (most recent call last):
...
UnregisteredTypeError: Unregistered type: 'missing'

Source code in src/doppy_di/container.py

class UnregisteredTypeError(KeyError):
    """Raised when an override targets an unregistered key.

    Examples:
        >>> raise UnregisteredTypeError("missing")
        Traceback (most recent call last):
        ...
        UnregisteredTypeError: Unregistered type: 'missing'
    """

    def __init__(self, key: Key) -> None:
        self.key = key
        super().__init__(f"Unregistered type: {key!r}")

UnresolvableDependencyError

Bases: Exception

Raised when an auto-wired dependency cannot be resolved.

Source code in src/doppy_di/auto_wiring.py

class UnresolvableDependencyError(Exception):
    """Raised when an auto-wired dependency cannot be resolved."""

    def __init__(self, key: Key, dep: Key) -> None:
        self.key = key
        self.dep = dep
        super().__init__(f"Unresolvable dependency: {key!r} -> {dep!r}")

ValidatingContainer

Container view that applies order policy and validation.

Wraps a base Container with resolution ordering and optional validation rules.

Examples:

>>> from doppy_di.container import ContainerBuilder
>>> builder = ContainerBuilder()
>>> builder.value("x", 1)
>>> base = builder.build()
>>> wrapped = ValidatingContainer(
...     base, UnorderedPolicy(), ValidationRunner()
... )
>>> wrapped.get("x")
1

Source code in src/doppy_di/devkit/validation.py

class ValidatingContainer:
    """Container view that applies order policy and validation.

    Wraps a base Container with resolution ordering and optional
    validation rules.

    Examples:
        >>> from doppy_di.container import ContainerBuilder
        >>> builder = ContainerBuilder()
        >>> builder.value("x", 1)
        >>> base = builder.build()
        >>> wrapped = ValidatingContainer(
        ...     base, UnorderedPolicy(), ValidationRunner()
        ... )
        >>> wrapped.get("x")
        1
    """

    __slots__ = ("_resolving", "nested", "policy", "validator", "wrapped")

    def __init__(
        self,
        wrapped: Container,
        policy: OrderPolicy,
        validator: Optional[ValidationRunner] = None,
        nested: Optional[NestedRules] = None,
    ) -> None:
        """Wrap container with validation and resolution guard."""
        self.wrapped = wrapped
        self.policy = policy
        self.validator = validator or ValidationRunner()
        self.nested = nested
        self._resolving: set[Key] = set()

    def get(self, key: Key) -> Any:
        """Resolve key with ordering and validation.

        Examples:
            >>> wrapped = ValidatingContainer(base, UnorderedPolicy())
            >>> wrapped.get("x")
            1
        """
        if key in self._resolving:
            raise CycleError([key])
        self._resolving.add(key)
        try:
            ctx = ResolveContext(self.wrapped)
            ruleset = self.wrapped.config.ruleset

            self.policy.before_resolve(key, ruleset, ctx)
            obj = self.wrapped.get(key)
            self.policy.after_resolve(key, obj, ruleset, ctx)

            self.validator.run(self.wrapped, key, obj)

            if self.nested is not None and key in self.nested.map:
                self.nested.validate_nested(key, self.wrapped, obj)

            return obj
        finally:
            self._resolving.remove(key)

    def has(self, key: Key) -> bool:
        """Check if key is registered.

        Examples:
            >>> wrapped.has("x")
            True
        """
        return self.wrapped.has(key)

    def scope(self, name: str) -> Scope:
        """Return a scope from the wrapped container.

        Examples:
            >>> s = wrapped.scope("req")
            >>> isinstance(s, Scope)
            True
        """
        return self.wrapped.scope(name)

    def override(self, key: Key, value: Any) -> OverrideContext:
        """Override a value in the wrapped container.

        Examples:
            >>> with wrapped.override("x", 2):
            ...     wrapped.get("x")
            2
        """
        return self.wrapped.override(key, value)

__init__(wrapped, policy, validator=None, nested=None)

Wrap container with validation and resolution guard.

Source code in src/doppy_di/devkit/validation.py

def __init__(
    self,
    wrapped: Container,
    policy: OrderPolicy,
    validator: Optional[ValidationRunner] = None,
    nested: Optional[NestedRules] = None,
) -> None:
    """Wrap container with validation and resolution guard."""
    self.wrapped = wrapped
    self.policy = policy
    self.validator = validator or ValidationRunner()
    self.nested = nested
    self._resolving: set[Key] = set()

get(key)

Resolve key with ordering and validation.

Examples:

>>> wrapped = ValidatingContainer(base, UnorderedPolicy())
>>> wrapped.get("x")
1

Source code in src/doppy_di/devkit/validation.py

def get(self, key: Key) -> Any:
    """Resolve key with ordering and validation.

    Examples:
        >>> wrapped = ValidatingContainer(base, UnorderedPolicy())
        >>> wrapped.get("x")
        1
    """
    if key in self._resolving:
        raise CycleError([key])
    self._resolving.add(key)
    try:
        ctx = ResolveContext(self.wrapped)
        ruleset = self.wrapped.config.ruleset

        self.policy.before_resolve(key, ruleset, ctx)
        obj = self.wrapped.get(key)
        self.policy.after_resolve(key, obj, ruleset, ctx)

        self.validator.run(self.wrapped, key, obj)

        if self.nested is not None and key in self.nested.map:
            self.nested.validate_nested(key, self.wrapped, obj)

        return obj
    finally:
        self._resolving.remove(key)

has(key)

Check if key is registered.

Examples:

>>> wrapped.has("x")
True

Source code in src/doppy_di/devkit/validation.py

def has(self, key: Key) -> bool:
    """Check if key is registered.

    Examples:
        >>> wrapped.has("x")
        True
    """
    return self.wrapped.has(key)

override(key, value)

Override a value in the wrapped container.

Examples:

>>> with wrapped.override("x", 2):
...     wrapped.get("x")
2

Source code in src/doppy_di/devkit/validation.py

def override(self, key: Key, value: Any) -> OverrideContext:
    """Override a value in the wrapped container.

    Examples:
        >>> with wrapped.override("x", 2):
        ...     wrapped.get("x")
        2
    """
    return self.wrapped.override(key, value)

scope(name)

Return a scope from the wrapped container.

Examples:

>>> s = wrapped.scope("req")
>>> isinstance(s, Scope)
True

Source code in src/doppy_di/devkit/validation.py

def scope(self, name: str) -> Scope:
    """Return a scope from the wrapped container.

    Examples:
        >>> s = wrapped.scope("req")
        >>> isinstance(s, Scope)
        True
    """
    return self.wrapped.scope(name)

ValidationError

Bases: Exception

Base class for static graph validation errors.

Examples:

>>> raise ValidationError("bad graph")
Traceback (most recent call last):
...
ValidationError: bad graph

Source code in src/doppy_di/container.py

class ValidationError(Exception):
    """Base class for static graph validation errors.

    Examples:
        >>> raise ValidationError("bad graph")
        Traceback (most recent call last):
        ...
        ValidationError: bad graph
    """

ValidationRule

Bases: Protocol

A validation rule executed after resolution.

Examples:

>>> class MyRule:
...     def check(self, container, key, obj):
...         assert obj is not None
>>> isinstance(MyRule(), ValidationRule)
True

Source code in src/doppy_di/devkit/validation.py

class ValidationRule(Protocol):
    """A validation rule executed after resolution.

    Examples:
        >>> class MyRule:
        ...     def check(self, container, key, obj):
        ...         assert obj is not None
        >>> isinstance(MyRule(), ValidationRule)
        True
    """

    def check(self, container: Container, key: Key, obj: Any) -> None:
        """Validate resolved object."""

check(container, key, obj)

Validate resolved object.

Source code in src/doppy_di/devkit/validation.py

def check(self, container: Container, key: Key, obj: Any) -> None:
    """Validate resolved object."""

ValidationRunner dataclass

Run a list of validation rules.

Examples:

>>> runner = ValidationRunner()
>>> len(runner.rules)
0
>>> runner.add(MyRule())
>>> len(runner.rules)
1

Source code in src/doppy_di/devkit/validation.py

@dataclass(frozen=True)
class ValidationRunner:
    """Run a list of validation rules.

    Examples:
        >>> runner = ValidationRunner()
        >>> len(runner.rules)
        0
        >>> runner.add(MyRule())
        >>> len(runner.rules)
        1
    """

    rules: tuple[ValidationRule, ...]

    def __init__(self, rules: Optional[List[ValidationRule]] = None) -> None:
        object.__setattr__(self, "rules", tuple(rules or ()))

    def add(self, rule: ValidationRule) -> None:
        """Append a validation rule.

        Examples:
            >>> runner = ValidationRunner()
            >>> runner.add(MyRule())
            >>> len(runner.rules)
            1
        """
        object.__setattr__(self, "rules", (*self.rules, rule))

    def run(self, container: Container, key: Key, obj: Any) -> None:
        """Execute all registered rules.

        Examples:
            >>> runner = ValidationRunner()
            >>> runner.run(container, "x", 42)
        """
        for rule in self.rules:
            rule.check(container, key, obj)

add(rule)

Append a validation rule.

Examples:

>>> runner = ValidationRunner()
>>> runner.add(MyRule())
>>> len(runner.rules)
1

Source code in src/doppy_di/devkit/validation.py

def add(self, rule: ValidationRule) -> None:
    """Append a validation rule.

    Examples:
        >>> runner = ValidationRunner()
        >>> runner.add(MyRule())
        >>> len(runner.rules)
        1
    """
    object.__setattr__(self, "rules", (*self.rules, rule))

run(container, key, obj)

Execute all registered rules.

Examples:

>>> runner = ValidationRunner()
>>> runner.run(container, "x", 42)

Source code in src/doppy_di/devkit/validation.py

def run(self, container: Container, key: Key, obj: Any) -> None:
    """Execute all registered rules.

    Examples:
        >>> runner = ValidationRunner()
        >>> runner.run(container, "x", 42)
    """
    for rule in self.rules:
        rule.check(container, key, obj)

YieldNotCalledError

Bases: Exception

Raised when a yield provider generator does not yield.

Examples:

>>> raise YieldNotCalledError("session")
Traceback (most recent call last):
...
YieldNotCalledError: Yield provider 'session' did not yield a value

Source code in src/doppy_di/container.py

class YieldNotCalledError(Exception):
    """Raised when a yield provider generator does not yield.

    Examples:
        >>> raise YieldNotCalledError("session")
        Traceback (most recent call last):
        ...
        YieldNotCalledError: Yield provider 'session' did not yield a value
    """

    def __init__(self, key: Key) -> None:
        self.key = key
        super().__init__(f"Yield provider {key!r} did not yield a value")

Depends(dependency=None)

Declare a dependency for injection.

Parameters:

Name	Type	Description	Default
dependency	Optional[Union[Type[Any], Callable[..., Any]]]	Type to resolve from container, callable to invoke, or None to fall back to the argument annotation.	None

Examples:

>>> Depends()
<doppy_di.inject._DependsMarker object at ...>

Source code in src/doppy_di/inject.py

def Depends(  # noqa: N802
    dependency: Optional[Union[Type[Any], Callable[..., Any]]] = None,
) -> Any:
    """Declare a dependency for injection.

    Args:
        dependency: Type to resolve from container, callable to invoke, or
            ``None`` to fall back to the argument annotation.

    Examples:
        >>> Depends()
        <doppy_di.inject._DependsMarker object at ...>
    """
    return _DependsMarker(dependency)

External()

Declare a runtime parameter supplied at build time (assisted injection).

Parameters whose default is an External() marker are not resolved from the container; the caller passes them as keyword arguments when calling build(**kwargs) (or through @inject's reflected call).

Examples:

>>> External()
<doppy_di.inject._ExternalMarker object at ...>

Source code in src/doppy_di/inject.py

def External() -> Any:  # noqa: N802
    """Declare a runtime parameter supplied at build time (assisted injection).

    Parameters whose default is an ``External()`` marker are *not* resolved
    from the container; the caller passes them as keyword arguments when
    calling ``build(**kwargs)`` (or through ``@inject``'s reflected call).

    Examples:
        >>> External()
        <doppy_di.inject._ExternalMarker object at ...>
    """
    return _ExternalMarker()

assisted(func, container=None)

Build an assisted-injection factory.

Injected (annotated) parameters resolve from container (or the active scope); parameters declared with External() are supplied at build(**kwargs) / abuild(**kwargs) time.

Examples:

>>> from doppy_di.container import ContainerBuilder
>>> builder = ContainerBuilder()
>>> builder.value("repo", object())
>>> container = builder.build()
>>> def make(repo, user_id: int = External()):
...     return user_id
>>> assisted(make, container=container).build(user_id=7)
7

Source code in src/doppy_di/inject.py

def assisted(
    func: Callable[..., Any],
    container: Optional[Container] = None,
) -> _AssistedBuilder:
    """Build an assisted-injection factory.

    Injected (annotated) parameters resolve from ``container`` (or the active
    scope); parameters declared with ``External()`` are supplied at
    ``build(**kwargs)`` / ``abuild(**kwargs)`` time.

    Examples:
        >>> from doppy_di.container import ContainerBuilder
        >>> builder = ContainerBuilder()
        >>> builder.value("repo", object())
        >>> container = builder.build()
        >>> def make(repo, user_id: int = External()):
        ...     return user_id
        >>> assisted(make, container=container).build(user_id=7)
        7
    """
    return _AssistedBuilder(func, container, _build_plan(func))

injectable(cls=None, *, scope=None, qualifier=None)

Mark a class as a candidate for auto-registration.

Usable bare (@injectable) or with options (@injectable(scope=..., qualifier=...)).

Parameters:

Name	Type	Description	Default
cls	Optional[Type[Any]]	Class being decorated (bare usage).	None
scope	Optional[str]	Default lifetime for the class.	None
qualifier	Optional[str]	Named qualifier.	None

Examples:

>>> @injectable(scope="singleton")
... class Service:
...     pass
>>> Service.__doppy_injectable__
True

Source code in src/doppy_di/auto_wiring.py

def injectable(
    cls: Optional[Type[Any]] = None,
    *,
    scope: Optional[str] = None,
    qualifier: Optional[str] = None,
) -> Any:
    """Mark a class as a candidate for auto-registration.

    Usable bare (``@injectable``) or with options
    (``@injectable(scope=..., qualifier=...)``).

    Args:
        cls: Class being decorated (bare usage).
        scope: Default lifetime for the class.
        qualifier: Named qualifier.

    Examples:
        >>> @injectable(scope="singleton")
        ... class Service:
        ...     pass
        >>> Service.__doppy_injectable__
        True
    """

    def decorate(target: Type[Any]) -> Type[Any]:
        setattr(target, _INJECTABLE_FLAG, True)
        setattr(
            target,
            _INJECTABLE_META,
            {"scope": scope, "qualifier": qualifier},
        )
        return target

    if cls is not None:
        return decorate(cls)
    return decorate

is_injectable(cls)

Return True when cls is marked with @injectable.

Examples:

>>> @injectable
... class Service:
...     pass
>>> is_injectable(Service)
True
>>> is_injectable(int)
False

Source code in src/doppy_di/auto_wiring.py

def is_injectable(cls: type) -> TypeGuard[type]:
    """Return True when ``cls`` is marked with ``@injectable``.

    Examples:
        >>> @injectable
        ... class Service:
        ...     pass
        >>> is_injectable(Service)
        True
        >>> is_injectable(int)
        False
    """
    return bool(getattr(cls, _INJECTABLE_FLAG, False))

Rich diagnostic errors

Errors carry resolution paths, scope names, and optional registration sources.

Error classes

• MissingDependencyError — deep dependency chain failure. Carries key, resolution_path, scope, registration_source. Subclasses ServiceNotFoundError.
• DependencyCycleError — dependency cycle. Carries cycle. Subclasses CycleError.
• InvalidLifetimeError — unknown lifetime. Carries lifetime. Subclasses ValueError.
• ScopeViolationError — scope/lifetime violation. Carries key, scope, violation_type.
• FactoryExecutionError — wraps a factory-body exception. Carries key, original_exception, resolution_path. Raised only when wrap_factory_errors=True.
• DuplicateRegistrationError — duplicate key under the FAIL policy. Carries key, existing_source, new_source. Subclasses KeyError.
• ResourceFinalizationError — yield-provider cleanup failure. Carries errors: list[(key, exception)]. Raised only when finalization_errors=True.

Builder flags

builder = ContainerBuilder(
    track_sources=True,          # capture filename/lineno on registration
    wrap_factory_errors=True,    # wrap factory exceptions
    finalization_errors=True,    # raise on yield-provider cleanup failure
    check_cycles_on_register=False,  # defer cycle detection to resolve
)

All flags default to False (or True for cycle checking), preserving existing behavior.

Modern typing support

The public API supports modern typing features for improved static checking (mypy, pyright). All features are annotation-only — zero runtime overhead.

TypeAlias

TypeAlias can be used as a service key:

from typing import TypeAlias

DatabaseService: TypeAlias = Database

builder = ContainerBuilder()
builder.service(DatabaseService, make=lambda: Database())
container = builder.build()
db = container.get(DatabaseService)

TypedDict

TypedDict classes are resolvable as dependencies:

from typing import TypedDict

class DBConfig(TypedDict):
    host: str
    port: int

builder = ContainerBuilder()
builder.service(DBConfig, make=lambda: {"host": "localhost", "port": 5432})
container = builder.build()
config = container.get(DBConfig)

ParamSpec factories

Factory protocol and Provider alias accept ParamSpec-typed callables:

from typing import Callable, ParamSpec, TypeVar
from doppy_di import Factory, Provider

P = ParamSpec("P")
T = TypeVar("T")

def provider(factory: Callable[P, T]) -> Callable[P, T]:
    return factory

builder = ContainerBuilder()
builder.service(Database, make=provider(lambda: Database()))

TypeGuard detection

is_injectable() narrows types at runtime:

from doppy_di import injectable, is_injectable

@injectable
class Service:
    pass

if is_injectable(Service):
    # Service is narrowed to type here
    ...

Self fluent builder

ContainerBuilder.service(), value(), and alias() return Self for chaining:

builder = ContainerBuilder()
builder.value("x", 1).service("y", lambda: 2).alias("z", "x")
container = builder.build()

Compile / plan mode

Container.compile() returns an immutable ExecutionPlan that captures a topological ordering of the registered rules. The plan validates the graph up front (raising MissingDependencyError for unregistered dependencies and DependencyCycleError for cycles), then resolves through the live container so lifetimes, singleton caches and scopes keep identical semantics. The feature is fully opt-in: if compile() is never called there is zero overhead.

from doppy_di import CompilePolicy, ContainerBuilder

builder = ContainerBuilder(compile_policy=CompilePolicy.ALLOW_OVERRIDE)
builder.value("a", 1)
builder.service("b", lambda a: a + 1, deps=["a"])

container = builder.build()
plan = container.compile()
plan.get("b")  # 2

# ALLOW_OVERRIDE: overrides still apply through the live container
with container.override("a", 10):
    plan.get("b")  # 11

# STRICT: after compile() the container rejects further overrides
strict = ContainerBuilder(compile_policy=CompilePolicy.STRICT).build()
strict.compile()
# strict.override("a", 1)  # raises RuntimeError

ExecutionPlan.serialize() / ExecutionPlan.deserialize() persist the graph topology, rule metadata and resolved singletons to JSON for caching or cross-process reuse.

When a dependency subtree is fully sync (no yield/async/nested rules) and its transients only depend on singletons or values, the plan builds a flattened resolver at compile time (issue #40): the root closure evaluates each shared singleton once into local bindings, then calls the transient factories directly with those bindings. This removes the per-node loop and the intermediate transient closure frames from the hot path, while preserving singleton identity, transient freshness, override visibility and thread safety. Fallback composed resolvers (kind == "composed") and the generic _resolve_fast walk are used for everything else.

Provider facade

Declarative providers convert to container rules on attribute assignment. They are inert data objects: no resolution logic, zero overhead until assigned. Import from doppy_di.providers; the package-level Factory protocol is untouched.

from doppy_di import Container, Scope
from doppy_di.providers import Factory, Singleton, Value, Resource

services = Container()
services.config = Value({"debug": True})
services.db = Resource(create_db, Scope.APP)
services.repo = Factory(UserRepository, db=services.db)
services.service = Singleton(UserService, repo=services.repo)

Assigning a class factory registers both the named key and a type key. Dependencies may reference providers before assignment; unbound placeholders resolve by name at rule registration.

Provider classes

• Factory — transient factory.
• Singleton — singleton factory.
• Scoped — factory cached per scope.
• Value — constant value.
• Resource — yield-based resource finalized on scope exit.
• Coroutine — async factory.
• Alias — points at another key.
• Dependency — mandatory key; raises on a missing target. Strict alias.
• Selector — picks one provider at resolution time.
• ListOf — aggregates providers into a list.
• DictOf — aggregates named providers into a dict.

Async resolution

Container.aget() resolves sync and async factories, sync and async resources, and resolves independent dependency branches concurrently. Sync factories are called directly with no await overhead. Container.ascope() provides an async scope; async yield providers are finalized on scope exit and on cancellation.

async def make_db():
    return Database("async")

builder.service("db", make_db)
container = builder.build()
db = await container.aget("db")

async with container.ascope("req") as scope:
    session = await scope.aget("session")
# async resources finalized on scope exit

Container.get_many(keys, parallel=True) resolves independent keys concurrently.

Async errors

• AsyncDependencyInSyncContextError — async dependency resolved via sync get().
• SyncFactoryReturningAwaitableError — sync factory returned an awaitable.
• ResolutionCancelledError — aget() cancelled after partially creating resources; partially-created resources are finalized automatically.

Resolution policies

Pluggable policies control the order of dependency resolution. Policies are opt-in; default behaviour is unchanged when none is specified.

from doppy_di import (
    ResolutionChildrenFirstPolicy,
    DefaultResolutionPolicy,
    EagerPolicy,
    ParallelPolicy,
)

# container-wide policy
container = builder.build(policy=ResolutionChildrenFirstPolicy())

# per-call policy
container.get("a", policy=EagerPolicy())

Built-in policies

• DefaultResolutionPolicy — resolve only the requested key (historical behaviour).
• LazyPolicy — same as default; nothing resolved until get().
• ParentFirstPolicy — parents before children.
• ChildrenFirstPolicy — children before parents.
• EagerPolicy — resolve the entire graph up front.
• ParallelPolicy — resolve dependency levels concurrently in aget() (sequential in sync get()).

Implement the ResolutionPolicy protocol (order(graph, root)) for custom strategies.

Name disambiguation

The top-level ChildrenFirstPolicy/ParentFirstPolicy names belong to the devkit nested-field ordering. Resolution policies are exported under the aliases ResolutionChildrenFirstPolicy/ResolutionParentFirstPolicy. Import resolution policies from doppy_di using the aliases, or directly from doppy_di.resolution.

Graph introspection

Container.graph() returns a DependencyGraph for programmatic querying.

g = container.graph()
g.nodes()               # all registered keys
g.edges()               # (key, dependency) pairs
g.dependencies_of("a")  # direct deps
g.dependents_of("a")    # direct dependents
g.to_mermaid()          # mermaid
g.to_dot()              # graphviz
g.to_json()             # dict
g.to_text()             # text tree

DependencyGraph is exported from the package root and doppy_di.graph.

Profiles and child containers

Derive environment-specific containers without mutating the base.

builder.value("env", "base")
container = builder.build()
prod = container.with_profile("prod", {"env": "prod"})

• with_profile(name, overrides) — layered container with profile overrides.
• child(name=None) — container layered over the parent; parent rules added later stay visible.
• diff(other) — returns DiffReport of added/removed/changed keys.
• export_config(format="json") — serialize the effective configuration to JSON.

Tracing

Container.set_tracer() registers a callback invoked after each successful resolution.

def tracer(key, duration, cache_hit, scope):
    ...

container.set_tracer(tracer)
container.set_tracer(None)  # disable

The callback receives (key, duration, cache_hit, scope). With no tracer set there is no timing and no dispatch — zero overhead. Child containers inherit the parent tracer.

OpenTelemetry

pip install "doppy-di[otel]"

from doppy_di.ext.otel import otel_adapter

container.set_tracer(otel_adapter())
container.get("a")  # emits doppy.resolve:'a' span

TracerFn is the callback protocol: Callable[[Key, float, bool, Optional[str]], None].