tessl/pypi-dependency-injector

Dependency injection framework for Python

—

Pending

Overview

Eval results

Files

Schema Processing

Name: tessl/pypi-dependency-injector
Author: tessl

Schema processing utilities enable creation of containers from configuration schemas, allowing dynamic provider creation and dependency graph construction from structured data formats like JSON and YAML.

Capabilities

Schema Processor

Core class for processing container schemas and creating providers dynamically.

class SchemaProcessorV1:
    """Processes container schemas from dictionaries."""
    def __init__(self, schema):
        """
        Initialize schema processor.
        
        Parameters:
        - schema: Container schema dictionary
        """
    
    def process(self):
        """Process the schema and create providers."""
    
    def get_providers(self):
        """
        Return the created providers.
        
        Returns:
        Dictionary of provider name to provider instance mappings
        """

Schema Types

Type aliases for schema structure definitions.

ContainerSchema = Dict[Any, Any]
"""Type alias for container schema dictionary."""

ProviderSchema = Dict[Any, Any] 
"""Type alias for provider schema dictionary."""

Schema Structure

Schemas define containers and their providers in a structured format that can be loaded from configuration files.

Basic Schema Format

schema = {
    "container": {
        "provider_name": {
            "provider": "Factory",
            "provides": "myapp.services.UserService",
            "args": [
                {"provider": "database"}
            ],
            "kwargs": {
                "timeout": 30
            }
        },
        "database": {
            "provider": "Singleton", 
            "provides": "myapp.database.Database",
            "args": [
                "postgresql://localhost/mydb"
            ]
        }
    }
}

Schema Processing Example

from dependency_injector import schema, containers

# Define schema
container_schema = {
    "container": {
        "config": {
            "provider": "Configuration"
        },
        "database": {
            "provider": "Singleton",
            "provides": "myapp.database.Database",
            "args": [
                {"provider": "config.database.url"}
            ]
        },
        "user_service": {
            "provider": "Factory",
            "provides": "myapp.services.UserService", 
            "kwargs": {
                "database": {"provider": "database"}
            }
        }
    }
}

# Process schema
processor = schema.SchemaProcessorV1(container_schema)
processor.process()
providers_dict = processor.get_providers()

# Create container with processed providers
container = containers.DynamicContainer()
container.set_providers(**providers_dict)

JSON Schema Integration

Loading container schemas from JSON files.

import json
from dependency_injector import schema, containers

# Load schema from JSON file
with open("container_schema.json", "r") as f:
    schema_data = json.load(f)

# Process schema
processor = schema.SchemaProcessorV1(schema_data)
processor.process()

# Create container
container = containers.DynamicContainer()
container.set_providers(**processor.get_providers())

Example JSON schema file (container_schema.json):

{
  "container": {
    "config": {
      "provider": "Configuration"
    },
    "logger": {
      "provider": "Singleton",
      "provides": "logging.Logger",
      "args": ["myapp"]
    },
    "database": {
      "provider": "Singleton", 
      "provides": "myapp.database.Database",
      "args": [
        {"provider": "config.database.url"}
      ]
    },
    "user_repository": {
      "provider": "Factory",
      "provides": "myapp.repositories.UserRepository",
      "kwargs": {
        "database": {"provider": "database"},
        "logger": {"provider": "logger"}
      }
    },
    "user_service": {
      "provider": "Factory",
      "provides": "myapp.services.UserService",
      "kwargs": {
        "repository": {"provider": "user_repository"}
      }
    }
  }
}

YAML Schema Integration

Loading container schemas from YAML files.

import yaml
from dependency_injector import schema, containers

# Load schema from YAML file
with open("container_schema.yaml", "r") as f:
    schema_data = yaml.safe_load(f)

# Process schema
processor = schema.SchemaProcessorV1(schema_data)
processor.process()

# Create container
container = containers.DynamicContainer()
container.set_providers(**processor.get_providers())

Example YAML schema file (container_schema.yaml):

container:
  config:
    provider: Configuration
  
  logger:
    provider: Singleton
    provides: logging.Logger
    args:
      - myapp
  
  database:
    provider: Singleton
    provides: myapp.database.Database
    args:
      - provider: config.database.url
  
  redis:
    provider: Resource
    provides: myapp.resources.RedisResource
    args:
      - provider: config.redis.host
      - provider: config.redis.port
  
  user_repository:
    provider: Factory
    provides: myapp.repositories.UserRepository
    kwargs:
      database:
        provider: database
      logger:
        provider: logger
  
  user_service:
    provider: Factory
    provides: myapp.services.UserService
    kwargs:
      repository:
        provider: user_repository
      cache:
        provider: redis

Provider Type Mapping

Schema processor supports various provider types that map to dependency injector providers.

# Supported provider types in schemas
provider_types = {
    "Factory": providers.Factory,
    "Singleton": providers.Singleton,
    "ThreadSafeSingleton": providers.ThreadSafeSingleton,
    "ThreadLocalSingleton": providers.ThreadLocalSingleton,
    "Callable": providers.Callable,
    "Coroutine": providers.Coroutine,
    "Configuration": providers.Configuration,
    "Resource": providers.Resource,
    "Object": providers.Object,
    "List": providers.List,
    "Dict": providers.Dict,
    "Dependency": providers.Dependency,
    "Selector": providers.Selector
}

Advanced Schema Features

Provider References

Reference other providers within the schema:

container:
  database:
    provider: Singleton
    provides: myapp.database.Database
    args:
      - "postgresql://localhost/mydb"
  
  user_service:
    provider: Factory
    provides: myapp.services.UserService
    kwargs:
      database:
        provider: database  # Reference to database provider

Configuration References

Reference configuration values:

container:
  config:
    provider: Configuration
  
  database:
    provider: Singleton
    provides: myapp.database.Database
    args:
      - provider: config.database.url  # Reference config value

Nested Provider Arguments

Complex argument structures:

container:
  email_service:
    provider: Factory
    provides: myapp.services.EmailService
    kwargs:
      smtp_config:
        provider: Dict
        kwargs:
          host:
            provider: config.smtp.host
          port:
            provider: config.smtp.port
          username:
            provider: config.smtp.username
          password:
            provider: config.smtp.password

Container Integration

Integrate schema-created providers with declarative containers.

from dependency_injector import containers, providers, schema

class BaseContainer(containers.DeclarativeContainer):
    """Base container with core providers."""
    config = providers.Configuration()
    logger = providers.Singleton(logging.Logger, "myapp")

# Load additional providers from schema
schema_data = load_schema_from_file("additional_providers.yaml")
processor = schema.SchemaProcessorV1(schema_data)
processor.process()
additional_providers = processor.get_providers()

# Extend base container
class ApplicationContainer(BaseContainer):
    pass

# Add schema-based providers
for name, provider in additional_providers.items():
    setattr(ApplicationContainer, name, provider)

# Use combined container
container = ApplicationContainer()
container.config.from_yaml("config.yaml")

Dynamic Provider Creation

Create providers dynamically based on runtime conditions.

def create_container_from_environment():
    """Create container based on environment configuration."""
    environment = os.getenv("ENVIRONMENT", "development")
    
    # Load environment-specific schema
    schema_file = f"schemas/{environment}_schema.yaml"
    with open(schema_file, "r") as f:
        schema_data = yaml.safe_load(f)
    
    # Process schema
    processor = schema.SchemaProcessorV1(schema_data)
    processor.process()
    
    # Create container
    container = containers.DynamicContainer()
    container.set_providers(**processor.get_providers())
    
    return container

# Usage
container = create_container_from_environment()

Schema Validation

Validate schema structure before processing.

def validate_schema(schema_data):
    """Validate schema structure."""
    required_keys = ["container"]
    
    if not isinstance(schema_data, dict):
        raise ValueError("Schema must be a dictionary")
    
    if "container" not in schema_data:
        raise ValueError("Schema must contain 'container' key")
    
    container_def = schema_data["container"]
    if not isinstance(container_def, dict):
        raise ValueError("Container definition must be a dictionary")
    
    for provider_name, provider_def in container_def.items():
        if not isinstance(provider_def, dict):
            raise ValueError(f"Provider '{provider_name}' must be a dictionary")
        
        if "provider" not in provider_def:
            raise ValueError(f"Provider '{provider_name}' must specify 'provider' type")
    
    return True

# Use validation
try:
    validate_schema(schema_data)
    processor = schema.SchemaProcessorV1(schema_data)
    processor.process()
except ValueError as e:
    print(f"Schema validation error: {e}")

Error Handling

Handle errors during schema processing.

def safe_process_schema(schema_data):
    """Process schema with error handling."""
    try:
        processor = schema.SchemaProcessorV1(schema_data)
        processor.process()
        return processor.get_providers()
    except ImportError as e:
        print(f"Failed to import provider class: {e}")
        return {}
    except Exception as e:
        print(f"Schema processing error: {e}")
        return {}

# Usage
providers_dict = safe_process_schema(schema_data)
if providers_dict:
    container.set_providers(**providers_dict)
else:
    print("Using fallback configuration")

Install with Tessl CLI

npx tessl i tessl/pypi-dependency-injector

docs