tessl/pypi-dishka

Cute DI framework with scopes and agreeable API for Python dependency injection

—

Pending

Overview

Eval results

Files

Container Management

Name: tessl/pypi-dishka
Author: tessl

Core container functionality for creating, configuring, and managing dependency injection containers. Containers handle dependency creation, caching, lifecycle management, and scope transitions.

Capabilities

Container Creation

Factory functions for creating dependency injection containers with provider registration and configuration options.

def make_container(
    *providers: BaseProvider,
    scopes: type[BaseScope] = Scope,
    context: dict[Any, Any] | None = None,
    lock_factory: Callable[[], Any] = Lock,
    skip_validation: bool = False,
    start_scope: BaseScope | None = None,
    validation_settings: ValidationSettings = DEFAULT_VALIDATION
) -> Container:
    """
    Create a synchronous dependency injection container.
    
    Parameters:
    - providers: Provider instances containing dependency definitions
    - scopes: Scope hierarchy class (default: Scope)
    - context: Initial context values (deprecated)
    - lock_factory: Factory for creating thread locks
    - skip_validation: Skip dependency graph validation
    - start_scope: Initial scope to enter
    - validation_settings: Validation configuration
    
    Returns:
    Container instance ready for dependency resolution
    """

def make_async_container(
    *providers: BaseProvider,
    scopes: type[BaseScope] = Scope,
    context: dict[Any, Any] | None = None,
    lock_factory: Callable[[], AbstractAsyncContextManager[Any]] | None = Lock,
    skip_validation: bool = False,
    start_scope: BaseScope | None = None,
    validation_settings: ValidationSettings = DEFAULT_VALIDATION
) -> AsyncContainer:
    """
    Create an asynchronous dependency injection container.
    
    Parameters:
    - providers: Provider instances containing dependency definitions
    - scopes: Scope hierarchy class (default: Scope)
    - context: Initial context values (deprecated)
    - lock_factory: Factory for creating async context manager locks
    - skip_validation: Skip dependency graph validation
    - start_scope: Initial scope to enter
    - validation_settings: Validation configuration
    
    Returns:
    AsyncContainer instance ready for async dependency resolution
    """

Usage Example:

from dishka import make_container, Provider, Scope

# Create providers
provider1 = Provider()
provider2 = Provider(scope=Scope.APP)

# Create container
container = make_container(
    provider1, 
    provider2,
    skip_validation=False,
    start_scope=Scope.APP
)

Synchronous Container

Main container class for synchronous dependency injection with lifecycle management and scope handling.

class Container:
    def __init__(
        self,
        registry: Registry,
        *child_registries: Registry,
        parent_container: Container | None = None,
        context: dict[Any, Any] | None = None,
        lock_factory: Callable[[], Any] = Lock,
        close_parent: bool = False
    ): ...
    
    def get(self, dependency_type: type[T], component: str = DEFAULT_COMPONENT) -> T:
        """
        Resolve a dependency by type and optional component.
        
        Parameters:
        - dependency_type: The type to resolve
        - component: Component identifier for isolation (default: "")
        
        Returns:
        Instance of the requested type
        
        Raises:
        - NoFactoryError: No factory found for the dependency
        - DependencyCycleError: Circular dependency detected
        """
    
    def __call__(
        self,
        context: dict[Any, Any] | None = None,
        lock_factory: Callable[[], Any] | None = None,
        scope: BaseScope | None = None
    ) -> ContextWrapper:
        """
        Enter a child scope, returning a context manager.
        
        Parameters:
        - context: Additional context values (deprecated)
        - lock_factory: Lock factory for the child container
        - scope: Specific scope to enter (default: next in hierarchy)
        
        Returns:
        ContextWrapper that provides access to child container
        """
    
    def close(self, exception: BaseException | None = None) -> None:
        """
        Close the container and clean up resources.
        
        Parameters:
        - exception: Exception that caused the close (for cleanup context)
        """
    
    @property
    def scope(self) -> BaseScope:
        """Current scope of the container"""
    
    @property 
    def context(self) -> MutableMapping[DependencyKey, Any]:
        """Context mapping (deprecated, use from_context instead)"""

Usage Example:

# Get dependencies
client = container.get(SomeClient)  # APP-scoped dependency
client2 = container.get(SomeClient)  # Same instance

# Enter child scope
with container() as request_container:
    service = request_container.get(Service)  # REQUEST-scoped
    service2 = request_container.get(Service)  # Same instance in this scope
# Automatically closed when exiting context

# Clean up
container.close()

Asynchronous Container

Asynchronous container class providing the same functionality as Container but with async/await support.

class AsyncContainer:
    def __init__(
        self,
        registry: Registry,
        *child_registries: Registry,
        parent_container: AsyncContainer | None = None,
        context: dict[Any, Any] | None = None,
        lock_factory: Callable[[], Any] = Lock,
        close_parent: bool = False
    ): ...
    
    async def get(self, dependency_type: type[T], component: str = DEFAULT_COMPONENT) -> T:
        """
        Asynchronously resolve a dependency by type and optional component.
        
        Parameters:
        - dependency_type: The type to resolve
        - component: Component identifier for isolation (default: "")
        
        Returns:
        Instance of the requested type
        
        Raises:
        - NoFactoryError: No factory found for the dependency
        - DependencyCycleError: Circular dependency detected
        """
    
    def __call__(
        self,
        context: dict[Any, Any] | None = None,
        lock_factory: Callable[[], Any] | None = None,
        scope: BaseScope | None = None
    ) -> AsyncContextWrapper:
        """
        Enter a child scope, returning an async context manager.
        
        Parameters:
        - context: Additional context values (deprecated)
        - lock_factory: Lock factory for the child container
        - scope: Specific scope to enter (default: next in hierarchy)
        
        Returns:
        AsyncContextWrapper that provides access to child container
        """
    
    async def close(self, exception: BaseException | None = None) -> None:
        """
        Asynchronously close the container and clean up resources.
        
        Parameters:
        - exception: Exception that caused the close (for cleanup context)
        """
    
    @property
    def scope(self) -> BaseScope:
        """Current scope of the container"""
    
    @property
    def context(self) -> MutableMapping[DependencyKey, Any]:
        """Context mapping (deprecated, use from_context instead)"""

Usage Example:

import asyncio
from dishka import make_async_container

async def main():
    # Create async container
    container = make_async_container(provider)
    
    # Get dependencies
    client = await container.get(SomeClient)
    
    # Enter child scope
    async with container() as request_container:
        service = await request_container.get(Service)
        # Use service
    # Automatically closed when exiting context
    
    # Clean up
    await container.close()

asyncio.run(main())

Context Management

The context wrapper classes manage scope transitions and provide clean resource management.

class ContextWrapper:
    """Context manager for synchronous container scope transitions"""
    def __enter__(self) -> Container: ...
    def __exit__(self, exc_type, exc_val, exc_tb) -> None: ...

class AsyncContextWrapper:
    """Async context manager for asynchronous container scope transitions"""
    async def __aenter__(self) -> AsyncContainer: ...
    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: ...

Error Handling

Common exceptions that may be raised during container operations:

class DishkaError(Exception):
    """Base exception class for all Dishka-related errors"""

class NoFactoryError(DishkaError):
    """
    Raised when no factory is found for a requested dependency.
    
    Attributes:
    - requested: DependencyKey for the missing dependency
    - path: Sequence of dependencies that led to this request
    """

class CycleDependenciesError(DishkaError):
    """
    Raised when a circular dependency is detected in the dependency graph.
    
    Attributes:
    - path: Sequence of dependencies forming the cycle
    """

class NoChildScopesError(DishkaError):
    """Raised when trying to enter a child scope but none are available"""

class ChildScopeNotFoundError(DishkaError):
    """
    Raised when a specific child scope cannot be found.
    
    Attributes:
    - child_scope: The scope that was requested
    - current_scope: The current container scope
    """

class ExitError(DishkaError):  
    """Raised when an error occurs during container exit/cleanup"""

class NoContextValueError(DishkaError):
    """Raised when a required context value is not found"""

class InvalidGraphError(DishkaError):
    """Raised when the dependency graph is invalid or malformed"""

class UnsupportedFactoryError(DishkaError):
    """Raised when a factory type is not supported by the container"""

Install with Tessl CLI