dioxide¶
dioxide: Fast, Rust-backed declarative dependency injection for Python.
dioxide is a modern dependency injection framework that combines: - Declarative Python API with hexagonal architecture support - High-performance Rust-backed container implementation - Type-safe dependency resolution with IDE autocomplete support - Profile-based configuration for different environments
- Quick Start (using instance container - recommended):
>>> from dioxide import Container, service, adapter, Profile >>> from typing import Protocol >>> >>> class EmailPort(Protocol): ... async def send(self, to: str, subject: str, body: str) -> None: ... >>> >>> @adapter.for_(EmailPort, profile=Profile.PRODUCTION) ... class SendGridAdapter: ... async def send(self, to: str, subject: str, body: str) -> None: ... pass # Real implementation >>> >>> @service ... class UserService: ... def __init__(self, email: EmailPort): ... self.email = email >>> >>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> user_service = container.resolve(UserService) >>> # Or use bracket syntax: >>> user_service = container[UserService]
- Global container (convenient for simple scripts):
>>> from dioxide import container, Profile >>> >>> container.scan(profile=Profile.PRODUCTION) >>> user_service = container.resolve(UserService) >>> >>> # For testing with global container, use reset_global_container() >>> from dioxide import reset_global_container >>> reset_global_container()
- Testing (fresh container per test - recommended):
>>> from dioxide.testing import fresh_container >>> from dioxide import Profile >>> >>> async with fresh_container(profile=Profile.TEST) as test_container: ... service = test_container.resolve(UserService) ... # Test with isolated container
For more information, see the README and documentation.
Submodules¶
Attributes¶
Exceptions¶
Raised when no adapter is registered for a port in the active profile. |
|
Raised when a longer-lived scope depends on a shorter-lived scope. |
|
Raised when circular dependencies are detected among @lifecycle components. |
|
Base class for all dioxide errors with rich formatting. |
|
Base class for dependency resolution failures. |
|
Raised when scope-related operations fail. |
|
Raised when a service or component cannot be resolved. |
Classes¶
Dependency injection container. |
|
A scoped container for REQUEST-scoped dependency resolution. |
|
Extensible, type-safe profile identifier for adapter selection. |
|
Component lifecycle scope. |
Functions¶
Reset the global container to an empty state. |
|
|
Mark a class for lifecycle management with initialization and cleanup. |
|
Mark a class as a core domain service. |
|
Create a fresh, isolated container for testing. |
Package Contents¶
- class dioxide.Container(allowed_packages=None, profile=None)[source]¶
Dependency injection container.
The Container manages component registration and dependency resolution for your application. It supports both automatic discovery via the @component decorator and manual registration for fine-grained control.
The container is backed by a high-performance Rust implementation that handles provider caching, singleton management, and type resolution.
- Features:
Type-safe dependency resolution with full IDE support
Automatic dependency injection based on type hints
SINGLETON and FACTORY lifecycle scopes
Thread-safe singleton caching (Rust-backed)
Automatic discovery via @component decorator
Manual registration for non-decorated classes
Examples
- Automatic discovery with @component:
>>> from dioxide import Container, component >>> >>> @component ... class Database: ... def query(self, sql): ... return f'Executing: {sql}' >>> >>> @component ... class UserService: ... def __init__(self, db: Database): ... self.db = db >>> >>> container = Container() >>> container.scan() # Auto-discover @component classes >>> service = container.resolve(UserService) >>> result = service.db.query('SELECT * FROM users')
- Manual registration:
>>> from dioxide import Container >>> >>> class Config: ... def __init__(self, env: str): ... self.env = env >>> >>> container = Container() >>> container.register_singleton(Config, lambda: Config('production')) >>> config = container.resolve(Config) >>> assert config.env == 'production'
- Factory scope for per-request objects:
>>> from dioxide import Container, component, Scope >>> >>> @component(scope=Scope.FACTORY) ... class RequestContext: ... def __init__(self): ... self.id = id(self) >>> >>> container = Container() >>> container.scan() >>> ctx1 = container.resolve(RequestContext) >>> ctx2 = container.resolve(RequestContext) >>> assert ctx1 is not ctx2 # Different instances
Note
The container should be created once at application startup and reused throughout the application lifecycle. Each container maintains its own singleton cache and registration state.
- Parameters:
profile (dioxide.profile_enum.Profile | str | None)
- register_instance(component_type, instance)[source]¶
Register a pre-created instance for a given type.
This method registers an already-instantiated object that will be returned whenever the type is resolved. Useful for registering configuration objects or external dependencies.
Type safety is enforced at runtime: the instance must be an instance of component_type (or a subclass). For Protocol types, structural compatibility is checked.
- Parameters:
component_type (type[T]) – The type to register. This is used as the lookup key when resolving dependencies.
instance (T) – The pre-created instance to return for this type. Must be an instance of component_type or a compatible type.
- Raises:
- Return type:
None
Example
>>> from dioxide import Container >>> >>> class Config: ... def __init__(self, debug: bool): ... self.debug = debug >>> >>> container = Container() >>> config_instance = Config(debug=True) >>> container.register_instance(Config, config_instance) >>> resolved = container.resolve(Config) >>> assert resolved is config_instance >>> assert resolved.debug is True
- Type safety example:
>>> container = Container() >>> container.register_instance(str, 42) # Raises TypeError Traceback (most recent call last): ... TypeError: instance must be of type 'str', got 'int'
- register_class(component_type, implementation)[source]¶
Register a class to instantiate for a given type.
Registers a class that will be instantiated with no arguments when the type is resolved. The class’s __init__ method will be called without parameters.
- Parameters:
- Raises:
KeyError – If the type is already registered in this container.
- Return type:
None
Example
>>> from dioxide import Container >>> >>> class DatabaseConnection: ... def __init__(self): ... self.connected = True >>> >>> container = Container() >>> container.register_class(DatabaseConnection, DatabaseConnection) >>> db = container.resolve(DatabaseConnection) >>> assert db.connected is True
Note
For classes requiring constructor arguments, use register_singleton_factory() or register_transient_factory() with a lambda that provides the arguments.
- register_singleton_factory(component_type, factory)[source]¶
Register a singleton factory function for a given type.
The factory will be called once when the type is first resolved, and the result will be cached. All subsequent resolve() calls for this type will return the same cached instance.
- Parameters:
component_type (type[T]) – The type to register. This is used as the lookup key when resolving dependencies.
factory (collections.abc.Callable[[], T]) – A callable that takes no arguments and returns an instance of component_type. Called exactly once, on first resolve().
- Raises:
KeyError – If the type is already registered in this container.
- Return type:
None
Example
>>> from dioxide import Container >>> >>> class ExpensiveService: ... def __init__(self, config_path: str): ... self.config_path = config_path ... self.initialized = True >>> >>> container = Container() >>> container.register_singleton_factory(ExpensiveService, lambda: ExpensiveService('/etc/config.yaml')) >>> service1 = container.resolve(ExpensiveService) >>> service2 = container.resolve(ExpensiveService) >>> assert service1 is service2 # Same instance
Note
This is the recommended registration method for most services, as it provides lazy initialization and instance sharing.
- register_transient_factory(component_type, factory)[source]¶
Register a transient factory function for a given type.
The factory will be called every time the type is resolved, creating a new instance for each resolve() call. Use this for stateful objects that should not be shared.
- Parameters:
component_type (type[T]) – The type to register. This is used as the lookup key when resolving dependencies.
factory (collections.abc.Callable[[], T]) – A callable that takes no arguments and returns an instance of component_type. Called on every resolve() to create a fresh instance.
- Raises:
KeyError – If the type is already registered in this container.
- Return type:
None
Example
>>> from dioxide import Container >>> >>> class RequestHandler: ... _counter = 0 ... ... def __init__(self): ... RequestHandler._counter += 1 ... self.request_id = RequestHandler._counter >>> >>> container = Container() >>> container.register_transient_factory(RequestHandler, lambda: RequestHandler()) >>> handler1 = container.resolve(RequestHandler) >>> handler2 = container.resolve(RequestHandler) >>> assert handler1 is not handler2 # Different instances >>> assert handler1.request_id != handler2.request_id
Note
Use this for objects with per-request or per-operation lifecycle. For shared services, use register_singleton_factory() instead.
- register_singleton(component_type, factory)[source]¶
Register a singleton provider manually.
Convenience method that calls register_singleton_factory(). The factory will be called once when the type is first resolved, and the result will be cached for the lifetime of the container.
- Parameters:
component_type (type[T]) – The type to register. This is used as the lookup key when resolving dependencies.
factory (collections.abc.Callable[[], T]) – A callable that takes no arguments and returns an instance of component_type. Called exactly once, on first resolve().
- Raises:
KeyError – If the type is already registered in this container.
- Return type:
None
Example
>>> from dioxide import Container >>> >>> class Config: ... def __init__(self, db_url: str): ... self.db_url = db_url >>> >>> container = Container() >>> container.register_singleton(Config, lambda: Config('postgresql://localhost')) >>> config = container.resolve(Config) >>> assert config.db_url == 'postgresql://localhost'
Note
This is an alias for register_singleton_factory() provided for convenience and clarity.
- register_factory(component_type, factory)[source]¶
Register a transient (factory) provider manually.
Convenience method that calls register_transient_factory(). The factory will be called every time the type is resolved, creating a new instance for each resolve() call.
- Parameters:
component_type (type[T]) – The type to register. This is used as the lookup key when resolving dependencies.
factory (collections.abc.Callable[[], T]) – A callable that takes no arguments and returns an instance of component_type. Called on every resolve() to create a fresh instance.
- Raises:
KeyError – If the type is already registered in this container.
- Return type:
None
Example
>>> from dioxide import Container >>> >>> class Transaction: ... _id_counter = 0 ... ... def __init__(self): ... Transaction._id_counter += 1 ... self.tx_id = Transaction._id_counter >>> >>> container = Container() >>> container.register_factory(Transaction, lambda: Transaction()) >>> tx1 = container.resolve(Transaction) >>> tx2 = container.resolve(Transaction) >>> assert tx1.tx_id != tx2.tx_id # Different instances
Note
This is an alias for register_transient_factory() provided for convenience and clarity.
- resolve(component_type)[source]¶
Resolve a component instance.
Retrieves or creates an instance of the requested type based on its registration. For singletons, returns the cached instance (creating it on first call). For factories, creates a new instance every time.
For multi-bindings, use
list[Port]type hint to resolve all adapters registered withmulti=Truefor that port.- Parameters:
component_type (type[T]) – The type to resolve. Must have been previously registered via scan() or manual registration methods. Can be a
list[Port]type to resolve multi-bindings.- Returns:
An instance of the requested type. For SINGLETON scope, the same instance is returned on every call. For FACTORY scope, a new instance is created on each call. For
list[Port], returns a list of all multi-binding adapters for that port.- Raises:
AdapterNotFoundError – If the type is a port (Protocol/ABC) and no adapter is registered for the current profile.
ServiceNotFoundError – If the type is a service/component that cannot be resolved (not registered or has unresolvable dependencies).
ScopeError – If trying to resolve a REQUEST-scoped component outside of a scope context. Use
container.create_scope()to create a scope.
- Return type:
T
Example
>>> from dioxide import Container, component >>> >>> @component ... class Logger: ... def log(self, msg: str): ... print(f'LOG: {msg}') >>> >>> @component ... class Application: ... def __init__(self, logger: Logger): ... self.logger = logger >>> >>> container = Container() >>> container.scan() >>> app = container.resolve(Application) >>> app.logger.log('Application started')
Note
Type annotations in constructors enable automatic dependency injection. The container recursively resolves all dependencies.
- __getitem__(component_type)[source]¶
Resolve a component using bracket syntax.
Provides an alternative, more Pythonic syntax for resolving components. This method is equivalent to calling resolve() and simply delegates to it.
- Parameters:
component_type (type[T]) – The type to resolve. Must have been previously registered via scan() or manual registration methods.
- Returns:
An instance of the requested type. For SINGLETON scope, the same instance is returned on every call. For FACTORY scope, a new instance is created on each call.
- Raises:
KeyError – If the type is not registered in this container.
- Return type:
T
Example
>>> from dioxide import container, component >>> >>> @component ... class Logger: ... def log(self, msg: str): ... print(f'LOG: {msg}') >>> >>> container.scan() >>> logger = container[Logger] # Bracket syntax >>> logger.log('Using bracket notation')
Note
This is purely a convenience method. Both container[Type] and container.resolve(Type) work identically and return the same instance for singleton-scoped components.
- is_empty()[source]¶
Check if container has no registered providers.
- Returns:
True if no types have been registered, False if at least one type has been registered.
- Return type:
Example
>>> from dioxide import Container >>> >>> container = Container() >>> assert container.is_empty() >>> >>> container.scan() # Register @component classes >>> # If any @component classes exist, container is no longer empty
- __len__()[source]¶
Get count of registered providers.
- Returns:
The number of types that have been registered in this container.
- Return type:
Example
>>> from dioxide import Container, component >>> >>> @component ... class ServiceA: ... pass >>> >>> @component ... class ServiceB: ... pass >>> >>> container = Container() >>> assert len(container) == 0 >>> container.scan() >>> assert len(container) == 2
- __repr__()[source]¶
Return an informative string representation for debugging.
Shows the active profile, port count, and service count so agents and developers can inspect container state in a REPL or debugger.
- Returns:
A string like
Container(profile=Profile('production'), ports=5, services=3)orContainer(profile=None, ports=0, services=0)when no profile is set.- Return type:
Example
>>> from dioxide import Container, Profile >>> container = Container(profile=Profile.PRODUCTION) >>> repr(container) "Container(profile=Profile('production'), ports=..., services=...)"
- list_registered()[source]¶
List all types registered in this container.
Returns a list of all type objects (classes, protocols, ABCs) that have been registered with this container, either through scan() or manual registration methods.
This is useful for debugging registration issues - when you get a “not registered” error, call this method to see what IS registered.
- Returns:
List of type objects registered in this container. The list order is not guaranteed to be consistent between calls.
- Return type:
Example
>>> from dioxide import Container, Profile, adapter, service >>> >>> class EmailPort(Protocol): ... async def send(self, to: str) -> None: ... >>> >>> @adapter.for_(EmailPort, profile=Profile.PRODUCTION) ... class SendGridAdapter: ... async def send(self, to: str) -> None: ... pass >>> >>> @service ... class UserService: ... pass >>> >>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> registered = container.list_registered() >>> # registered contains [EmailPort, UserService]
See also
is_registered()- Check if a specific type is registeredget_adapters_for()- Get adapter details for a port
- is_registered(port_or_service)[source]¶
Check if a type is registered in this container.
Useful for verifying that a type has been registered before attempting to resolve it, or for test assertions about container configuration.
- Parameters:
port_or_service (type[Any]) – The type to check. Can be a port (Protocol/ABC) or a service class.
- Returns:
True if the type is registered, False otherwise.
- Return type:
Example
>>> from dioxide import Container, Profile, adapter >>> >>> class EmailPort(Protocol): ... async def send(self, to: str) -> None: ... >>> >>> @adapter.for_(EmailPort, profile=Profile.PRODUCTION) ... class SendGridAdapter: ... async def send(self, to: str) -> None: ... pass >>> >>> container = Container() >>> assert container.is_registered(EmailPort) is False >>> container.scan(profile=Profile.PRODUCTION) >>> assert container.is_registered(EmailPort) is True
See also
list_registered()- Get all registered typesresolve()- Actually resolve a registered type
- property active_profile: dioxide.profile_enum.Profile | None¶
Get the profile this container was scanned with.
Returns the Profile value used when scan() was called, or None if scan() hasn’t been called yet. This is useful for debugging to verify which profile is active.
- Returns:
The Profile value if scan() was called with a profile, None if scan() hasn’t been called or was called without a profile.
- Return type:
dioxide.profile_enum.Profile | None
Example
>>> from dioxide import Container, Profile >>> >>> container = Container() >>> assert container.active_profile is None >>> >>> container.scan(profile=Profile.PRODUCTION) >>> assert container.active_profile == Profile.PRODUCTION >>> >>> # Or with constructor profile: >>> container2 = Container(profile=Profile.TEST) >>> assert container2.active_profile == Profile.TEST
See also
scan()- Set the active profile during scanningdioxide.Profile- Extensible profile identifiers
- get_adapters_for(port)[source]¶
Get all adapters registered for a port across all profiles.
Inspects the global adapter registry to find all adapters that implement the specified port, organized by profile. This is useful for debugging to see which adapters are available for a port in different profiles.
Note: This method looks at the global adapter registry, not just what’s registered in this container instance. This allows you to see all available adapters even if the container was scanned with a different profile.
- Parameters:
port (type[Any]) – The port type (Protocol/ABC) to find adapters for.
- Returns:
Dictionary mapping Profile enum values to adapter classes. Returns an empty dict if no adapters are registered for the port.
- Return type:
dict[dioxide.profile_enum.Profile, type[Any]]
Example
>>> from dioxide import Container, Profile, adapter >>> >>> class EmailPort(Protocol): ... async def send(self, to: str) -> None: ... >>> >>> @adapter.for_(EmailPort, profile=Profile.PRODUCTION) ... class SendGridAdapter: ... async def send(self, to: str) -> None: ... pass >>> >>> @adapter.for_(EmailPort, profile=Profile.TEST) ... class FakeEmailAdapter: ... async def send(self, to: str) -> None: ... pass >>> >>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> adapters = container.get_adapters_for(EmailPort) >>> # adapters = { >>> # Profile.PRODUCTION: SendGridAdapter, >>> # Profile.TEST: FakeEmailAdapter, >>> # }
See also
is_registered()- Check if a port has an adapteradapter.for_()- Register adapters for ports
- debug(file=None)[source]¶
Print a summary of all registered components.
Shows services, adapters (grouped by port), and active profile. Useful for verifying what’s actually registered in the container.
- Parameters:
file (Any) – Optional file-like object to write to (default: returns string). If provided, also writes the output to the file.
- Returns:
Formatted debug string with container summary.
- Return type:
Example
>>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> print(container.debug()) === dioxide Container Debug === Active Profile: production
- Services (2):
UserService (SINGLETON)
NotificationService (SINGLETON)
- Adapters by Port:
- EmailPort:
SendGridAdapter (profiles: production)
- DatabasePort:
PostgresAdapter (profiles: production, lifecycle)
- explain(cls)[source]¶
Explain how a type would be resolved.
Shows the resolution path, which adapter/service is selected, and all transitive dependencies in a tree format.
- Parameters:
cls (type[Any]) – The type to explain resolution for. Can be a service, port (Protocol/ABC), or any registered type.
- Returns:
Formatted string showing the resolution tree.
- Return type:
Example
>>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> print(container.explain(UserService)) === Resolution: UserService ===
UserService (SINGLETON) +– db: DatabasePort | +– PostgresAdapter (profile: production) | +– config: AppConfig +– email: EmailPort
- +– SendGridAdapter (profile: production)
+– config: AppConfig
- graph(format='mermaid')[source]¶
Generate a dependency graph visualization.
Creates a visual representation of the dependency graph that can be rendered with Mermaid (default) or Graphviz DOT format.
- Parameters:
format (str) – Output format, either ‘mermaid’ (default) or ‘dot’.
- Returns:
String containing the graph in the requested format.
- Return type:
Example
>>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> print(container.graph()) graph TD subgraph Services UserService[UserService<br/>SINGLETON] end
- subgraph Ports
EmailPort{{EmailPort}}
end
- subgraph Adapters
SendGridAdapter[SendGridAdapter<br/>production]
end
UserService –> EmailPort EmailPort -.-> SendGridAdapter
- scan(package=None, profile=None)[source]¶
Discover and register all @component and @adapter decorated classes.
Scans the global registries for all classes decorated with @component or @adapter and registers them with the container. Dependencies are automatically resolved based on constructor type hints.
This is the primary method for setting up the container in a declarative style. Call it once after all components are imported.
- Parameters:
package (str | None) – Optional package name to scan. If None, scans all registered components. If provided, imports all modules in the specified package (including sub-packages) to trigger decorator execution, then scans only components from that package.
profile (str | dioxide.profile_enum.Profile | None) – Optional profile to filter components/adapters. Accepts either a Profile enum value (Profile.PRODUCTION, Profile.TEST, etc.) or a string profile name. If None, registers all components/adapters regardless of profile. If provided, only registers components/adapters that have the matching profile in their __dioxide_profiles__ attribute. Components/ adapters decorated with Profile.ALL (“*”) are registered in all profiles. Profile names are normalized to lowercase for matching.
- Return type:
None
- Registration behavior:
SINGLETON scope (default): Creates singleton factory with caching
FACTORY scope: Creates transient factory for new instances
Manual registrations take precedence over @component/@adapter decorators
Already-registered types are silently skipped
Profile filtering applies to components/adapters with @profile decorator
Adapters are registered under their port type (Protocol/ABC)
Multiple adapters for same port+profile raises ValueError
Example
>>> from dioxide import Container, Profile, component, adapter, Scope, profile >>> >>> # Define a port (Protocol) >>> class EmailPort(Protocol): ... async def send(self, to: str, subject: str, body: str) -> None: ... >>> >>> # Create adapter for production >>> @adapter.for_(EmailPort, profile='production') ... class SendGridAdapter: ... async def send(self, to: str, subject: str, body: str) -> None: ... pass >>> >>> # Create service that depends on port >>> @component ... class UserService: ... def __init__(self, email: EmailPort): ... self.email = email >>> >>> # Scan with Profile enum (recommended) >>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> service = container.resolve(UserService) >>> # service.email is a SendGridAdapter instance >>> >>> # Or with string profile (also supported) >>> container2 = Container() >>> container2.scan(profile='production') # Same as above
- Raises:
ValueError – If multiple adapters are registered for the same port and profile combination (ambiguous registration)
- Parameters:
package (str | None)
profile (str | dioxide.profile_enum.Profile | None)
- Return type:
None
Note
Ensure all component/adapter classes are imported before calling scan()
Constructor dependencies must have type hints
Circular dependencies will cause infinite recursion
Manual registrations (register_*) take precedence over scan()
Profile names are case-insensitive (normalized to lowercase)
- async start()[source]¶
Initialize all @lifecycle components in dependency order.
Resolves all registered components and calls initialize() on those decorated with @lifecycle. Components are initialized in dependency order (dependencies before their dependents).
The list of lifecycle instances is cached during start() and reused during stop() to ensure all initialized components are disposed.
If initialization fails for any component, all previously initialized components are disposed in reverse order (rollback).
- Raises:
Exception – If any component’s initialize() method raises an exception. Already-initialized components are disposed before re-raising.
- Return type:
None
Example
>>> from dioxide import Container, service, lifecycle, Profile >>> >>> @service ... @lifecycle ... class Database: ... async def initialize(self) -> None: ... print('Database connected') ... ... async def dispose(self) -> None: ... print('Database disconnected') >>> >>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> await container.start() Database connected
- async stop()[source]¶
Dispose all @lifecycle components in reverse dependency order.
Calls dispose() on all components decorated with @lifecycle. Components are disposed in reverse dependency order (dependents before their dependencies).
Uses the cached list of lifecycle instances from start() to ensure exactly the components that were initialized are disposed.
If disposal fails for any component, continues disposing remaining components (does not raise until all disposals are attempted).
Example
>>> from dioxide import Container, service, lifecycle, Profile >>> >>> @service ... @lifecycle ... class Database: ... async def initialize(self) -> None: ... pass ... ... async def dispose(self) -> None: ... print('Database disconnected') >>> >>> container = Container() >>> container.scan(profile=Profile.PRODUCTION) >>> await container.start() >>> await container.stop() Database disconnected
- Return type:
None
- async __aenter__()[source]¶
Enter async context manager - calls start().
Example
>>> from dioxide import Container, service, lifecycle >>> >>> @service ... @lifecycle ... class Database: ... async def initialize(self) -> None: ... print('Connected') ... ... async def dispose(self) -> None: ... print('Disconnected') >>> >>> async with Container() as container: ... container.scan() ... # Use container Connected Disconnected
- Return type:
- async __aexit__(exc_type, exc_val, exc_tb)[source]¶
Exit async context manager - calls stop().
- Parameters:
exc_type (type[BaseException] | None) – Exception type if an exception was raised
exc_val (BaseException | None) – Exception value if an exception was raised
exc_tb (Any) – Exception traceback if an exception was raised
- Return type:
None
- reset()[source]¶
Clear cached instances for test isolation.
Clears the singleton cache but preserves provider registrations. Use this between tests to ensure fresh instances without re-scanning.
This method is particularly useful in pytest fixtures to ensure test isolation while avoiding the overhead of re-scanning:
Example:
@pytest.fixture(autouse=True) def setup_container(): container.scan(profile=Profile.TEST) yield container.reset() # Fresh instances for next test
For complete isolation (including new provider registrations), consider using fresh Container instances instead.
Note
Instance registrations (via register_instance) are NOT cleared because they reference external objects
Provider registrations are preserved (no need to re-scan)
Lifecycle instance cache is cleared
See also
Container: Create fresh instances for complete isolation
- Return type:
None
- create_scope()[source]¶
Create a new scope for REQUEST-scoped dependency resolution.
Returns an async context manager that provides a ScopedContainer for resolving REQUEST-scoped dependencies. Each scope maintains its own cache of REQUEST-scoped instances.
Usage:
async with container.create_scope() as scope: # REQUEST-scoped components are cached within this scope handler = scope.resolve(RequestHandler) # Same scope = same instance handler2 = scope.resolve(RequestHandler) assert handler is handler2 # Scope exits - REQUEST components are disposed
- Scope behavior:
SINGLETON: Resolved from parent container (shared)
REQUEST: Cached within scope (fresh per scope)
FACTORY: New instance each resolution
- Lifecycle management:
REQUEST-scoped components decorated with @lifecycle have their dispose() method called when the scope exits.
- Returns:
An async context manager that yields a ScopedContainer.
- Return type:
Example
>>> from dioxide import Container, service, Scope >>> >>> @service(scope=Scope.REQUEST) ... class RequestContext: ... def __init__(self): ... self.request_id = str(uuid.uuid4()) >>> >>> container = Container() >>> container.scan() >>> >>> async with container.create_scope() as scope: ... ctx1 = scope.resolve(RequestContext) ... ctx2 = scope.resolve(RequestContext) ... assert ctx1 is ctx2 # Same within scope >>> >>> async with container.create_scope() as scope2: ... ctx3 = scope2.resolve(RequestContext) ... assert ctx3 is not ctx1 # Different scope = different instance
See also
ScopedContainer- The scoped container typedioxide.scope.Scope- Scope enumdioxide.exceptions.ScopeError- Scope errors
- class dioxide.ScopedContainer(parent, scope_id)[source]¶
A scoped container for REQUEST-scoped dependency resolution.
ScopedContainer provides a context for resolving REQUEST-scoped dependencies. It wraps a parent Container and maintains its own cache of REQUEST-scoped instances that are unique to this scope.
- Key behaviors:
SINGLETON: Resolved from parent container (shared across all scopes)
REQUEST: Cached within this scope (fresh per scope, shared within scope)
FACTORY: New instance each time (same as parent container)
- Creating a ScopedContainer:
Use the async context manager pattern via
container.create_scope():async with container.create_scope() as scope: # REQUEST-scoped components are cached within this scope handler = scope.resolve(RequestHandler) # Same scope = same instance handler2 = scope.resolve(RequestHandler) assert handler is handler2 # Scope exits - REQUEST components are disposed
Each scope has a unique ID for tracking and debugging:
async with container.create_scope() as scope: print(f'Scope ID: {scope.scope_id}') # e.g., "abc123..."
- REQUEST-scoped dependencies:
Components decorated with
@service(scope=Scope.REQUEST)require a scope context for resolution:@service(scope=Scope.REQUEST) class RequestContext: def __init__(self): self.request_id = str(uuid.uuid4()) # Outside scope - raises ScopeError container.resolve(RequestContext) # Error! # Inside scope - works async with container.create_scope() as scope: ctx = scope.resolve(RequestContext) # OK
- Lifecycle management:
REQUEST-scoped components with
@lifecycleare disposed when the scope exits:@service(scope=Scope.REQUEST) @lifecycle class DbConnection: async def initialize(self) -> None: self.conn = await create_connection() async def dispose(self) -> None: await self.conn.close() async with container.create_scope() as scope: db = scope.resolve(DbConnection) # db.initialize() called automatically # db.dispose() called automatically on scope exit
See also
Container.create_scope()- How to create scopesdioxide.scope.Scope- Scope enum (SINGLETON, REQUEST, FACTORY)dioxide.exceptions.ScopeError- Raised for scope violations
- __repr__()[source]¶
Return an informative string representation for debugging.
Shows the active profile from the parent container and the parent type so agents and developers can inspect scoped container state.
- Returns:
A string like
ScopedContainer(profile=Profile('test'), parent=Container).- Return type:
Example
>>> async with container.create_scope() as scope: ... repr(scope) "ScopedContainer(profile=Profile('test'), parent=Container)"
- resolve(component_type)[source]¶
Resolve a component instance within this scope.
- Resolution behavior depends on the component’s scope:
SINGLETON: Delegates to parent container (shared instance)
REQUEST: Caches in this scope (fresh per scope)
FACTORY: New instance each resolution (no caching)
- Parameters:
component_type (type[T]) – The type to resolve.
- Returns:
An instance of the requested type.
- Raises:
AdapterNotFoundError – If the type is a port with no adapter.
ServiceNotFoundError – If the type is an unregistered service.
- Return type:
T
Example
>>> async with container.create_scope() as scope: ... # REQUEST-scoped: cached within scope ... ctx1 = scope.resolve(RequestContext) ... ctx2 = scope.resolve(RequestContext) ... assert ctx1 is ctx2 # Same instance ... ... # SINGLETON: shared with parent ... config = scope.resolve(AppConfig)
- __getitem__(component_type)[source]¶
Resolve a component using bracket syntax.
Equivalent to calling
scope.resolve(component_type).- Parameters:
component_type (type[T]) – The type to resolve.
- Returns:
An instance of the requested type.
- Return type:
T
Example
>>> async with container.create_scope() as scope: ... ctx = scope[RequestContext] # Same as scope.resolve(RequestContext)
- create_scope()[source]¶
Nested scopes are not supported in v0.3.0.
- Raises:
ScopeError – Always raises, as nested scopes are not supported.
- Return type:
- dioxide.reset_global_container()[source]¶
Reset the global container to an empty state.
This function replaces the global container’s internal state with a fresh Rust container instance, clearing all registrations and cached singletons. The global container object reference remains the same, so any code holding a reference to
containerwill see the reset state.Warning
This function is intended for testing only.
Calling this in production code will cause unpredictable behavior as all registered services and adapters will be lost. Any code that has already resolved dependencies will hold stale references.
Use this function in test fixtures to ensure test isolation:
import pytest from dioxide import container, reset_global_container, Profile @pytest.fixture(autouse=True) def isolated_container(): container.scan(profile=Profile.TEST) yield reset_global_container()
For most testing scenarios, consider using
dioxide.testing.fresh_container()instead, which creates completely isolated Container instances:from dioxide.testing import fresh_container async def test_something(): async with fresh_container(profile=Profile.TEST) as c: service = c.resolve(MyService) # ... test with isolated container
- Returns:
None
- Return type:
None
Example
>>> from dioxide import container, reset_global_container, service >>> >>> @service ... class MyService: ... pass >>> >>> container.scan() >>> assert not container.is_empty() >>> reset_global_container() >>> assert container.is_empty()
See also
Container.reset(): Clears singleton cache but preserves registrationsdioxide.testing.fresh_container(): Creates isolated container instances
- exception dioxide.AdapterNotFoundError(message='', *, port=None, profile=None, available_adapters=None)[source]¶
Bases:
ResolutionErrorRaised when no adapter is registered for a port in the active profile.
This error occurs when trying to resolve a Protocol or ABC (port) but no concrete implementation (adapter) is registered for the current profile. It indicates a profile mismatch or missing adapter registration.
In hexagonal architecture, ports are abstract interfaces (Protocols/ABCs) and adapters are concrete implementations. The container injects the active adapter based on the current profile. This error means: 1. No adapter exists for this port + profile combination, OR 2. An adapter exists but for a different profile, OR 3. The adapter wasn’t imported before container.scan()
- The error message includes:
Port type: Which Protocol/ABC couldn’t be resolved
Active profile: Current profile from container.scan(profile=…)
Available adapters: List of adapters for this port in other profiles
Registration hint: Code example showing how to fix
- When This Occurs:
container.resolve(PortType)- Port has no adapter for active profilecontainer.resolve(ServiceType)- Service depends on port with no adaptercontainer.start()- Lifecycle component depends on port with no adapter
- Common Causes:
Profile mismatch: Adapter registered for PRODUCTION, scanning TEST
Missing test adapter: Production adapter exists, no TEST fake created
Typo in profile name: ‘test’ vs ‘testing’ (case-insensitive)
Adapter not imported: Decorator not executed before scan()
Forgot @adapter.for_() decorator: Class exists but not registered
Examples
Profile mismatch (most common):
from typing import Protocol from dioxide import Container, adapter, Profile class EmailPort(Protocol): async def send(self, to: str, subject: str, body: str) -> None: ... # Only production adapter registered @adapter.for_(EmailPort, profile=Profile.PRODUCTION) class SendGridAdapter: async def send(self, to: str, subject: str, body: str) -> None: pass container = Container() container.scan(profile=Profile.TEST) # Scanning TEST profile try: container.resolve(EmailPort) # No TEST adapter! except AdapterNotFoundError as e: print(e) # Output: # No adapter registered for port EmailPort with profile 'test'. # # Available adapters for EmailPort: # SendGridAdapter (profiles: production) # # Hint: Add an adapter for profile 'test': # @adapter.for_(EmailPort, profile='test') # Solution: Add TEST adapter @adapter.for_(EmailPort, profile=Profile.TEST) class FakeEmailAdapter: def __init__(self): self.sent_emails = [] async def send(self, to: str, subject: str, body: str) -> None: self.sent_emails.append({'to': to, 'subject': subject, 'body': body})
Missing adapter completely:
class DatabasePort(Protocol): async def query(self, sql: str) -> list[dict]: ... @service class UserService: def __init__(self, db: DatabasePort): # Depends on DatabasePort self.db = db container = Container() container.scan(profile=Profile.PRODUCTION) try: container.resolve(UserService) # UserService needs DatabasePort except AdapterNotFoundError as e: print(e) # Output: # No adapter registered for port DatabasePort with profile 'production'. # # No adapters registered for DatabasePort. # # Hint: Register an adapter: # @adapter.for_(DatabasePort, profile='production') # class YourAdapter: # ... # Solution: Register adapter @adapter.for_(DatabasePort, profile=Profile.PRODUCTION) class PostgresAdapter: async def query(self, sql: str) -> list[dict]: pass
Universal adapter (works in all profiles):
@adapter.for_(LoggerPort, profile=Profile.ALL) # '*' means all profiles class ConsoleLogger: def log(self, message: str) -> None: print(message) # Works with any profile container.scan(profile=Profile.TEST) logger = container.resolve(LoggerPort) # Success!
- Troubleshooting:
Check profile: Verify
container.scan(profile=X)matches adapter profileList available: Look at “Available adapters” section in error message
Check imports: Ensure adapter module is imported before scan()
Verify decorator: Check
@adapter.for_(Port, profile=...)is presentUse Profile enum: Prefer
Profile.TESTover string'test'Case-insensitive: ‘Test’, ‘TEST’, ‘test’ all match (normalized to lowercase)
- Best Practices:
Create fake adapters: Every production adapter needs a test fake
Use Profile.ALL sparingly: Only for truly universal adapters (logging, etc.)
Fail fast: Resolve all services at startup to catch missing adapters early
Explicit profiles: Use
Profileenum instead of stringsImport all adapters: Use
container.scan(package="myapp")for auto-import
See also
dioxide.adapter.adapter- How to register adaptersdioxide.container.Container.scan- Profile-based scanningdioxide.container.Container.resolve- Where this is raiseddioxide.profile_enum.Profile- Standard profile values
- Parameters:
- exception dioxide.CaptiveDependencyError(message='', *, parent=None, parent_scope=None, child=None, child_scope=None)[source]¶
Bases:
DioxideErrorRaised when a longer-lived scope depends on a shorter-lived scope.
This error occurs during
container.scan()when a SINGLETON component depends on a REQUEST-scoped component. This is called a “captive dependency” because the REQUEST component would be “captured” by the SINGLETON and never refreshed, defeating the purpose of request scoping.- The problem with captive dependencies:
SINGLETON lives for the container’s lifetime
REQUEST should be fresh for each scope
If SINGLETON holds REQUEST, the same REQUEST instance is reused forever
This violates the REQUEST scope contract and causes subtle bugs
- The error message includes:
Parent component: The SINGLETON that incorrectly depends on REQUEST
Child component: The REQUEST-scoped dependency
Explanation: Why this combination is invalid
Fix suggestions: How to restructure the dependencies
- When This Occurs:
container.scan()- During dependency graph validationEarly detection prevents runtime issues
- Common Causes:
SINGLETON depends on REQUEST: Most common case
Scope mismatch: Accidentally used wrong scope on decorator
Transitive dependency: SINGLETON -> SERVICE -> REQUEST
- Valid Scope Dependencies:
SINGLETON -> SINGLETON (OK: same lifetime)
SINGLETON -> FACTORY (OK: creates new instance)
REQUEST -> SINGLETON (OK: shorter uses longer)
REQUEST -> REQUEST (OK: same scope)
REQUEST -> FACTORY (OK: creates new instance)
FACTORY -> any (OK: always creates new)
- Invalid Scope Dependencies:
SINGLETON -> REQUEST (INVALID: captive dependency)
Examples
Captive dependency detected at scan time:
from dioxide import service, Scope, Container @service(scope=Scope.REQUEST) class RequestContext: def __init__(self): self.request_id = '...' @service # SINGLETON (default) class GlobalService: def __init__(self, ctx: RequestContext): # BAD: SINGLETON -> REQUEST self.ctx = ctx container = Container() try: container.scan() except CaptiveDependencyError as e: print(e) # Output: # Captive dependency detected: GlobalService (SINGLETON) depends on # RequestContext (REQUEST). # # SINGLETON components cannot depend on REQUEST-scoped components because # the REQUEST instance would be captured and never refreshed. # # Solutions: # 1. Change GlobalService to REQUEST scope: # @service(scope=Scope.REQUEST) # 2. Change RequestContext to SINGLETON scope (if appropriate) # 3. Use a factory/provider pattern to get fresh instances
Valid dependency structure:
@service(scope=Scope.SINGLETON) class AppConfig: pass @service(scope=Scope.REQUEST) class RequestHandler: def __init__(self, config: AppConfig): # OK: REQUEST -> SINGLETON self.config = config
- Solutions:
Change parent scope:
# Make the parent REQUEST-scoped too @service(scope=Scope.REQUEST) class RequestService: def __init__(self, ctx: RequestContext): self.ctx = ctx
Change child scope (if appropriate):
# If the child doesn't truly need request scope @service # SINGLETON class SharedContext: pass
Use factory/provider pattern:
@service # SINGLETON class GlobalService: def __init__(self, container: Container): self.container = container def get_context(self) -> RequestContext: # Get fresh instance from current scope return current_scope.resolve(RequestContext)
- Best Practices:
Review scope assignments: Ensure scopes match component lifetimes
Fail fast: Error at scan() time prevents runtime surprises
Draw dependency graph: Visualize scope relationships
Default to REQUEST: For components that vary per-request
See also
dioxide.scope.Scope- Scope enum (SINGLETON, REQUEST, FACTORY)dioxide.container.Container.scan- Where this error is raisedScopeError- For runtime scope errors
- Parameters:
message (str)
parent (type | None)
parent_scope (dioxide.scope.Scope | None)
child (type | None)
child_scope (dioxide.scope.Scope | None)
- exception dioxide.CircularDependencyError(message='')[source]¶
Bases:
DioxideErrorRaised when circular dependencies are detected among @lifecycle components.
This error occurs during
container.start()when @lifecycle components have circular dependencies that prevent topological sorting. The container needs to determine initialization order (dependencies before dependents), but a cycle makes this impossible.This error ONLY applies to @lifecycle components during startup. Regular services without @lifecycle can have circular dependencies (though not recommended) because they’re instantiated lazily on-demand, not in dependency order at startup.
- A circular dependency exists when:
Component A depends on B
Component B depends on C
Component C depends on A (cycle!)
The container cannot determine which component to initialize first because each depends on another being already initialized.
- The error message includes:
Unprocessed components: Set of components involved in the cycle
Context: Which lifecycle components couldn’t be sorted
- When This Occurs:
await container.start()- During lifecycle initialization order calculationasync with container:- When entering the context manager
- Common Causes:
Direct cycle: A → B → A
Indirect cycle: A → B → C → D → A
Self-dependency: Component depends on itself (rare)
Bidirectional deps: Two components that need each other
Examples
Direct circular dependency:
from dioxide import service, lifecycle, Container @service @lifecycle class ServiceA: def __init__(self, b: 'ServiceB'): # Depends on B self.b = b async def initialize(self) -> None: pass async def dispose(self) -> None: pass @service @lifecycle class ServiceB: def __init__(self, a: ServiceA): # Depends on A - CYCLE! self.a = a async def initialize(self) -> None: pass async def dispose(self) -> None: pass container = Container() container.scan() try: await container.start() except CircularDependencyError as e: print(e) # Output: # Circular dependency detected involving: {<ServiceA>, <ServiceB>}
Indirect circular dependency:
@service @lifecycle class CacheService: def __init__(self, user_repo: UserRepository): self.user_repo = user_repo @service @lifecycle class UserRepository: def __init__(self, db: DatabaseAdapter): self.db = db @adapter.for_(DatabasePort, profile=Profile.PRODUCTION) @lifecycle class DatabaseAdapter: def __init__(self, cache: CacheService): # CYCLE! self.cache = cache # Cycle: CacheService → UserRepository → DatabaseAdapter → CacheService await container.start() # CircularDependencyError
- Solutions:
Break dependency with interface:
# Instead of depending on concrete class, depend on port class CachePort(Protocol): def get(self, key: str) -> Any: ... @service @lifecycle class ServiceA: def __init__(self, cache: CachePort): # Depend on abstraction self.cache = cache
Remove @lifecycle from one component:
# If only one component truly needs lifecycle, remove from others @service # No @lifecycle - lazy initialization class ServiceB: def __init__(self, a: ServiceA): self.a = a @service @lifecycle # Only this one has lifecycle class ServiceA: async def initialize(self) -> None: pass
Lazy resolution:
@service @lifecycle class ServiceA: def __init__(self, container: Container): self.container = container self._b = None @property def b(self) -> ServiceB: if self._b is None: self._b = self.container.resolve(ServiceB) return self._b
Redesign to remove cycle:
# Extract shared logic to a third service @service class SharedLogic: pass @service @lifecycle class ServiceA: def __init__(self, shared: SharedLogic): self.shared = shared @service @lifecycle class ServiceB: def __init__(self, shared: SharedLogic): self.shared = shared
- Troubleshooting:
Identify cycle: Look at “involving” set in error message
Map dependencies: Draw dependency graph on paper
Find weak link: Identify which dependency is least essential
Remove @lifecycle: Not all components need lifecycle management
Use abstractions: Depend on ports instead of concrete classes
Lazy initialization: Defer resolution to first use
- Best Practices:
Avoid circular dependencies: Design for acyclic dependency graphs
Use hexagonal architecture: Depend on abstractions (ports) at boundaries
Limit @lifecycle: Only use for components that truly need init/dispose
Dependency injection: Let container manage dependencies, avoid manual creation
Single Responsibility: Components with clear responsibilities rarely cycle
Test initialization: Integration test that calls
container.start()
Note
Non-lifecycle services CAN have circular dependencies (though not recommended). The container resolves them lazily on-demand. This error ONLY applies to @lifecycle components during
start()because they need explicit ordering.See also
dioxide.lifecycle.lifecycle- Lifecycle management decoratordioxide.container.Container.start- Where this error is raiseddioxide.services.service- For marking servicesdioxide.adapter.adapter- For marking adapters
- Parameters:
message (str)
- exception dioxide.DioxideError(message='')[source]¶
Bases:
ExceptionBase class for all dioxide errors with rich formatting.
DioxideError provides structured error information including: - A clear title describing the error - Context dict with relevant state at error time - Suggestions for how to fix the issue - Optional code example showing the fix - Documentation URL for detailed troubleshooting
Subclasses should set appropriate defaults for title and docs_url, and populate context, suggestions, and example based on the specific error.
- Parameters:
message (str)
- with_context(**kwargs)[source]¶
Add context information to the error.
- Parameters:
**kwargs (object) – Key-value pairs to add to the context dict.
- Returns:
Self for method chaining.
- Return type:
Self
- exception dioxide.ResolutionError(message='')[source]¶
Bases:
DioxideErrorBase class for dependency resolution failures.
ResolutionError is raised when the container cannot resolve a requested type. This is the parent class for more specific resolution errors like AdapterNotFoundError and ServiceNotFoundError.
- Parameters:
message (str)
- exception dioxide.ScopeError(message='', *, component=None, required_scope=None)[source]¶
Bases:
DioxideErrorRaised when scope-related operations fail.
This error occurs when: 1. Attempting to resolve a REQUEST-scoped component outside of a scope context 2. Attempting to create nested scopes (not supported in v0.3.0) 3. Other scope lifecycle violations
REQUEST-scoped components require an active scope created via
container.create_scope(). Attempting to resolve them from the parent container or outside any scope context will raise this error.- The error message includes:
Component type: Which component couldn’t be resolved
Scope requirement: Why a scope is needed
Fix hint: How to create a scope context
- When This Occurs:
container.resolve(RequestScopedType)- REQUEST component outside scopescope.create_scope()- Nested scope attempt (not supported)Other scope lifecycle violations
- Common Causes:
No scope context: Resolving REQUEST component from parent container
Scope not started: Scope context manager not entered
Nested scope: Trying to create scope within another scope
Examples
REQUEST component outside scope:
from dioxide import service, Scope, Container @service(scope=Scope.REQUEST) class RequestContext: pass container = Container() container.scan() try: container.resolve(RequestContext) # No scope! except ScopeError as e: print(e) # Output: # Cannot resolve RequestContext: REQUEST-scoped components require an active scope. # # Hint: Use container.create_scope() to create a scope context: # async with container.create_scope() as scope: # ctx = scope.resolve(RequestContext) # Solution: Create a scope async with container.create_scope() as scope: ctx = scope.resolve(RequestContext) # Works!
Nested scope attempt:
async with container.create_scope() as outer: try: async with outer.create_scope() as inner: # Nested! pass except ScopeError as e: print(e) # Output: # Nested scopes are not supported in v0.3.0
- Best Practices:
Create scope at entry points: Web request handlers, CLI commands, background tasks
Pass scope to dependencies: Or let container inject scoped dependencies
One scope per request: Don’t nest scopes; use one scope per logical request
Use async context manager:
async with container.create_scope() as scope:
See also
dioxide.container.Container.create_scope- How to create scopesdioxide.container.ScopedContainer- The scoped container typedioxide.scope.Scope- Scope enum including REQUEST
- Parameters:
message (str)
component (type | None)
required_scope (dioxide.scope.Scope | None)
- exception dioxide.ServiceNotFoundError(message='', *, service=None, profile=None, dependencies=None, failed_dependency=None)[source]¶
Bases:
ResolutionErrorRaised when a service or component cannot be resolved.
This error occurs when trying to resolve a service/component that either: 1. Is not registered in the container (missing
@servicedecorator), OR 2. Has dependencies that cannot be resolved (missing adapters or services), OR 3. Was not imported beforecontainer.scan()was calledUnlike AdapterNotFoundError (for ports), this error applies to concrete classes marked with
@serviceor@component. The error message helps identify whether the service itself is missing or one of its dependencies is unresolvable.- The error message includes:
Service type: Which service/component couldn’t be resolved
Active profile: Current profile (if relevant to the error)
Dependencies: Constructor parameters and their types
Missing dependency: Which specific dependency failed (if applicable)
Registration hint: Code example showing how to fix
- When This Occurs:
container.resolve(ServiceType)- Service not registered or has missing depscontainer.resolve(OtherService)- OtherService depends on unregistered servicecontainer.start()- Lifecycle component can’t be resolved
- Common Causes:
Missing @service decorator: Class not decorated, not in registry
Unresolvable dependency: Service depends on unregistered port or service
Not imported: Service module not imported before scan()
Profile mismatch on dependency: Dependency is an adapter with wrong profile
Typo in type hint: Constructor parameter references non-existent type
Examples
Service not registered:
from dioxide import Container # Forgot @service decorator! class UserService: def create_user(self, name: str): pass container = Container() container.scan() try: container.resolve(UserService) except ServiceNotFoundError as e: print(e) # Output: # Cannot resolve UserService. # # UserService is not registered in the container. # # Hint: Register the service: # @service # class UserService: # ... # Solution: Add @service decorator from dioxide import service @service class UserService: def create_user(self, name: str): pass
Service with unresolvable dependency:
from dioxide import service, Container @service class UserService: def __init__(self, db: DatabasePort): # DatabasePort not registered! self.db = db container = Container() container.scan() try: container.resolve(UserService) except ServiceNotFoundError as e: print(e) # Output: # Cannot resolve UserService. # # UserService has dependencies: db: DatabasePort # # One or more dependencies could not be resolved. # Check that all dependencies are registered with @service or @adapter.for_(). # Solution: Register adapter for DatabasePort from dioxide import adapter, Profile @adapter.for_(DatabasePort, profile=Profile.PRODUCTION) class PostgresAdapter: async def query(self, sql: str) -> list[dict]: pass
Service with multiple dependencies:
@service class NotificationService: def __init__(self, email: EmailPort, sms: SMSPort, db: DatabasePort): self.email = email self.sms = sms self.db = db # If any dependency is missing, error shows ALL dependencies try: container.resolve(NotificationService) except ServiceNotFoundError as e: # Shows: email: EmailPort, sms: SMSPort, db: DatabasePort # Helps identify which specific dependency is missing
Circular dependency (non-lifecycle):
@service class ServiceA: def __init__(self, b: 'ServiceB'): # Forward reference self.b = b @service class ServiceB: def __init__(self, a: ServiceA): self.a = a # This will cause RecursionError during resolution # (CircularDependencyError only applies to @lifecycle components) try: container.resolve(ServiceA) except RecursionError: # Redesign to break circular dependency pass
- Troubleshooting:
Check decorator: Verify
@serviceor@componentis presentVerify imports: Ensure service module is imported before scan()
Check dependencies: Look at “has dependencies” section in error message
Resolve dependencies first: Manually resolve each dependency to find which one fails
Check type hints: Ensure constructor parameters have correct type annotations
Profile mismatch: If dependency is a port, check adapter profile matches
Forward references: Use string quotes for forward references:
'ServiceB'
- Best Practices:
Fail fast: Resolve all services at startup to catch missing registrations early
Integration tests: Test that all services can be resolved in each profile
Explicit imports: Import all service modules before calling scan()
Use scan(package=”myapp”): Auto-import all modules in a package
Type hints required: Constructor parameters must have type annotations
Check profiles: Dependency adapters must match active profile
See also
dioxide.services.service- How to register servicesdioxide.adapter.adapter- How to register adapters (for dependencies)dioxide.container.Container.scan- Auto-discovery and registrationdioxide.container.Container.resolve- Where this is raisedAdapterNotFoundError- For port resolution errors
- Parameters:
- dioxide.lifecycle(cls)[source]¶
Mark a class for lifecycle management with initialization and cleanup.
The @lifecycle decorator marks a service or adapter as requiring lifecycle management, which means it needs to be initialized before use and disposed of when the application shuts down. This is essential for managing resources like database connections, caches, message queues, and other infrastructure components that require setup and teardown.
The decorator performs compile-time validation to ensure the decorated class implements the required async methods. This provides early error detection (at import time) rather than runtime failures.
- Required Methods:
The decorated class MUST implement both of these async methods:
async def initialize(self) -> None:Called once when the container starts (via
container.start()orasync with container:). Use this to establish connections, load resources, warm caches, etc. This method is called in dependency order (dependencies are initialized before their dependents).
async def dispose(self) -> None:Called once when the container stops (via
container.stop()or when exiting theasync withblock). Use this to close connections, flush buffers, release resources, etc. This method is called in reverse dependency order (dependents are disposed before their dependencies). Should be idempotent and not raise exceptions.
- Decorator Composition:
@lifecycle works with both @service and @adapter.for_() decorators. Decorator order does not affect functionality - both orderings work identically because dioxide decorators only add metadata attributes.
For consistency, we recommend @lifecycle as the innermost decorator:
@service+@lifecycle- For stateful core logic (rare)@adapter.for_()+@lifecycle- For infrastructure adapters (common)
Both orders work:
# Recommended (but both work identically) @adapter.for_(Port, profile=Profile.PRODUCTION) @lifecycle class MyAdapter: ... # Also works (not recommended for consistency) @lifecycle @adapter.for_(Port, profile=Profile.PRODUCTION) class MyAdapter: ...
- Parameters:
cls (T) – The class to mark for lifecycle management. Must implement both
initialize()anddispose()methods as async coroutines.- Returns:
The decorated class with
_dioxide_lifecycle = Trueattribute set. The class can be used normally and will be discovered by the container.- Raises:
- Return type:
T
Examples
Service with lifecycle (stateful core logic):
from dioxide import service, lifecycle @service @lifecycle class CacheWarmer: def __init__(self, db: DatabasePort): self.db = db self.cache = {} async def initialize(self) -> None: # Load all users into memory cache users = await self.db.query('SELECT * FROM users') for user in users: self.cache[user.id] = user print(f'Cache warmed with {len(users)} users') async def dispose(self) -> None: # Flush any pending writes self.cache.clear() print('Cache cleared')
Adapter with lifecycle (infrastructure connection):
from dioxide import adapter, Profile, lifecycle @adapter.for_(DatabasePort, profile=Profile.PRODUCTION) @lifecycle class PostgresAdapter: def __init__(self, config: AppConfig): self.config = config self.engine = None async def initialize(self) -> None: # Establish database connection pool self.engine = create_async_engine(self.config.database_url, pool_size=10, max_overflow=20) # Verify connection async with self.engine.connect() as conn: await conn.execute('SELECT 1') print('Database connection established') async def dispose(self) -> None: # Close all connections in pool if self.engine: await self.engine.dispose() self.engine = None print('Database connection closed') async def query(self, sql: str) -> list[dict]: async with self.engine.connect() as conn: result = await conn.execute(sql) return [dict(row) for row in result]
Multiple lifecycle components with dependencies:
# Database adapter (no dependencies) - initialized first @adapter.for_(DatabasePort, profile=Profile.PRODUCTION) @lifecycle class PostgresAdapter: async def initialize(self) -> None: self.engine = create_async_engine(...) async def dispose(self) -> None: await self.engine.dispose() # Service depends on database - initialized after database @service @lifecycle class UserRepository: def __init__(self, db: DatabasePort): self.db = db self.initialized = False async def initialize(self) -> None: # Database is already initialized at this point # Run migrations or setup await self.db.query('CREATE TABLE IF NOT EXISTS users ...') self.initialized = True async def dispose(self) -> None: self.initialized = False # Container handles dependency order automatically: # 1. PostgresAdapter.initialize() # 2. UserRepository.initialize() # ... application runs ... # 1. UserRepository.dispose() # 2. PostgresAdapter.dispose()
Validation errors at decoration time:
@service @lifecycle class BrokenService: # Missing initialize() and dispose() methods pass # Raises TypeError: BrokenService must implement initialize() method @service @lifecycle class SyncService: def initialize(self) -> None: # Not async! pass async def dispose(self) -> None: pass # Raises TypeError: SyncService.initialize() must be async
- Best Practices:
Keep initialize() fast: Avoid expensive operations, connection checks only
Make dispose() idempotent: Safe to call multiple times (check if resource exists)
Don’t raise in dispose(): Log errors but continue cleanup (best-effort)
Use for adapters: Infrastructure components at the seams (databases, queues, etc.)
Rare for services: Core domain logic is usually stateless (no lifecycle needed)
Consistent ordering: For readability, use
@adapter.for_() @lifecycle class ...(though both orders work identically)
See also
dioxide.container.Container.start- Initialize all lifecycle componentsdioxide.container.Container.stop- Dispose all lifecycle componentsdioxide.adapter.adapter- For marking infrastructure adaptersdioxide.services.service- For marking core domain servicesLifecycle Methods: Async/Sync Patterns - Async/sync patterns guide
- class dioxide.Profile[source]¶
Bases:
strExtensible, type-safe profile identifier for adapter selection.
Profile is a string subclass that provides type safety while remaining fully extensible. Built-in profiles are available as class attributes, and users can create custom profiles for their specific needs.
Built-in Profiles:
Profile.PRODUCTION- Production environmentProfile.TEST- Test environmentProfile.DEVELOPMENT- Development environmentProfile.STAGING- Staging environmentProfile.CI- Continuous integration environmentProfile.ALL- Universal profile (matches all environments)
Usage:
Use built-in profiles for common environments:
@adapter.for_(EmailPort, profile=Profile.PRODUCTION) @adapter.for_(CachePort, profile=[Profile.TEST, Profile.DEVELOPMENT]) @adapter.for_(LogPort, profile=Profile.ALL)
Create custom profiles for specific needs:
# Define custom profiles (type-safe) INTEGRATION = Profile('integration') PREVIEW = Profile('preview') LOAD_TEST = Profile('load-test') @adapter.for_(Port, profile=INTEGRATION) @adapter.for_(Port, profile=[PREVIEW, Profile.STAGING])
Type Safety:
All profiles are instances of
Profile, providing static type checking:def configure(profile: Profile) -> None: ... configure(Profile.PRODUCTION) # OK configure(Profile('custom')) # OK configure('raw-string') # Type error (if strict)
Backward Compatibility:
Profile is a
strsubclass, so it works anywhere strings are expected. Raw strings are still accepted at runtime for backward compatibility, but usingProfile(...)is recommended for type safety.Examples
>>> Profile.PRODUCTION 'production' >>> Profile.PRODUCTION == 'production' True >>> isinstance(Profile.PRODUCTION, str) True >>> Profile('custom') == 'custom' True >>> type(Profile('custom')) <class 'dioxide.profile_enum.Profile'>
- class dioxide.Scope[source]¶
-
Component lifecycle scope.
Defines how instances of a component are created and cached:
SINGLETON: One shared instance for the lifetime of the container. The factory is called once and the result is cached. Subsequent resolve() calls return the same instance.
FACTORY: New instance created on each resolve() call. The factory is invoked every time the component is requested, creating a fresh instance.
Example
>>> from dioxide import Container, component, Scope >>> >>> @component # Default: SINGLETON scope ... class Database: ... pass >>> >>> @component(scope=Scope.FACTORY) ... class RequestHandler: ... request_id: int = 0 ... ... def __init__(self): ... RequestHandler.request_id += 1 ... self.id = RequestHandler.request_id >>> >>> container = Container() >>> container.scan() >>> >>> # Singleton: same instance every time >>> db1 = container.resolve(Database) >>> db2 = container.resolve(Database) >>> assert db1 is db2 >>> >>> # Factory: new instance every time >>> handler1 = container.resolve(RequestHandler) >>> handler2 = container.resolve(RequestHandler) >>> assert handler1 is not handler2 >>> assert handler1.id != handler2.id
- SINGLETON = 'singleton'¶
One shared instance for the lifetime of the container.
The component factory is called once and the result is cached. All subsequent resolve() calls return the same instance.
Use for: - Database connections - Configuration objects - Services with shared state - Expensive-to-create objects
- FACTORY = 'factory'¶
New instance created on each resolve() call.
The component factory is invoked every time the component is requested, creating a fresh instance.
Use for: - Request handlers - Transient data objects - Stateful components that shouldn’t be shared - Objects with per-request lifecycle
- REQUEST = 'request'¶
New instance created per request scope.
Similar to FACTORY but intended for request-scoped contexts like web frameworks where the same instance should be reused within a single request but fresh instances created for each new request.
Use for: - Request-scoped services in web frameworks - Per-request database sessions - Request context objects - User authentication/authorization state per request
Note: Request scope behavior requires integration with a request context provider (e.g., FastAPI dependencies, Flask request context). Without such integration, REQUEST scope behaves like FACTORY.
- dioxide.service(cls: type[T]) type[T][source]¶
- dioxide.service(*, scope: dioxide.scope.Scope = Scope.SINGLETON) collections.abc.Callable[[type[T]], type[T]]
Mark a class as a core domain service.
Services are components that represent core business logic. They are available in all profiles (production, test, development) and support automatic dependency injection.
Key characteristics: - Uses SINGLETON scope by default (one shared instance) - Can use FACTORY scope for fresh instances per resolution - Can use REQUEST scope for per-request instances - Does not require profile specification (available everywhere) - Represents core domain logic in hexagonal architecture
- Usage:
- Basic service (SINGLETON by default):
>>> from dioxide import service >>> >>> @service ... class UserService: ... def create_user(self, name: str) -> dict: ... return {'name': name, 'id': 1}
- Service with dependencies:
>>> @service ... class EmailService: ... pass >>> >>> @service ... class NotificationService: ... def __init__(self, email: EmailService): ... self.email = email
- Factory-scoped service (new instance each time):
>>> from dioxide import service, Scope >>> >>> @service(scope=Scope.FACTORY) ... class TransactionContext: ... def __init__(self): ... self.transaction_id = str(uuid.uuid4()) >>> >>> # Each resolve() returns a fresh instance: >>> ctx1 = container.resolve(TransactionContext) >>> ctx2 = container.resolve(TransactionContext) >>> assert ctx1 is not ctx2
- Request-scoped service:
>>> from dioxide import service, Scope >>> >>> @service(scope=Scope.REQUEST) ... class RequestContext: ... def __init__(self): ... self.request_id = str(uuid.uuid4())
- Auto-discovery and resolution:
>>> from dioxide import container >>> >>> container.scan() >>> notifications = container.resolve(NotificationService) >>> assert isinstance(notifications.email, EmailService)
- Parameters:
cls – The class being decorated (when used without parentheses).
scope – The lifecycle scope for this service. Defaults to SINGLETON. - SINGLETON: One shared instance for the lifetime of the container - REQUEST: One instance per scope (via container.create_scope()) - FACTORY: New instance on every resolve()
- Returns:
The decorated class with dioxide metadata attached, or a decorator function if called with keyword arguments.
Note
Services default to SINGLETON scope
Services are available in all profiles
Dependencies are resolved from constructor (__init__) type hints
For profile-specific implementations, use @adapter.for_()
- async dioxide.fresh_container(profile=None, package=None)[source]¶
Create a fresh, isolated container for testing.
This context manager creates a new Container instance, scans for components, manages lifecycle (start/stop), and ensures complete isolation between tests.
This function does NOT require pytest to be installed.
- Parameters:
profile (dioxide.profile_enum.Profile | str | None) – Profile to scan with (e.g., Profile.TEST). If None, scans all profiles.
package (str | None) – Optional package to scan. If None, scans all registered components.
- Yields:
A fresh Container instance with lifecycle management.
- Return type:
Example
- async with fresh_container(profile=Profile.TEST) as container:
service = container.resolve(UserService) # … test with isolated container
# Container automatically cleaned up