orionis.container.container.Container

THE ORIONIS API

Build with clarity.

Explore the building blocks of an async-first Python framework. Every module, class, and method — connected, searchable, and ready to build with.

Explore modules View classes

class documentation

class Container(IContainer): (source)

Known subclasses: orionis.Application

Constructor: Container(*args, **kwargs)

View In Hierarchy

Resolve services from contracts, aliases and constructor signatures.

Concurrency

__new__ is thread-safe: concurrent construction of the same subclass always yields a single instance, guarded by _lock.

Inside one event loop, the one-shot work behind Lifetime.SINGLETON, Lifetime.SCOPED and deferred providers is serialised per key, so concurrent tasks share a single construction instead of duplicating it.

No other guarantee is provided: registration methods mutate plain dictionaries without locks, and containers shared between distinct event loops fall back to per-loop serialisation only.

Static Method	`__isBeingResolved`	Determine whether a concrete type is already resolving in this task.
Method	`__aliasService`	Validate and normalize a service alias string.
Async Method	`__autoResolveCallable`	Resolve and invoke a callable, injecting dependencies.
Async Method	`__autoResolveClass`	Automatically instantiate a class with injected dependencies.
Async Method	`__awaitPendingProvider`	Wait for a provider that has registered services but is still booting.
Method	`__bind`	Bind a concrete implementation to an abstract contract with a given lifetime.
Async Method	`__bootDeferredProvider`	Register a deferred provider once and complete its boot hook.
Async Method	`__createScoped`	Build and store the instance backing a scoped binding.
Async Method	`__createSingleton`	Build and cache the single instance backing a singleton binding.
Method	`__creationLock`	Return the creation lock for a key, bound to the running loop.
Method	`__ensureCanOverrideGlobal`	Ensure that a service or alias can be overridden globally.
Method	`__ensureCanOverrideScope`	Ensure that a service can be overridden in the current scope.
Method	`__ensureConcreteImplements`	Ensure that a concrete class implements the specified abstract class.
Method	`__ensureInstanceImplements`	Ensure that an instance implements the specified abstract class.
Method	`__init__`	Initialize the internal state of the container.
Method	`__new__`	Create and return a singleton instance for each class in the hierarchy.
Async Method	`__resolve`	Resolve an instance from a binding according to its lifetime.
Async Method	`__resolveArgument`	Resolve a single argument for dependency injection.
Async Method	`__resolveDeferredProvider`	Resolve and register a deferred service provider for a given service.
Async Method	`__resolveKey`	Resolve a service key to its abstract type.
Async Method	`__resolveOrBuild`	Resolve a binding for an abstract service or build it when unbound.
Async Method	`__resolveSchemaArgument`	Resolve an argument that is a subclass of msgspec.Struct.
Async Method	`__resolveSignature`	Resolve arguments for a callable signature using dependency injection.
Method	`beginScope`	Begin a new scope context manager for scoped services.
Method	`bound`	Determine if a key is bound in the container or current scope.
Async Method	`build`	Build and return an instance of the specified type.
Async Method	`call`	Invoke a method on an object instance with automatic dependency injection.
Method	`getCurrentScope`	Get the current active scope context for scoped services.
Method	`instance`	Register an object instance as a singleton in the container.
Async Method	`invoke`	Invoke a callable with automatic dependency injection.
Async Method	`make`	Resolve and return a service instance by key.
Method	`scoped`	Register a scoped service binding.
Method	`singleton`	Register a singleton service binding.
Method	`transient`	Register a transient service binding.
Class Variable	`_instances`	Undocumented
Class Variable	`_lock`	Undocumented
Instance Variable	`__aliases`	Undocumented
Instance Variable	`__bindings`	Undocumented
Instance Variable	`__cache_resolve_deferred_providers`	Undocumented
Instance Variable	`__creation_locks`	Undocumented
Instance Variable	`__pending_deferred`	Undocumented
Instance Variable	`__registered_deferred`	Undocumented
Instance Variable	`__singleton_cache`	Undocumented
Instance Variable	`_Container__initialized`	Undocumented
Instance Variable	`_deferred_providers`	Undocumented

@staticmethod

def __isBeingResolved(concrete: type[Any]) -> bool: (source) ¶

Determine whether a concrete type is already resolving in this task.

Parameters
concrete:`type[Any]`	Concrete class about to be constructed.

Returns
`bool`	True when the type is already on the current resolution stack, which means the caller must skip the creation lock it already owns.

def __aliasService(self, alias: str | None) -> str | None: (source) ¶

Validate and normalize a service alias string.

Parameters
alias:`str \| None`	The alias string to validate and normalize.

Returns
`str \| None`	The validated and normalized alias string, or None if not provided.

Raises
`TypeError`	If the alias is not a string.
`ValueError`	If the alias is empty after stripping.

async def __autoResolveCallable(self, type_: Callable[..., Any], *args: object, **kwargs: object) -> type[Any]: (source) ¶

Resolve and invoke a callable, injecting dependencies.

Parameters
type_:`Callable[..., Any]`	The callable to invoke.
*args:`object`	Positional arguments for the callable.
**kwargs:`object`	Keyword arguments for the callable.

Returns
`Any`	The result of the callable invocation.

Raises
`OrionisContainerCircularDependencyException`	If a circular dependency is detected.
`Exception`	If the callable cannot be auto-resolved.

async def __autoResolveClass(self, type_: Callable[..., Any], *args: tuple[Any, ...], **kwargs: dict[str, Any]) -> Any: (source) ¶

Automatically instantiate a class with injected dependencies.

Parameters
type_:`Callable[..., Any]`	The class to instantiate.
*args:`tuple[Any, ...]`	Positional arguments for the constructor.
**kwargs:`dict[str, Any]`	Keyword arguments for the constructor.

Returns
`Any`	The instantiated object with dependencies resolved.

Raises
`CircularDependencyException`	If a circular dependency is detected.
`Exception`	If the type cannot be auto-resolved.

async def __awaitPendingProvider(self, key: type[Any] | str): (source) ¶

Wait for a provider that has registered services but is still booting.

Parameters
key:`type[Any] \| str`	Service type or alias requested by a resolver.

Returns
`None`	The provider is ready, or the caller is its own bootstrap task.

def __bind(self, lifetime: Lifetime, abstract: type[Any] | None, concrete: type[Any], *, alias: str | None, override: bool) -> bool: (source) ¶

Bind a concrete implementation to an abstract contract with a given lifetime.

Parameters
lifetime:`Lifetime`	The lifetime of the binding (singleton, scoped, or transient).
abstract:`type[Any] \| None`	The abstract contract class to associate with the concrete class, or None to use the concrete class as the contract.
concrete:`type[Any]`	The concrete class to register.
alias:`str \| None`	An optional alias for the registration.
override:`bool`	If True, override any existing registration.

Returns
`bool`	True if the binding was registered successfully.

Raises
`TypeError`	If the concrete is not a class type or type validation fails.
`ValueError`	If the alias is invalid or already registered, or if the contract is already registered and override is False.

async def __bootDeferredProvider(self, provider_key: tuple[str, str]): (source) ¶

Parameters
provider_key:`tuple[str, str]`	Module and class identifying a provider whose lock is held.

Returns
`None`	The provider is marked ready after its boot hook succeeds.

async def __createScoped(self, binding: Binding, scope: ScopeManager, *args: tuple[Any, ...], **kwargs: dict[str, Any]) -> Any: (source) ¶

Build and store the instance backing a scoped binding.

Parameters
binding:`Binding`	The scoped binding to materialize.
scope:`ScopeManager`	The active scope that owns the resulting instance.
*args:`tuple[Any, ...]`	Positional arguments for the constructor.
**kwargs:`dict[str, Any]`	Keyword arguments for the constructor.

Returns
`Any`	The scoped instance, built by this call or by the task that won the creation lock.

async def __createSingleton(self, binding: Binding, *args: tuple[Any, ...], **kwargs: dict[str, Any]) -> Any: (source) ¶

Build and cache the single instance backing a singleton binding.

Parameters
binding:`Binding`	The singleton binding to materialize.
*args:`tuple[Any, ...]`	Positional arguments for the constructor.
**kwargs:`dict[str, Any]`	Keyword arguments for the constructor.

Returns
`Any`	The cached instance, built by this call or by the task that won the creation lock.

def __creationLock(self, key: Any) -> asyncio.Lock: (source) ¶

Return the creation lock for a key, bound to the running loop.

Parameters
key:`Any`	Contract or deferred provider key whose construction is guarded.

Returns
`asyncio.Lock`	Lock owned by the running loop for this key. A new lock replaces any entry created on a different loop.

def __ensureCanOverrideGlobal(self, override: bool, abstract: type[Any], alias: str | None): (source) ¶

Ensure that a service or alias can be overridden globally.

Parameters
override:`bool`	Whether to allow overriding existing registrations.
abstract:`type[Any]`	The abstract contract type to check.
alias:`str \| None`	The alias to check for conflicts.

Returns
`None`	This method does not return a value.

Raises
`ValueError`	If the service or alias already exists globally and override is False.

def __ensureCanOverrideScope(self, override: bool, abstract: type[Any], scope: ScopeManager): (source) ¶

Ensure that a service can be overridden in the current scope.

Parameters
override:`bool`	Whether to allow overriding existing registrations.
abstract:`type[Any]`	The abstract contract type to check.
scope:`ScopeManager`	The current scoped service registry.

Returns
`None`	This method does not return a value.

Raises
`ValueError`	If the service already exists in the current scope and override is False.

def __ensureConcreteImplements(self, abstract: type[Any], concrete: type[Any]): (source) ¶

Ensure that a concrete class implements the specified abstract class.

Parameters
abstract:`type[Any]`	The abstract class type to check against.
concrete:`type[Any]`	The concrete class type to validate.

Returns
`None`	This method does not return a value.

Raises
`TypeError`	If `abstract` or `concrete` is not a class type, or if `concrete` does not implement `abstract`.

def __ensureInstanceImplements(self, abstract: type[Any], instance: object): (source) ¶

Ensure that an instance implements the specified abstract class.

Parameters
abstract:`type[Any]`	The abstract class type to check against.
instance:`object`	The object instance to validate.

Returns
`None`	This method does not return a value.

Raises
`TypeError`	If `abstract` is not a class type or `instance` does not implement it.

def __init__(self): (source) ¶

overridden in orionis.Application

Initialize the internal state of the container.

Sets up internal data structures for dependency injection and ensures single initialization per instance.

Returns
`None`	This method does not return a value.

def __new__(cls, *args: object, **kwargs: object) -> Self: (source) ¶

Create and return a singleton instance for each class in the hierarchy.

Ensures thread-safe singleton instantiation for each subclass of Container. Uses double-checked locking to avoid race conditions and optimize performance.

Parameters
cls	Undocumented
*args:`object`	Value supplied for `*args`.
**kwargs:`object`	Value supplied for `**kwargs`.

Returns
`Self`	The singleton instance of the calling class.

async def __resolve(self, binding: Binding, *args: tuple[Any, ...], **kwargs: dict[str, Any]) -> Any: (source) ¶

Resolve an instance from a binding according to its lifetime.

Parameters
binding:`Binding`	The binding to resolve.
*args:`tuple[Any, ...]`	Positional arguments for the constructor.
**kwargs:`dict[str, Any]`	Keyword arguments for the constructor.

Returns
`Any`	The resolved instance according to the binding's lifetime.

Raises
`RuntimeError`	If there is no active scope for scoped services.

async def __resolveArgument(self, argument: Argument) -> Any: (source) ¶

Resolve a single argument for dependency injection.

Parameters
argument:`Argument`	The argument metadata to resolve.

Returns
`Any`	The resolved value for the argument.

Raises
`TypeError`	If the argument cannot be resolved or is a built-in type.

async def __resolveDeferredProvider(self, key: type[Any] | str): (source) ¶

Resolve and register a deferred service provider for a given service.

Notes

Loads and registers a deferred provider for the specified service. Returns early if the provider is already resolved or is a built-in.

Parameters
key:`type[Any] \| str`	The service type or fully qualified class name for which to find the deferred provider.

Returns
`None`	This method does not return a value. Registers the deferred service provider in the application container if found.

async def __resolveKey(self, key: type[Any] | str) -> type[Any]: (source) ¶

Resolve a service key to its abstract type.

Parameters
key:`type[Any] \| str`	Service identifier as an abstract type or alias string.

Returns
`type[Any]`	The resolved abstract service type.

Raises
`ValueError`	If a string alias is not registered after deferred resolution.

async def __resolveOrBuild(self, abstract: type[Any], key: type[Any] | str, *args: tuple[Any, ...], **kwargs: dict[str, Any]) -> Any: (source) ¶

Resolve a binding for an abstract service or build it when unbound.

Parameters
abstract:`type[Any]`	Abstract service type to resolve.
key:`type[Any] \| str`	Original service key used for resolution and error messages.
*args:`tuple[Any, ...]`	Positional arguments passed to the resolver or builder.
**kwargs:`dict[str, Any]`	Keyword arguments passed to the resolver or builder.

Returns
`Any`	Resolved service instance.

Raises
`ValueError`	If no binding exists and the service cannot be resolved.

async def __resolveSchemaArgument(self, argument: Argument) -> msgspec.Struct: (source) ¶

Resolve an argument that is a subclass of msgspec.Struct.

Parameters
argument:`Argument`	The argument metadata to resolve.

Returns
`msgspec.Struct`	The resolved value for the msgspec.Struct argument.

Raises
`Exception`	If there is an error during resolution of the msgspec.Struct argument.

async def __resolveSignature(self, arguments: tuple[Argument, ...], *args: object, **kwargs: object) -> tuple[list[Any], dict[str, Any]]: (source) ¶

Resolve arguments for a callable signature using dependency injection.

Parameters
arguments:`tuple[Argument, ...]`	Parameter metadata in declaration order.
*args:`object`	Positional arguments to pass to the callable.
**kwargs:`object`	Keyword arguments to pass to the callable.

Returns
`tuple[list[Any], dict[str, Any]]`	A tuple containing the resolved positional and keyword arguments.

def beginScope(self) -> ScopeManager: (source) ¶

overrides orionis.container.contracts.container.IContainer.beginScope

Begin a new scope context manager for scoped services.

Parameters
self:`Container`	The container instance.

Returns
`ScopeManager`	Context manager for managing the lifecycle of scoped services.

def bound(self, key: type[Any] | str) -> bool: (source) ¶

overrides orionis.container.contracts.container.IContainer.bound

Determine if a key is bound in the container or current scope.

Parameters
key:`type[Any] \| str`	The abstract type or alias to check for binding.

Returns
`bool`	True if the key is bound in the current scope or container, otherwise False.

async def build(self, type_: Callable[..., Any], *args: tuple[Any, ...], **kwargs: dict[str, Any]) -> Any: (source) ¶

overrides orionis.container.contracts.container.IContainer.build

Build and return an instance of the specified type.

Notes

Resolves deferred providers before attempting instantiation.

Parameters
type_:`Callable[..., Any]`	The class to instantiate.
*args:`tuple[Any, ...]`	Positional arguments for the constructor.
**kwargs:`dict[str, Any]`	Keyword arguments for the constructor.

Returns
`Any`	Instantiated object of the specified type.

Raises
`TypeError`	If the type cannot be auto-resolved by the container.

async def call(self, instance: object, method_name: str, *args: object, **kwargs: object) -> Any: (source) ¶

overrides orionis.container.contracts.container.IContainer.call

Invoke a method on an object instance with automatic dependency injection.

Parameters
instance:`object`	The object instance containing the method.
method_name:`str`	The name of the method to invoke.
*args:`object`	Positional arguments for the method.
**kwargs:`object`	Keyword arguments for the method.

Returns
`Any`	The result of the method invocation with dependencies resolved.

Raises
`AttributeError`	If the method is not found on the instance.
`TypeError`	If the attribute is not callable.

def getCurrentScope(self) -> ScopeManager | None: (source) ¶

overrides orionis.container.contracts.container.IContainer.getCurrentScope

Get the current active scope context for scoped services.

Notes

Returns None if there is no active scope. Use beginScope() to create a new scope context before accessing scoped services.

Parameters
self:`Container`	The container instance.

Returns
`ScopeManager \| None`	The current active scope context if available, otherwise None. The scope context is a dictionary-like object that contains instances of scoped services registered in the current scope.

def instance(self, abstract: type[Any] | None, instance: object, *, alias: str | None = None, override: bool = False) -> bool: (source) ¶

overrides orionis.container.contracts.container.IContainer.instance

Parameters
abstract:`type[Any] \| None`	The abstract contract class to associate with the instance, or None.
instance:`object`	The initialized object to register.
alias:`str \| None, optional`	An optional alias for the registration.
override:`bool, optional`	If True, override any existing registration.

Returns
`bool`	True if the instance was registered successfully.

Raises
`TypeError`	If the instance is a class, or if type validation fails.
`ValueError`	If the alias is invalid or already registered, or if the contract is already registered and override is False.

async def invoke(self, fn: Callable[..., Any], *args: tuple[Any, ...], **kwargs: dict[str, Any]) -> Any: (source) ¶

overrides orionis.container.contracts.container.IContainer.invoke

Invoke a callable with automatic dependency injection.

Parameters
fn:`Callable[..., Any]`	The callable to invoke. Must not be a class or type.
*args:`tuple[Any, ...]`	Positional arguments for the callable.
**kwargs:`dict[str, Any]`	Keyword arguments for the callable.

Returns
`Any`	The result of the callable execution with dependencies injected.

Raises
`TypeError`	If `fn` is not a callable or is a class/type.

async def make(self, key: type[Any] | str, *args: tuple[Any, ...], **kwargs: dict[str, Any]) -> Any: (source) ¶

overrides orionis.container.contracts.container.IContainer.make

Resolve and return a service instance by key.

Parameters
key:`type[Any] \| str`	The abstract type or alias to resolve.
*args:`tuple[Any, ...]`	Positional arguments for instantiation.
**kwargs:`dict[str, Any]`	Keyword arguments for instantiation.

Returns
`Any`	The resolved service instance.

Raises
`ValueError`	If the service is not registered and cannot be auto-resolved.

def scoped(self, abstract: type[Any] | None, concrete: type[Any], *, alias: str | None = None, override: bool = False) -> bool: (source) ¶

overrides orionis.container.contracts.container.IContainer.scoped

Parameters
abstract:`type[Any] \| None`	The abstract contract type to bind, or None to use the concrete type.
concrete:`type[Any]`	The concrete implementation type to register.
alias:`str \| None, optional`	An optional alias for the service.
override:`bool, optional`	Whether to override an existing registration.

Returns
`bool`	True if the binding was registered successfully.

def singleton(self, abstract: type[Any] | None, concrete: type[Any], *, alias: str | None = None, override: bool = False) -> bool: (source) ¶

overrides orionis.container.contracts.container.IContainer.singleton

Parameters
abstract:`type[Any] \| None`	The abstract contract type to bind, or None to use the concrete type.
concrete:`type[Any]`	The concrete implementation type to register.
alias:`str \| None, optional`	An optional alias for the service.
override:`bool, optional`	Whether to override an existing registration.

Returns
`bool`	True if the binding was registered successfully.

def transient(self, abstract: type[Any] | None, concrete: type[Any], *, alias: str | None = None, override: bool = False) -> bool: (source) ¶

overrides orionis.container.contracts.container.IContainer.transient

Parameters
abstract:`type[Any] \| None`	The abstract contract type to bind, or None to use the concrete type.
concrete:`type[Any]`	The concrete implementation type to register.
alias:`str \| None, optional`	An optional alias for the service.
override:`bool, optional`	Whether to override an existing registration.

Returns
`bool`	True if the binding was registered successfully.

_instances: ClassVar[dict] = (source) ¶

Undocumented

_lock: ClassVar[threading.RLock] = (source) ¶

Undocumented

__aliases: dict[str, type] = (source) ¶

Undocumented

__bindings: dict[Any, Binding] = (source) ¶

Undocumented

__cache_resolve_deferred_providers: set[Any] = (source) ¶

Undocumented

__creation_locks: dict[Any, tuple[asyncio.AbstractEventLoop, asyncio.Lock]] = (source) ¶

Undocumented

__pending_deferred: dict[str, tuple[str, str]] = (source) ¶

Undocumented

__registered_deferred: dict[tuple[str, str], IServiceProvider] = (source) ¶

Undocumented

__singleton_cache: dict[str, Any] = (source) ¶

Undocumented

_Container__initialized: bool = (source) ¶

Undocumented

_deferred_providers: dict[str, dict[str, str]] = (source) ¶

overridden in orionis.Application

Undocumented