API reference

Everything public is importable straight from action0.service:

from action0.service import Registry, Scope, Named, Ref, injected

The modules below are where the pieces live; the module paths matter only when subclassing internals (custom ScopePolicy implementations) or reading tracebacks.

action0.service.registry

The service registry: registration, lookup, injection, scopes, lifecycle.

A Registry maps types and names to service definitions and builds instances on demand, resolving missing constructor parameters from its own registrations:

>>> from action0.service import Registry
>>> class Engine:
...     def __init__(self, url: str = "sqlite://"):
...         self.url = url
>>> class Repository:
...     def __init__(self, engine: Engine):
...         self.engine = engine
>>> registry = Registry()
>>> _ = registry.register(Engine, params={"url": "postgres://db/app"})
>>> _ = registry.register(Repository)
>>> repository = registry.get(Repository)
>>> repository.engine.url
'postgres://db/app'
>>> registry.get(Repository) is repository  # singletons by default
True

class action0.service.registry.Registry(*, parent=None, profiles=None)

Bases: object

Container that registers service definitions and resolves instances.

Registrations map a provider (class, factory callable, or ready-made instance) to the type it provides, optionally under a name. Lookups go by type (subclass-aware) or by name; missing constructor parameters are injected from the registry based on type annotations.

Instances live according to their Scope. A registry can have a parent: lookups fall back to the parent, local registrations shadow it — useful for request-scoped wiring and tests.

Registrations may be limited to profiles (e.g. dev / prod); definitions whose profiles do not intersect the registry’s active profiles are invisible to every lookup. See profiles.

Registries are context managers; leaving the with block calls close(), which disposes managed instances in reverse creation order. Async applications use the a-prefixed twins instead — aget(), abuild(), awarmup(), aclose(), async with — which additionally support async def factories; one registry is meant to be driven from a single event loop.

property profiles: frozenset[str]

The active profiles this registry was created with (immutable).

register(provider, *, name=None, scope=Scope.SINGLETON, params=None, provides=None, default=None, eager=False, profiles=None, replace=False)

Register a class or factory callable as a service.

Parameters:

• provider (type[TypeVar(_T)] | Callable[..., TypeVar(_T)]) – the class to instantiate, or a factory callable whose return annotation tells the provided type

• name (str | None, default: None) – optional service name; unnamed registrations are the default implementation for their type

• scope (Scope | str, default: <Scope.SINGLETON: 'singleton'>) – instance lifetime, one of the built-in Scope members / their string values or a custom scope key (default: singleton)

• params (Mapping[str, Any] | None, default: None) – constructor parameters to apply; values may contain Ref markers referencing other services

• provides (type[Any] | None, default: None) – the type to register under; defaults to the class itself or the factory’s return annotation; may be a runtime-checkable typing.Protocol the provider satisfies structurally

• default (bool | None, default: None) – whether this definition wins ambiguous type lookups; defaults to True for unnamed and False for named services

• eager (bool, default: False) – instantiate this service in warmup()

• profiles (Iterable[str] | None, default: None) – limit this definition to the given profiles (a bare string counts as one profile); it is only visible in registries whose profiles intersect. Default: active everywhere

• replace (bool, default: False) – overwrite a colliding registration instead of raising DuplicateServiceError

Return type:

Definition

Returns:

the stored Definition

Raises:

• DefinitionError – if the provider is not callable, the provided type cannot be inferred, or provides does not match

• ScopeError – if scope names no registered scope

• DuplicateServiceError – on collisions without replace

register_instance(instance, *, name=None, provides=None, default=None, profiles=None, replace=False)

Register an already-built object as a service.

The instance is served as-is (singleton semantics). Because the registry did not create it, close() will not dispose it.

Parameters:

• instance (Any) – the object to serve

• name (str | None, default: None) – optional service name

• provides (type[Any] | None, default: None) – the type to register under; defaults to type(instance); may be a runtime-checkable typing.Protocol the instance satisfies (verified with isinstance, non-method members included)

• default (bool | None, default: None) – whether this definition wins ambiguous type lookups; defaults to True for unnamed and False for named services

• profiles (Iterable[str] | None, default: None) – limit this definition to the given profiles; see register()

• replace (bool, default: False) – overwrite a colliding registration instead of raising

Return type:

Definition

Returns:

the stored Definition

Raises:

• DefinitionError – if instance is not an instance of provides

• DuplicateServiceError – on collisions without replace

service(name=None, /, *, scope=Scope.SINGLETON, params=None, provides=None, default=None, eager=False, profiles=None, replace=False)

Class/factory decorator form of register().

Overloads:

• self, name (_C) → _C

• self, name (str | None), scope (Scope | str), params (Mapping[str, Any] | None), provides (type[Any] | None), default (bool | None), eager (bool), profiles (Iterable[str] | None), replace (bool) → Callable[[_C], _C]

Works bare (@registry.service) or with arguments (@registry.service("mailer.bulk", scope=Scope.THREAD)); the decorated class or factory is returned unchanged.

Parameters:

• name (Any, default: None) – the service name, or — in the bare form — the decorated class/factory itself

• scope (Scope | str, default: <Scope.SINGLETON: 'singleton'>) – see register()

• params (Mapping[str, Any] | None, default: None) – see register()

• provides (type[Any] | None, default: None) – see register()

• default (bool | None, default: None) – see register()

• eager (bool, default: False) – see register()

• profiles (Iterable[str] | None, default: None) – see register()

• replace (bool, default: False) – see register()

Returns:

the decorated object, or the decorator to apply

register_scope(key, policy)

Register a custom scope under key (or replace a built-in one).

Parameters:

• key (str) – the scope key used in definitions (e.g. "request")

• policy (ScopePolicy) – the ScopePolicy managing instances for this scope

Return type:

None

load_yaml(source, *, replace=False, lazy=False)

Load service definitions from a YAML file (requires PyYAML).

See action0.service.loader for the accepted format.

With lazy=True the factory and provides dotted paths are not imported at load time. Each definition imports them on first use: when the service is built (by-name lookups import nothing else), when any type-based lookup consults the registry layer (type scans need the real provided types, so they import all still-lazy definitions of that layer), and during validate() and warmup(). Import errors then surface as DefinitionError naming the service — call validate() at boot to collect them early.

Parameters:

• source (str | PathLike[str] | IO[str]) – a file path, or an open text stream (e.g. io.StringIO) containing the YAML document

• replace (bool, default: False) – overwrite colliding registrations instead of raising

• lazy (bool, default: False) – defer importing factory paths until first use

Return type:

list[Definition]

Returns:

the definitions that were registered, in file order

Raises:

• ServiceError – if PyYAML is not installed

• DefinitionError – if the document is malformed

load_entry_points(group, *, replace=False)

Register services advertised by installed packages via entry points.

Every entry point in group is loaded and applied with one of two conventions, decided by what the entry point resolves to:

• a setup hook — a plain function with exactly one required parameter — is called with this registry and may register any number of services itself;

• anything else (a class or factory callable) is registered under the entry point’s name, exactly like register(obj, name=entry_point.name).

Plugins advertise themselves in their pyproject.toml:

[project.entry-points."myapp.services"]
blob-storage = "myapp_blob.storage:BlobStorage"
extras = "myapp_extras.plugin:setup"

Note that a factory function with exactly one required parameter is indistinguishable from a setup hook — give the parameter a default, or use a class or setup hook instead.

Parameters:

• group (str) – the entry-point group to scan (e.g. "myapp.services")

• replace (bool, default: False) – overwrite colliding registrations instead of raising

Return type:

list[Definition]

Returns:

the definitions registered by all entry points, in load order (for setup hooks: every definition the hook added)

Raises:

DefinitionError – if an entry point fails to load, resolves to an unusable object, or its registration fails; the error names the entry point and its distribution

get(key, *, name=None)

Return the service instance for a type or name.

Overloads:

• self, key (type[_T]), name (str | None) → _T

• self, key (str) → Any

Type lookups are subclass-aware: a service registered as PostgresDb also answers get(Database). Requesting a runtime-checkable typing.Protocol matches structurally: every registration whose provided type satisfies the protocol is a candidate. With several candidates, the (single) one marked default wins; otherwise an exact type match; otherwise AmbiguousServiceError is raised. Lookups not satisfied locally fall back to the parent registry.

Parameters:

• key (type[TypeVar(_T)] | str) – the requested type, or a service name

• name (str | None, default: None) – with a type key: request the service with this name and verify it provides the requested type

Returns:

the service instance, built and cached per its scope

Raises:

• ServiceNotFoundError – if nothing matches

• AmbiguousServiceError – if several services match a type request and none is clearly the default

find(key, *, name=None)

Like get(), but return None when nothing matches.

Overloads:

• self, key (type[_T]), name (str | None) → _T | None

• self, key (str) → Any

Ambiguity still raises — an ambiguous request is a configuration problem, not an absence.

Parameters:

• key (type[TypeVar(_T)] | str) – the requested type, or a service name

• name (str | None, default: None) – with a type key: request the service with this name

Returns:

the service instance, or None if nothing is registered

get_all(key)

Return instances of all services providing key.

Parent registrations come first, then local ones, in registration order — handy for plugin-style multi-registrations.

Parameters:

key (type[TypeVar(_T)]) – the requested type

Return type:

list[TypeVar(_T)]

Returns:

one instance per matching definition (may be empty)

build(provider, /, **params)

Construct an instance with injection without registering it.

Useful for one-off objects that want their dependencies wired up: missing constructor parameters are resolved exactly as for registered services; params override everything.

Parameters:

• provider (type[TypeVar(_T)] | Callable[..., TypeVar(_T)]) – the class or factory to call

• params (Any) – explicit parameter values

Return type:

TypeVar(_T)

Returns:

the new instance (never cached)

inject(func)

Decorate a function so parameters defaulting to injected are resolved.

Only parameters whose default is the injected sentinel (or that are explicitly passed as injected) are filled in — the signature stays honest for callers and type checkers:

@registry.inject
def send_report(report: str, mailer: Mailer = injected) -> None: ...


send_report("weekly")  # mailer resolved from the registry

On an async def function the wrapper is itself a coroutine function and resolves the sentinel parameters through the async paths — async services can be injected into async functions.

Parameters:

func (Callable[[ParamSpec(_P, bound= None)], TypeVar(_R)]) – the function to wrap

Return type:

Callable[[ParamSpec(_P, bound= None)], TypeVar(_R)]

Returns:

a wrapper with the same signature

Raises:

InjectionError – at call time, if a sentinel parameter cannot be resolved

override(key, instance)

Temporarily replace a service with instance (for tests).

While the with block is active, lookups for key — by type or by name, including injections into other services being built — return instance. Two caveats: instances cached before the override (e.g. already-built singletons that had the real service injected) are not rewritten, and conversely a singleton first built during the override keeps the replacement — in tests, prefer a fresh registry (or a child registry) per test over overriding in a long-lived one. Overrides apply where they are declared: services owned by a parent registry build against the parent’s world, so override on the registry that owns the service.

Parameters:

• key (type[Any] | str) – the type or service name to replace

• instance (Any) – the replacement object

Return type:

Iterator[Any]

Returns:

a context manager yielding instance

warmup()

Instantiate every definition registered with eager=True.

Call this at application boot to fail fast and to pay construction costs up front. Definitions inactive under the registry’s profiles are skipped.

Return type:

list[Any]

Returns:

the instances that were created (or already cached)

validate()

Statically check every local definition without instantiating anything.

Detects: unknown params keys, required parameters that neither params, defaults, nor any registration can satisfy, dangling Ref targets, ambiguous injections, and dependency cycles. Definitions inactive under the registry’s profiles are skipped — they could never be built here.

Raises:

ValidationError – listing all problems found

Return type:

None

close()

Dispose managed instances and shut the registry down.

Every scope hands over the instances it can see from the calling thread and context; instances the registry created (not those from register_instance()) get their close() method called if they have one, dependents before dependencies. Errors are logged, not raised. After closing, any use of the registry raises ServiceError. Instances that only offer an async aclose() method cannot be disposed here — they are skipped with a warning; use aclose() for those.

Return type:

None

async aget(key, *, name=None)

Async get(): also resolves async def factories.

Overloads:

• self, key (type[_T]), name (str | None) → _T

• self, key (str) → Any

Lookup rules are identical to get(); the difference is the build path: async providers are awaited, and the dependencies of an async build may themselves be async. Sync definitions resolve fine through here too (and share their caches with the sync methods), so async code can use aget() throughout. An async registry is meant to be driven from a single event loop.

Parameters:

• key (type[TypeVar(_T)] | str) – the requested type, or a service name

• name (str | None, default: None) – with a type key: request the service with this name

Returns:

the service instance, built and cached per its scope

Raises:

• ServiceNotFoundError – if nothing matches

• AmbiguousServiceError – if several services match a type request and none is clearly the default

async afind(key, *, name=None)

Async find(): like aget(), but None when absent.

Overloads:

• self, key (type[_T]), name (str | None) → _T | None

• self, key (str) → Any

Parameters:

• key (type[TypeVar(_T)] | str) – the requested type, or a service name

• name (str | None, default: None) – with a type key: request the service with this name

Returns:

the service instance, or None if nothing is registered

async aget_all(key)

Async get_all(): instances of all services providing key.

Parameters:

key (type[TypeVar(_T)]) – the requested type

Return type:

list[TypeVar(_T)]

Returns:

one instance per matching definition (may be empty)

async abuild(provider, /, **params)

Async build(): one-off construction with async injection.

provider may be an async def factory (its result is awaited) or any sync provider whose dependencies include async services.

Parameters:

• provider (type[TypeVar(_T)] | Callable[..., TypeVar(_T)]) – the class or factory to call

• params (Any) – explicit parameter values

Return type:

TypeVar(_T)

Returns:

the new instance (never cached)

async awarmup()

Async warmup(): eagerly build definitions, async ones included.

Return type:

list[Any]

Returns:

the instances that were created (or already cached)

async aclose()

Async close(): also awaits aclose() disposal methods.

Teardown follows the same rules as close(), except that a managed instance with an aclose() method gets that awaited in preference to a sync close(). Errors are logged, not raised.

Return type:

None

definitions()

Return a snapshot of this registry’s own definitions (parents excluded).

Return type:

tuple[Definition, ...]

Returns:

the definitions in registration order

action0.service.scopes

Service scopes: how long a built instance is kept and who shares it.

The built-in scopes are addressed through the Scope enum (or their string values, e.g. in YAML files):

• singleton — one instance per registry, shared by everyone (the default)

• transient — a fresh instance for every request

• thread — one instance per thread

• context — one instance per contextvars context (which makes it task-local under asyncio)

Custom scopes are plain ScopePolicy subclasses registered with action0.service.registry.Registry.register_scope().

Every policy also has an async twin, ScopePolicy.aget(), used by the a-prefixed registry methods. The built-in caching scopes dedupe concurrent first builds with asyncio.Lock instances; an async registry is meant to be driven from a single event loop.

class action0.service.scopes.Scope(*values)

Bases: Enum

Names of the built-in scopes.

Everywhere a scope is expected, the enum member and its string value ("singleton", "transient", "thread", "context") are interchangeable; custom scopes are addressed by their registered string.

class action0.service.scopes.ScopePolicy

Bases: ABC

Strategy deciding whether to reuse a stored instance or build a new one.

A policy instance belongs to exactly one Registry and holds the instances for every definition using its scope.

caches: bool = True

Whether instances are stored and shared.

Caching scopes build their instances in the context of the registry that owns the definition, so a shared instance can never capture registrations of the (possibly short-lived) registry that happened to request it first. Set to False for scopes that build fresh instances every time; those resolve dependencies through the requesting registry.

abstractmethod get(definition, build)

Return the instance for definition, building it if necessary.

Parameters:

• definition (Definition) – the definition being resolved (usable as a dict key; definitions hash by identity)

• build (Callable[[], Any]) – zero-argument callable producing a new instance

Return type:

Any

Returns:

the (new or cached) service instance

async aget(definition, abuild)

Async twin of get(), used by the registry’s a methods.

The default implementation only fits non-caching scopes (it builds fresh every time); caching scopes must override it so that stored instances are found and concurrent first builds are deduplicated.

Parameters:

• definition (Definition) – the definition being resolved

• abuild (Callable[[], Awaitable[Any]]) – zero-argument coroutine function producing a new instance

Return type:

Any

Returns:

the (new or cached) service instance

Raises:

ScopeError – on a caching policy that did not override this

drain()

Hand over all stored instances visible from the calling thread and context for disposal and forget them.

Return type:

list[tuple[Definition, Any]]

Returns:

(definition, instance) pairs, in reverse creation order (dependents before their dependencies)

class action0.service.scopes.SingletonScope

Bases: ScopePolicy

One shared instance per registry.

get(definition, build)

Return the stored instance, building it under the creation lock once.

Return type:

Any

async aget(definition, abuild)

Return the stored instance, building it at most once per definition.

Concurrent first requests from several tasks are deduplicated with a lazily created per-definition asyncio.Lock. The threading creation lock is only held for the lock bookkeeping — never across an await — so async builds cannot stall other threads.

Return type:

Any

drain()

Return all singletons in reverse creation order and clear the store.

Return type:

list[tuple[Definition, Any]]

class action0.service.scopes.TransientScope

Bases: ScopePolicy

A fresh instance on every request; nothing is stored (or disposed).

caches: bool = False

Whether instances are stored and shared.

Caching scopes build their instances in the context of the registry that owns the definition, so a shared instance can never capture registrations of the (possibly short-lived) registry that happened to request it first. Set to False for scopes that build fresh instances every time; those resolve dependencies through the requesting registry.

get(definition, build)

Build a new instance every time.

Return type:

Any

class action0.service.scopes.ThreadScope

Bases: ScopePolicy

One instance per thread (backed by threading.local).

get(definition, build)

Return the calling thread’s instance, building it on first use.

Return type:

Any

async aget(definition, abuild)

Return the calling thread’s instance, building it at most once.

All tasks of one event loop run on the loop’s thread and therefore share this thread’s store, so concurrent first requests are deduplicated with per-definition asyncio.Lock objects kept in the same thread-local (no threading lock needed: only this thread ever touches them).

Return type:

Any

drain()

Return the calling thread’s instances (other threads’ survive).

Return type:

list[tuple[Definition, Any]]

class action0.service.scopes.ContextScope

Bases: ScopePolicy

One instance per contextvars context (task-local in asyncio).

Standard contextvars semantics apply: a context copied (or an asyncio task spawned) after an instance was built inherits that instance; instances built inside a copy stay inside it.

get(definition, build)

Return the current context’s instance, building it on first use.

Return type:

Any

async aget(definition, abuild)

Return the current context’s instance, building it on first use.

No deduplication lock is needed: every asyncio task runs in its own context copy, so concurrent tasks build (and keep) their own instances — that is exactly the scope’s task-local semantics.

Return type:

Any

drain()

Return the instances visible in the current context and unset them.

Return type:

list[tuple[Definition, Any]]

action0.service.default

A process-wide default registry, for applications that want one.

Libraries and composable code should pass Registry instances around explicitly; but in an application or script there is often exactly one registry anyway, and threading it through every call site is ceremony. default_registry() provides that single instance on demand:

>>> from action0.service import Registry
>>> from action0.service import default_registry
>>> from action0.service import set_default_registry
>>> class Config:
...     def __init__(self, env: str = "prod"):
...         self.env = env
>>> _ = default_registry().register(Config)
>>> default_registry().get(Config).env
'prod'
>>> _ = set_default_registry(None)  # reset so unrelated code starts fresh

The default registry is created lazily on first access and is never closed automatically — closing it (and when) is the application’s responsibility. After closing it, call set_default_registry(None) so the next access creates a fresh instance.

action0.service.default.default_registry()

Return the process-wide default registry, creating it on first access.

Return type:

Registry

Returns:

the shared Registry instance (the same one on every call, until it is replaced with set_default_registry())

action0.service.default.set_default_registry(registry)

Install registry as the process-wide default.

Passing None resets the default, so the next default_registry() call creates a fresh instance — do this after closing the previous default.

The previous instance is returned but not closed; dispose it yourself if it held resources.

Parameters:

registry (Registry | None) – the new default registry, or None to reset

Return type:

Registry | None

Returns:

the previously installed registry, or None

action0.service.default.using_default_registry(registry)

Temporarily install registry as the process-wide default.

The previous default (or the not-yet-created state) is restored when the with block ends, even on error — the pattern for tests that exercise code relying on default_registry():

with Registry() as registry, using_default_registry(registry):
    registry.register_instance(FakeMailer(), provides=Mailer)
    code_under_test()

Note this swaps process-global state: tests doing this cannot run concurrently with other tests that touch the default registry.

Parameters:

registry (Registry) – the registry to install for the duration

Return type:

Iterator[Registry]

Returns:

a context manager yielding registry

action0.service.markers

Marker objects used to steer injection: Named, Ref, injected.

class action0.service.markers.Named(name)

Bases: object

Qualifier for type annotations: inject the service registered under a specific name.

Use it inside typing.Annotated when several services provide the same type and the parameter needs a particular one:

def __init__(self, db: Annotated[Database, Named("replica")]) -> None: ...

name: str

The service name to resolve.

class action0.service.markers.Ref(key)

Bases: object

Late-bound reference to another service, usable as a parameter value.

Put a Ref into the params mapping of a registration (or use the !ref tag in YAML) and it is replaced with the referenced service when the depending service is built:

registry.register(ReportJob, params={"db": Ref("db.replica")})

Parameters:

key (str | type[Any]) – the service name, or a type to resolve the default implementation for.

key: str | type[Any]

The service name or type to resolve when the value is needed.

action0.service.markers.injected: Any = <injected>

Default value marking a parameter to be filled in by action0.service.registry.Registry.inject(). Typed as Any so it can be used as the default for a parameter of any annotated type:

@registry.inject
def send_report(report: str, mailer: Mailer = injected) -> None: ...

action0.service.definitions

Service definitions and provider introspection.

A Definition is the stored form of one registration: the provider (class, factory, or a closure over a ready-made instance), what type it provides, its optional name, scope, and configured parameters.

The module also contains the reflection helpers the registry uses to decide what can be injected: provider_spec() extracts a provider’s parameters (with resolved type hints), and unwrap_annotation() normalizes an annotation into its core type plus optionality and an optional Named qualifier.

class action0.service.definitions.Definition(provider, provides, name, scope, params=<factory>, default=False, eager=False, profiles=frozenset({}), managed=True, introspect=True, factory_path=None, provides_path=None)

Bases: object

One registered service: provider, provided type, name, scope, parameters.

Definitions compare and hash by identity (eq=False), so they can be used as dictionary keys in scope stores.

provider: Callable[[...], Any] | None

The class or factory callable that produces the service instance.

None only while a lazily-loaded YAML definition has not been materialize()d yet; use resolved_provider() to read it safely.

provides: type[Any]

The type this service is registered under (used for type lookups).

For unmaterialized lazy definitions this is the object placeholder; type scans materialize before reading it.

name: str | None

The service name, or None for an unnamed (default) registration.

scope: str

Key of the scope policy governing the instance lifetime.

params: dict[str, Any]

Configured constructor parameters (may contain Ref markers).

default: bool = False

Whether this definition wins ambiguous type lookups.

eager: bool = False

Whether warmup() instantiates it.

profiles: frozenset[str] = frozenset({})

Profiles under which this definition is active (empty = always active).

managed: bool = True

Whether the registry created the instance and may dispose it on close.

introspect: bool = True

Whether the provider’s signature is inspected for injection.

factory_path: str | None = None

Unimported dotted path of a lazily-loaded factory, None otherwise.

provides_path: str | None = None

Unimported dotted path of a lazily-loaded provides type, if any.

is_async: bool = False

Whether the provider is a coroutine function (async def factory).

Async definitions can only be resolved through the a-prefixed registry methods (aget() and friends); the sync paths refuse them with a clear error. Re-detected by materialize() for lazy definitions, whose provider is not known at construction time.

label()

Return a short human-readable identifier for error messages.

Return type:

str

Returns:

the provided type’s name (the factory path while a lazy definition is unmaterialized), plus the service name if set

materialize()

Import a lazily-loaded factory (and provides) path, once.

A no-op for definitions that already carry a provider. Idempotent and thread-safe: the import happens under the same process-wide creation lock the scopes use, so no lock-ordering issues can arise.

Raises:

DefinitionError – if a dotted path cannot be imported, the factory is not callable, or the provided type cannot be determined — prefixed with the service name for context

Return type:

None

resolved_provider()

Return the provider, importing lazily-loaded paths first.

Return type:

Callable[..., Any]

Returns:

the class or factory callable

Raises:

DefinitionError – if a lazy path cannot be imported

class action0.service.definitions.AnonymousFactory(definition)

Bases: object

A nested, unregistered definition used as a parameter value.

Produced by the YAML loader for mappings that contain a factory key inside another service’s parameters; the wrapped definition is built fresh every time the owning service is built.

definition: Definition

The unregistered definition to build when the parameter is resolved.

class action0.service.definitions.ProviderParameter(name, positional_only, has_default, annotation)

Bases: object

One injectable parameter of a provider’s signature.

name: str

The parameter name.

positional_only: bool

Whether the parameter can only be passed positionally.

has_default: bool

Whether the provider declares a default for this parameter.

annotation: Any

The resolved type hint, or None if the parameter has none.

class action0.service.definitions.ProviderSpec(parameters, has_var_keyword, introspectable)

Bases: object

The injectable shape of a provider: parameters and **kwargs presence.

parameters: tuple[ProviderParameter, ...]

All positional/keyword parameters (*args/**kwargs excluded).

has_var_keyword: bool

Whether the provider accepts arbitrary keyword arguments.

introspectable: bool

False when the signature could not be determined (C builtins).

action0.service.definitions.provider_spec(provider)

Return the (cached) ProviderSpec for a class or factory.

Parameters:

provider (Callable[..., Any]) – the class or callable to inspect

Return type:

ProviderSpec

Returns:

the provider’s injectable parameters; for providers whose signature cannot be determined, a spec with introspectable=False

action0.service.definitions.infer_provides(provider)

Infer the provided type: the class itself, or a factory’s return annotation.

Parameters:

provider (Callable[..., Any]) – the class or factory callable

Return type:

type[Any] | None

Returns:

the provided type, or None if it cannot be inferred

action0.service.definitions.is_protocol(tp)

Return whether tp is a typing.Protocol class.

Parameters:

tp (type[Any]) – the type to test

Return type:

bool

Returns:

True for protocol classes, False for nominal classes

action0.service.definitions.is_runtime_checkable(tp)

Return whether protocol tp is decorated with typing.runtime_checkable().

Parameters:

tp (type[Any]) – the protocol class to test

Return type:

bool

Returns:

whether isinstance/issubclass checks are allowed on it

action0.service.definitions.check_requested_type(requested)

Verify that requested is usable as a type-lookup key.

Parameters:

requested (type[Any]) – the requested type

Raises:

ServiceError – if requested is a protocol that is not decorated with typing.runtime_checkable() — structural matching relies on issubclass, which such protocols refuse

Return type:

None

action0.service.definitions.matches_type(provides, requested)

Return whether a definition providing provides satisfies requested.

Nominal classes match by issubclass(). When requested is a runtime-checkable typing.Protocol, matching is structural: any provided type whose members satisfy the protocol matches, no inheritance required.

Parameters:

• provides (type[Any]) – the type a definition provides

• requested (type[Any]) – the requested type (a class or a runtime-checkable protocol)

Return type:

bool

Returns:

whether the definition satisfies the request

Raises:

ServiceError – if requested is a protocol that is not runtime-checkable, or one with non-method members (which issubclass cannot verify on a class)

action0.service.definitions.unwrap_annotation(annotation)

Normalize a type annotation for injection.

Peels typing.Annotated layers (collecting a Named qualifier if present) and unwraps X | None / Optional[X] into X plus an optional flag.

Parameters:

annotation (Any) – the annotation to normalize (may be None)

Return type:

tuple[Any, bool, str | None]

Returns:

(core, optional, named) where core is the remaining annotation (None if nothing injectable remains, e.g. for multi-type unions), optional tells whether None is allowed, and named is the qualifier name if one was attached

action0.service.errors

Exception hierarchy for action0.service.

Every exception raised by this package derives from ServiceError, so except ServiceError catches anything the framework can throw.

exception action0.service.errors.ServiceError

Bases: Exception

Base class for all errors raised by action0.service.

exception action0.service.errors.DefinitionError

Bases: ServiceError

A service definition is invalid.

Raised when a registration cannot be accepted: the provider is not callable, provides does not match the provider, a YAML definition is malformed, or configured parameters do not exist on the provider.

exception action0.service.errors.DuplicateServiceError

Bases: DefinitionError

A registration collides with an existing one.

Either a service with the same name already exists, or an unnamed (default) service for the same provides type is already registered. Pass replace=True to the registration call to overwrite instead.

exception action0.service.errors.ServiceNotFoundError

Bases: ServiceError

No registered service matches the requested type or name.

exception action0.service.errors.AmbiguousServiceError

Bases: ServiceError

More than one registered service matches a type request.

Disambiguate by requesting the service by name, or mark exactly one of the candidates as the default implementation (default=True).

exception action0.service.errors.InjectionError

Bases: ServiceError

A constructor or function parameter could not be resolved.

Raised when a required parameter has no configured value, no usable default, and no registered service matching its type annotation.

exception action0.service.errors.CircularDependencyError

Bases: ServiceError

Two or more services depend on each other, directly or indirectly.

exception action0.service.errors.ScopeError

Bases: ServiceError

A definition references a scope that is not registered.

exception action0.service.errors.ValidationError(problems)

Bases: ServiceError

action0.service.registry.Registry.validate() found problems.

The message lists every problem found, one per line.

action0.service.loader

Load service definitions from YAML files (requires PyYAML).

The document is a mapping of service names to definitions. Within a definition, a handful of reserved keys configure the registration; every other key is a constructor parameter:

mailer.bulk:
  factory: myapp.mail.SmtpMailer      # dotted path — required
  scope: singleton                    # optional (default: singleton)
  provides: myapp.mail.Mailer         # optional, defaults to the class
  default: true                       # optional: wins type lookups
  eager: false                        # optional: built by warmup()
  profiles: [prod]                    # optional: active only under these
                                      # profiles (string or list)
  api_key: !ENV ${MAILER_KEY}         # everything else: init params
  db: !ref database                   # inject another service by name
  retry_policy:                       # nested mapping with "factory":
    factory: myapp.util.Retry         # built fresh as an anonymous object
    attempts: 3

Supported YAML conveniences:

• !ENV substitutes ${VAR} / ${VAR:-fallback} from the process environment inside a scalar, at load time.

• !ref name injects the service registered under name at build time.

• Standard YAML anchors and merge keys (&base / <<: *base) work as usual for sharing configuration between definitions. Entries whose key starts with a . are templates: they are parsed (so their anchors can be referenced) but not registered.

• !ENV substitution is purely textual — values arrive as strings, they are not re-parsed as YAML (so a secret like "yes" or "0123" cannot change type behind your back).

• If a constructor parameter is itself named like a reserved key, put it under the params: mapping, which is passed through verbatim.

• With lazy=True the factory/provides dotted paths (including those of nested anonymous factories) are not imported at load time but on first use — see load_yaml().

Parsing uses a yaml.SafeLoader subclass, so documents cannot instantiate arbitrary Python objects during parsing — but factory paths are imported and called, so only load files you trust.

action0.service.loader.import_from_path(path)

Import an object from a dotted path like myapp.mail.SmtpMailer.

The longest importable module prefix is imported, the remaining segments are resolved with getattr (so nested classes work too).

Parameters:

path (str) – the dotted path

Return type:

Any

Returns:

the imported object

Raises:

DefinitionError – if the path cannot be resolved

action0.service.loader.load(registry, source, *, replace=False, lazy=False)

Parse a YAML document and register every service it defines.

Parameters:

• registry (Registry) – the registry to register into

• source (str | PathLike[str] | IO[str]) – a file path, or an open text stream containing YAML

• replace (bool, default: False) – overwrite colliding registrations instead of raising

• lazy (bool, default: False) – defer importing factory/provides paths until the definitions are first used

Return type:

list[Definition]

Returns:

the registered definitions, in document order

Raises:

DefinitionError – if the document or a definition is malformed