dioxide.container¶

Profile-based dependency injection container with lifecycle management.

The Container class is the heart of dioxide’s dependency injection system, providing profile-based component scanning, automatic dependency resolution, and opt-in lifecycle management for services and adapters.

In hexagonal architecture, the container serves as the composition root where you wire together services (core domain logic) and adapters (infrastructure implementations). By using profiles, you can swap out infrastructure implementations based on environment (production vs test vs development) without changing core logic.

Key Features:

Profile-based scanning: Activate different adapters per environment
Automatic dependency injection: Constructor parameters resolved via type hints
Lifecycle management: Optional initialize/dispose for infrastructure resources
Type-safe resolution: Full mypy support with IDE autocomplete
Port-based resolution: Resolve abstract ports, get active adapter
Singleton caching: Shared instances managed by high-performance Rust core
Async context manager: Automatic lifecycle with async with container:

Architecture Overview:

dioxide implements hexagonal architecture (ports and adapters pattern):

Ports: Abstract interfaces (Protocols/ABCs) defining contracts
Adapters: Concrete implementations of ports (infrastructure at the seams)
Services: Core domain logic depending on ports (not concrete adapters)
Container: Composition root that wires services to adapters based on profile

The container ensures services remain decoupled from infrastructure by: 1. Services declare dependencies on ports (abstractions) 2. Adapters register as implementations for ports with profiles 3. Container injects the active adapter when resolving the port 4. Tests use fast fake adapters, production uses real infrastructure

Profile System:

Profiles determine which adapter implementations are active:

Profile.PRODUCTION: Real infrastructure (SendGrid, PostgreSQL, Redis, etc.)
Profile.TEST: Fast fakes for testing (in-memory, no network calls)
Profile.DEVELOPMENT: Developer-friendly implementations (console, files, etc.)
Profile.STAGING: Staging environment configurations
Profile.CI: Continuous integration environment
Profile.ALL ('*'): Available in all profiles (universal adapters)

Services are profile-agnostic (available in ALL profiles) while adapters are profile-specific. This enables swapping infrastructure without changing domain logic.

Basic Example:

Automatic discovery with profile-based adapters:

from typing import Protocol
from dioxide import Container, adapter, service, Profile


# Port (interface) - defines contract
class EmailPort(Protocol):
    async def send(self, to: str, subject: str, body: str) -> None: ...


# Production adapter - real SendGrid
@adapter.for_(EmailPort, profile=Profile.PRODUCTION)
class SendGridAdapter:
    async def send(self, to: str, subject: str, body: str) -> None:
        # Real SendGrid API calls
        async with httpx.AsyncClient() as client:
            await client.post('https://api.sendgrid.com/v3/mail/send', ...)


# Test adapter - fast fake
@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})


# Service - depends on PORT, not concrete adapter
@service
class UserService:
    def __init__(self, email: EmailPort):
        self.email = email  # Container injects active adapter

    async def register(self, email_addr: str, name: str):
        # Core logic - doesn't know which adapter is active
        await self.email.send(email_addr, 'Welcome!', f'Hello {name}!')


# Production container - uses SendGridAdapter
prod_container = Container()
prod_container.scan(profile=Profile.PRODUCTION)
prod_service = prod_container.resolve(UserService)
# prod_service.email is SendGridAdapter

# Test container - uses FakeEmailAdapter
test_container = Container()
test_container.scan(profile=Profile.TEST)
test_service = test_container.resolve(UserService)
# test_service.email is FakeEmailAdapter

# Tests run fast with fakes, production uses real infrastructure
# Core domain logic (UserService) stays the same

Lifecycle Management Example:

Initialize and dispose resources automatically:

from dioxide import Container, adapter, lifecycle, Profile
from sqlalchemy.ext.asyncio import create_async_engine


@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:
        # Called automatically when container starts
        self.engine = create_async_engine(self.config.database_url)
        print('Database connected')

    async def dispose(self) -> None:
        # Called automatically when container stops
        if self.engine:
            await self.engine.dispose()
        print('Database disconnected')

    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]


# Manual lifecycle control
container = Container()
container.scan(profile=Profile.PRODUCTION)
await container.start()  # Calls PostgresAdapter.initialize()
db = container.resolve(DatabasePort)
users = await db.query('SELECT * FROM users')
await container.stop()  # Calls PostgresAdapter.dispose()

# Async context manager (recommended)
async with Container() as container:
    container.scan(profile=Profile.PRODUCTION)
    # PostgresAdapter.initialize() called here
    db = container.resolve(DatabasePort)
    users = await db.query('SELECT * FROM users')
# PostgresAdapter.dispose() called here (even if exception raised)

Advanced Example:

Multiple adapters with dependencies and lifecycle:

from dioxide import Container, adapter, service, lifecycle, Profile


# Cache adapter (no dependencies) - initialized first
@adapter.for_(CachePort, profile=Profile.PRODUCTION)
@lifecycle
class RedisCache:
    async def initialize(self) -> None:
        self.redis = await aioredis.create_redis_pool('redis://localhost')

    async def dispose(self) -> None:
        self.redis.close()
        await self.redis.wait_closed()

    async def get(self, key: str) -> str | None:
        return await self.redis.get(key)

    async def set(self, key: str, value: str) -> None:
        await self.redis.set(key, value)


# 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('postgresql://...')

    async def dispose(self) -> None:
        await self.engine.dispose()

    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]


# Service depends on cache and database - initialized last
@service
@lifecycle
class UserRepository:
    def __init__(self, cache: CachePort, db: DatabasePort):
        self.cache = cache
        self.db = db

    async def initialize(self) -> None:
        # Both adapters are already initialized
        # Warm cache with users from database
        users = await self.db.query('SELECT * FROM users')
        for user in users:
            await self.cache.set(f'user:{user["id"]}', user['email'])

    async def dispose(self) -> None:
        # Flush any pending operations
        pass


# Container manages initialization order automatically:
# 1. RedisCache.initialize()
# 2. PostgresAdapter.initialize()
# 3. UserRepository.initialize() (after dependencies ready)
# ... application runs ...
# 1. UserRepository.dispose() (before dependencies)
# 2. PostgresAdapter.dispose()
# 3. RedisCache.dispose()

async with Container() as container:
    container.scan(profile=Profile.PRODUCTION)
    repo = container.resolve(UserRepository)
    # All @lifecycle components initialized in dependency order
    user = await repo.find_by_email('alice@example.com')
# All @lifecycle components disposed in reverse dependency order

Testing Example:

Fast tests with fake adapters:

import pytest
from dioxide import Container, Profile


@pytest.fixture
async def container():
    async with Container() as c:
        c.scan(profile=Profile.TEST)
        # Fast fake adapters initialized (no real infrastructure)
        yield c
    # Cleanup happens automatically


async def test_user_registration(container):
    # Arrange
    service = container.resolve(UserService)
    email = container.resolve(EmailPort)  # FakeEmailAdapter

    # Act
    await service.register('alice@example.com', 'Alice')

    # Assert - check observable outcomes using fake's state
    assert len(email.sent_emails) == 1
    assert email.sent_emails[0]['to'] == 'alice@example.com'
    assert 'Welcome' in email.sent_emails[0]['subject']


# Tests run in milliseconds, no network calls, fully isolated

Global Container Instance:

For simple scripts and CLI tools, use the global singleton container:

from dioxide import container, Profile

# Setup once at application startup
container.scan(profile=Profile.PRODUCTION)

# Resolve services anywhere in your app
user_service = container.resolve(UserService)

# With lifecycle
async with container:
    # All @lifecycle components initialized
    await app.run()
# All @lifecycle components disposed

Note

For testing, libraries, and larger applications, prefer instance containers (Container()) over the global container. Instance containers provide better isolation and are easier to test. See the user guide for details: /docs/user_guide/container_patterns

Manual Registration Example:

from dioxide import Container


class Config:
    def __init__(self, env: str):
        self.env = env


container = Container()

# Register pre-created instance
config = Config('production')
container.register_instance(Config, config)

# Register singleton factory
container.register_singleton(Logger, lambda: Logger(config))

# Register transient factory (new instance each time)
container.register_factory(RequestContext, lambda: RequestContext())

# Resolve components
config = container.resolve(Config)
logger = container.resolve(Logger)

Security:

Restrict which packages can be scanned to prevent code execution:

# Only allow scanning within your application packages
container = Container(allowed_packages=['myapp', 'tests'])
container.scan(package='myapp.services')  # OK
container.scan(package='os')  # Raises ValueError

Error Handling:

Descriptive errors with troubleshooting hints:

from dioxide.exceptions import AdapterNotFoundError, ServiceNotFoundError

try:
    container.scan(profile=Profile.PRODUCTION)
    email = container.resolve(EmailPort)
except AdapterNotFoundError as e:
    # Shows:
    # - Which port couldn't be resolved
    # - Active profile
    # - Available adapters for other profiles
    # - How to register an adapter for this profile
    print(e)

try:
    service = container.resolve(UnregisteredService)
except ServiceNotFoundError as e:
    # Shows:
    # - Which service couldn't be resolved
    # - Missing dependencies
    # - How to register the service
    print(e)

Best Practices:

One container per application: Create once at startup, reuse everywhere
Use profiles: Swap infrastructure, keep domain logic unchanged
Global container for simplicity: Import from dioxide import container
Separate containers for testing: Isolated test containers per test
Lifecycle for adapters: Infrastructure resources need init/dispose
Services rarely need lifecycle: Core logic is usually stateless
Async context manager: async with container: handles lifecycle automatically

Thread Safety:

The global container singleton (from dioxide import container) is thread-safe for most common usage patterns:

Why it’s safe:

Module import guarantee: Python’s import system ensures modules are initialized exactly once, even when multiple threads import simultaneously. The GIL (Global Interpreter Lock) serializes module initialization, so container: Container = Container() executes atomically.
Singleton access: Once initialized, accessing the global container variable is a simple attribute lookup, which is atomic under the GIL.
Rust-backed resolution: The underlying Rust container uses thread-safe data structures for provider registration and singleton caching.

Safe operations (no external synchronization needed):

Importing: from dioxide import container
Resolving singletons: container.resolve(MyService)
Scanning at startup: container.scan(profile=Profile.PRODUCTION)

Best practices for multi-threaded applications:

Call container.scan() once during application startup, before spawning threads
Resolve services after scanning is complete
For per-thread isolation (e.g., request-scoped state), create separate Container instances or use container.create_scope()

When to use separate containers:

Multi-tenant applications requiring isolated dependency graphs
Testing scenarios requiring complete isolation
Per-request scoping in web frameworks (consider create_scope() first)

Example (multi-threaded web application):

import threading
from dioxide import container, Profile

# Startup: scan once before threads start
container.scan(profile=Profile.PRODUCTION)


def handle_request():
    # Safe: resolving from already-scanned container
    service = container.resolve(UserService)
    return service.process()


# Multiple threads can safely resolve from the same container
threads = [threading.Thread(target=handle_request) for _ in range(10)]
for t in threads:
    t.start()

See also

dioxide.adapter.adapter - For marking infrastructure adapters
dioxide.services.service - For marking core domain services
dioxide.lifecycle.lifecycle - For initialization/cleanup
dioxide.profile_enum.Profile - Standard profile enum values
dioxide.exceptions.AdapterNotFoundError - Port resolution error
dioxide.exceptions.ServiceNotFoundError - Service resolution error

Attributes¶

`logger`
`T`
`container`

Classes¶

`Container`	Dependency injection container.
`ScopedContainer`	A scoped container for REQUEST-scoped dependency resolution.
`ScopedContainerContextManager`	Async context manager for ScopedContainer.

Functions¶

reset_global_container()

Reset the global container to an empty state.

Module Contents¶

dioxide.container.logger[source]¶

dioxide.container.T[source]¶

class dioxide.container.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:

allowed_packages (list[str] | None)
profile (dioxide.profile_enum.Profile | str | None)

register_instance(component_type, instance)[source]¶

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:

TypeError – If the instance is not an instance of component_type.
KeyError – If the type is already registered in this container. Each type can only be registered once.

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]¶

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:

component_type (type[T]) – The type to register. This is used as the lookup key when resolving dependencies.
implementation (type[T]) – The class to instantiate. Must have a no-argument __init__ method (or no __init__ at all).

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]¶

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]¶

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]¶

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]¶

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 with multi=True for 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:

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:: bool

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:: int

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) or Container(profile=None, ports=0, services=0) when no profile is set.
Return type:: str

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:: list[type[Any]]

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 registered
get_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:: bool

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 types
resolve() - Actually resolve a registered type

property active_profile: dioxide.profile_enum.Profile | None[source]¶

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 scanning
dioxide.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 adapter
adapter.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:: str

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:: str

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:: str

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:: Container

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:: ScopedContainerContextManager

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 type
dioxide.scope.Scope - Scope enum
dioxide.exceptions.ScopeError - Scope errors

class dioxide.container.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 @lifecycle are 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

Parameters:

parent (Container)
scope_id (str)

scope_id[source]¶

Unique identifier for this scope

Return type:: str

parent[source]¶

The parent Container

Return type:: Container

See also

Container.create_scope() - How to create scopes
dioxide.scope.Scope - Scope enum (SINGLETON, REQUEST, FACTORY)
dioxide.exceptions.ScopeError - Raised for scope violations

property scope_id: str[source]¶

Get the unique identifier for this scope.

Return type:: str

property parent: Container[source]¶

Get the parent container.

Return type:: Container

__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:: str

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:

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:: ScopedContainerContextManager

class dioxide.container.ScopedContainerContextManager(parent)[source]¶

Async context manager for ScopedContainer.

This class manages the lifecycle of a ScopedContainer, handling setup on entry and disposal on exit.

Usage:

>>> async with container.create_scope() as scope:
...     handler = scope.resolve(RequestHandler)

Parameters:: parent (Container)

async __aenter__()[source]¶

Enter the scope context.

Creates a new ScopedContainer with a unique ID.

Returns:: The newly created ScopedContainer.
Return type:: ScopedContainer

async __aexit__(exc_type, exc_val, exc_tb)[source]¶

Exit the scope context.

Disposes all REQUEST-scoped lifecycle components.

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

dioxide.container.container: Container[source]¶

dioxide.container.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 container will 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 registrations dioxide.testing.fresh_container(): Creates isolated container instances