Skip to content

The provider registry

A registration is a type and a lifetime, and nothing else. app.provide(...) says this contract is satisfied by this factory, owned by this scope, and from then on every Annotated[T, Inject()] parameter in the application — a route handler, a service constructor, a plugin's capability — is filled from it. The registry is also where the framework can still say no cheaply, because a missing dependency and a dependency cycle are both decidable while the application is being built.

What a registration says

examples/learn_extend/registry01_providers.py
"""Walkthrough — the provider registry: what a registration says, and when it is resolved.

Run it with:

    python -m learn_extend.registry01_providers
"""

from __future__ import annotations

import asyncio
from typing import Annotated

from xyberos import App, Inject, Scope, Settings
from xyberos.errors import DependencyCycleError, MissingProviderError


class Clock:
    """An app-scoped capability: one instance, shared by everything that asks for it."""

    def now(self) -> str:
        return "2026-10-12T00:00:00+00:00"


class AuditTrail:
    """A capability whose dependency is declared in its own constructor.

    No factory argument and no manual wiring: the registry reads the annotation at registration time
    and builds the graph, which is also when a missing dependency or a cycle is refused.
    """

    def __init__(self, clock: Annotated[Clock, Inject()]) -> None:
        self._clock = clock
        self.events: list[str] = []

    def record(self, event: str) -> None:
        self.events.append(f"{self._clock.now()} {event}")


class RequestId:
    """A request-scoped capability: one per request context, closed with it."""

    created: int = 0

    def __init__(self) -> None:
        RequestId.created += 1
        self.number = RequestId.created


class First:
    """One half of a dependency cycle, which validation refuses before anything runs."""

    def __init__(self, other: Annotated[Second, Inject()]) -> None: ...


class Second:
    """The other half of the cycle."""

    def __init__(self, first: Annotated[First, Inject()]) -> None:
        self._first = first


async def main() -> None:
    """Register three shapes, read the graph, and watch the two refusals."""
    app = App(settings=Settings.load(environ={}, overrides={"app": {"name": "registry"}}))
    app.provide(Clock, instance=Clock())
    app.provide(AuditTrail)
    app.provide(RequestId, scope=Scope.REQUEST)

    print(app.diagnostics().render().strip())

    async with app.lifespan():
        audit = await app.resolve(AuditTrail)
        audit.record("started")
        print("an app-scoped instance is shared:", audit is await app.resolve(AuditTrail))
        print("its dependency was injected:", audit.events)

        async with app.request_context() as request:
            one = await request.resolve(RequestId)
            two = await request.resolve(RequestId)
            print("one request context, one instance:", one is two)
        async with app.request_context() as request:
            print(
                "a new request context is a new instance:",
                (await request.resolve(RequestId)) is not one,
            )

        try:
            await app.resolve(str)
        except MissingProviderError as error:
            print("an unregistered contract:", error)

    cyclic = App()
    cyclic.provide(First)
    cyclic.provide(Second)
    try:
        cyclic.validate()
    except DependencyCycleError as error:
        print("a cycle is refused at validation:", error)


if __name__ == "__main__":
    asyncio.run(main())

Clock is registered as an instance, AuditTrail as a class — which makes the class its own factory — and RequestId at request scope. AuditTrail's dependency is declared in its constructor, so no factory argument and no manual wiring is involved: the annotation is read at registration time, and the graph is built from what every registration declares.

providers:
  Clock [app] from application
  AuditTrail [app] from application
  RequestId [request] from application
plugins:
  (none)
an app-scoped instance is shared: True
its dependency was injected: ['2026-10-12T00:00:00+00:00 started']
one request context, one instance: True
a new request context is a new instance: True
an unregistered contract: no provider is registered for str
a cycle is refused at validation: dependency cycle detected: First -> Second -> First

The providers: block is not decoration: it names the contract, the scope it was registered with, and where the registration came from. from application is the default source, and a provider registered inside a plugin's register hook reports from plugin:<name> instead, so a surprising override is attributable to the code that made it.

Scope Who owns the instance When it is built When it is closed
Scope.APP the application at startup, or at first resolution unless eager=True application shutdown, in reverse order
Scope.REQUEST one request context on first resolution inside that context when that context exits
Scope.TRANSIENT the caller at every resolution nobody — so it may not return a managed resource

Rules the registry enforces at registration rather than at first use:

  • One registration per contract. A second provide for the same contract is a RegistrationError unless it is explicit: replace=True.
  • Instance or factory, not both. Passing an instance and a factory is refused; passing a non-class contract with no factory is refused too, because there is nothing to build.
  • eager is app-scoped only. A request-scoped eager provider is an argument about a lifetime that does not exist yet.
  • Injected parameters are annotated. Annotated[T, Inject()], at most one marker per parameter. An unannotated parameter on a factory or constructor is a mistake the registry reports instead of guessing.
  • A transient provider may not return a managed resource. Nobody would own closing it, so the runtime refuses the instance and tells you to use request or app scope.

validate() runs the whole graph before anything starts, which is why the cycle above is reported as a path — First -> Second -> First — rather than as a stack overflow in the middle of a request. Forward references are legal, because a class may name a dependency that is defined later in the module.

Replacing a provider

examples/learn_extend/registry02_overrides.py
"""Walkthrough — replacing a provider: what a test does, and what a deployment does.

Run it with:

    python -m learn_extend.registry02_overrides
"""

from __future__ import annotations

import asyncio
from typing import Annotated

from xyberos import App, Inject
from xyberos.errors import MissingProviderError


class Rates:
    """A capability that reads something the application does not control."""

    def __init__(self, table: dict[str, float]) -> None:
        self._table = table

    def discount(self, code: str) -> float:
        return self._table.get(code, 0.0)


class Pricing:
    """A service that depends on the rates and knows nothing about where they came from."""

    def __init__(self, rates: Annotated[Rates, Inject()]) -> None:
        self._rates = rates

    def quote(self, code: str, price: float) -> float:
        return round(price * (1 - self._rates.discount(code)), 2)


def production() -> App:
    """Compose the application the way a deployment does: the real rates, from configuration."""
    app = App()
    app.provide(Rates, instance=Rates({"SAVE10": 0.10}))
    app.provide(Pricing)
    return app


async def main() -> None:
    """Quote once for real, once for a test, and once for a deployment that changed its mind."""
    app = production()
    async with app.lifespan():
        pricing = await app.resolve(Pricing)
        print("the application's answer:", pricing.quote("SAVE10", 100.0))

        under_test = app.with_overrides({Rates: Rates({"SAVE10": 0.5})})
        async with under_test.lifespan():
            tested = await under_test.resolve(Pricing)
            print("the same code, a test's answer:", tested.quote("SAVE10", 100.0))
            print("the test app is a clone, not a restart:", under_test is not app)

        print("and the original is untouched:", pricing.quote("SAVE10", 100.0))

    replacement = production()
    replacement.provide(Rates, instance=Rates({"SAVE10": 0.25}), replace=True)
    async with replacement.lifespan():
        print(
            "a registration can be replaced before startup:",
            (await replacement.resolve(Pricing)).quote("SAVE10", 100.0),
        )

    try:
        app.with_overrides({str: "no such provider"})
    except MissingProviderError as error:
        print("overriding something unregistered:", error)


if __name__ == "__main__":
    asyncio.run(main())

What a test needs is not a restart but a different answer from one contract, which is exactly what with_overrides is for.

the application's answer: 90.0
the same code, a test's answer: 50.0
the test app is a clone, not a restart: True
and the original is untouched: 90.0
a registration can be replaced before startup: 75.0
overriding something unregistered: cannot override str because it is not registered

Three properties matter and all three are asserted above:

  • A clone, not a mutation. with_overrides returns a new application that shares the original's registrations and replaces the named ones. The original keeps its answer, so a test cannot change a fixture another test is using.
  • A clone is not a restart. The returned application has not been started, and its lifecycle is its own; when it is stopped, the providers it built are closed and the original's are untouched.
  • Only what exists can be overridden. A contract that was never registered is refused (MissingProviderError) rather than quietly added — otherwise a typo in a test would silently change nothing, and the test would pass for the wrong reason.

replace=True on provide is the deployment-side cousin: a settings-driven composition can register a default and then replace it with the configured implementation before startup. After start() neither route is available; the application is immutable from that moment, which is what makes a diagnosis about it meaningful.

  • Writing a module — the composable unit that registers capabilities and routes.
  • Writing a plugin — the same registrations, attributable to a named, versioned package.
  • Dependency injection — the exhaustive reference for scopes, factories, and lifecycle hooks.