Dependency injection framework for Python
—
Dependency wiring enables automatic injection of dependencies into functions, methods, and class attributes. It provides decorators and markers for seamless integration with existing code without requiring manual dependency resolution.
Primary functions for setting up and managing dependency injection.
def wire(container, modules=None, packages=None, from_package=None):
"""
Wire dependencies into modules and packages.
Parameters:
- container: Container with providers to inject
- modules: List of modules to wire
- packages: List of packages to wire
- from_package: Base package for relative imports
"""
def unwire(container=None, modules=None, packages=None, from_package=None):
"""
Remove dependency wiring from modules and packages.
Parameters:
- container: Container to unwire (optional)
- modules: List of modules to unwire
- packages: List of packages to unwire
- from_package: Base package for relative imports
"""
def inject(fn):
"""
Decorator for dependency injection into functions and methods.
Parameters:
- fn: Function or method to inject dependencies into
Returns:
Wrapped function with dependency injection
"""Usage example:
from dependency_injector import containers, providers
from dependency_injector.wiring import inject, Provide, wire
# Define container
class Container(containers.DeclarativeContainer):
user_service = providers.Factory(UserService)
# Wire container to current module
container = Container()
wire(container=container, modules=[__name__])
# Use dependency injection
@inject
def get_user(user_id: int, service: UserService = Provide[Container.user_service]):
return service.get_user(user_id)Markers specify what and how dependencies should be injected.
class Provide:
"""Marker for dependency injection - provides instance."""
def __class_getitem__(cls, provider): ...
class Provider:
"""Marker for provider injection - provides provider object."""
def __class_getitem__(cls, provider): ...
class Closing:
"""Marker for resource closing injection."""
def __class_getitem__(cls, provider): ...Usage examples:
# Provide instance injection
@inject
def handler(service: UserService = Provide[Container.user_service]):
return service.process()
# Provider injection (get the provider, not the instance)
@inject
def factory_handler(service_factory = Provider[Container.user_service]):
# Create multiple instances
service1 = service_factory()
service2 = service_factory()
return [service1, service2]
# Resource closing injection (for cleanup)
@inject
def process_with_cleanup(
database = Provide[Container.database],
db_resource = Closing[Container.database]
):
# Use database
result = database.query("SELECT * FROM users")
# db_resource will be closed automatically
return resultWiring supports Python type annotations for cleaner syntax.
from typing import Annotated
from dependency_injector.wiring import Provide
@inject
def typed_handler(
service: Annotated[UserService, Provide[Container.user_service]]
):
return service.process()
# Alternative syntax
@inject
def alt_handler(service: UserService = Provide[Container.user_service]):
return service.process()Modifiers for configuration providers to transform values during injection.
def as_int():
"""Return int type modifier."""
def as_float():
"""Return float type modifier."""
def as_(type_):
"""Return custom type modifier."""
def required():
"""Return required configuration modifier."""
def invariant(id):
"""Return invariant configuration modifier."""
def provided():
"""Return provided instance modifier."""Usage example:
from dependency_injector.wiring import as_int, required, provided
class Container(containers.DeclarativeContainer):
config = providers.Configuration()
database = providers.Singleton(Database, host=config.host)
@inject
def configure_app(
# Type conversion
port: int = Provide[Container.config.port, as_int()],
# Required configuration
secret_key: str = Provide[Container.config.secret_key, required()],
# Provided instance access
db_host: str = Provide[Container.database, provided().host]
):
app.config["PORT"] = port
app.config["SECRET_KEY"] = secret_key
app.config["DB_HOST"] = db_hostInject dependencies into class attributes using markers.
from dependency_injector.wiring import Provide
class UserController:
# Class attribute injection
user_service = Provide[Container.user_service]
config = Provider[Container.config]
def get_user(self, user_id: int):
return self.user_service.get_user(user_id)
def get_config_value(self, key: str):
return self.config().get(key)
# Wire the container
wire(container=container, modules=[__name__])
# Use the controller
controller = UserController()
user = controller.get_user(123)Wiring works seamlessly with async functions and coroutines.
@inject
async def async_handler(
service: AsyncUserService = Provide[Container.async_user_service]
):
result = await service.process_async()
return result
@inject
async def async_generator(
data_service: DataService = Provide[Container.data_service]
):
async for item in data_service.stream_data():
yield itemWiring integrates with popular Python web frameworks.
from fastapi import FastAPI, Depends
from dependency_injector.wiring import inject, Provide
app = FastAPI()
@app.get("/users/{user_id}")
@inject
def get_user(
user_id: int,
service: UserService = Depends(Provide[Container.user_service])
):
return service.get_user(user_id)
# Wire the container
container.wire(modules=[__name__])from flask import Flask
from dependency_injector.wiring import inject, Provide
app = Flask(__name__)
@app.route("/users/<int:user_id>")
@inject
def get_user(user_id: int, service: UserService = Provide[Container.user_service]):
return service.get_user(user_id)
# Wire the container
container.wire(modules=[__name__])from django.http import JsonResponse
from dependency_injector.wiring import inject, Provide
@inject
def user_view(request, user_id, service: UserService = Provide[Container.user_service]):
user = service.get_user(user_id)
return JsonResponse({"user": user})
# Wire in Django app configuration
container.wire(modules=["myapp.views"])Automatic wiring of containers when modules are imported.
class AutoLoader:
"""Auto-wiring module loader."""
def register_containers(self, *containers): ...
def unregister_containers(self, *containers): ...
def install(self): ...
def uninstall(self): ...
def register_loader_containers(*containers):
"""Register containers in auto-wiring module loader."""
def unregister_loader_containers(*containers):
"""Unregister containers from auto-wiring module loader."""
def install_loader():
"""Install auto-wiring module loader hook."""
def uninstall_loader():
"""Uninstall auto-wiring module loader hook."""
def is_loader_installed() -> bool:
"""Check if auto-wiring module loader hook is installed."""Usage example:
from dependency_injector.wiring import (
register_loader_containers,
install_loader
)
# Register containers for auto-loading
register_loader_containers(container)
install_loader()
# Now any imported module will be automatically wired
import myapp.handlers # Automatically wired
import myapp.services # Automatically wiredDifferent patterns for injecting dependencies into methods.
class UserController:
@inject
def get_user(
self,
user_id: int,
service: UserService = Provide[Container.user_service]
):
return service.get_user(user_id)class UserService:
@classmethod
@inject
def create_instance(
cls,
database: Database = Provide[Container.database]
):
return cls(database)class UserUtils:
@staticmethod
@inject
def validate_user(
user_data: dict,
validator: UserValidator = Provide[Container.user_validator]
):
return validator.validate(user_data)Warning and error handling for wiring issues.
class DIWiringWarning(RuntimeWarning):
"""Warning for dependency injection wiring issues."""Common wiring issues and solutions:
# Clear wiring cache if needed
from dependency_injector.wiring import clear_cache
clear_cache()
# Handle wiring warnings
import warnings
from dependency_injector.wiring import DIWiringWarning
warnings.filterwarnings("ignore", category=DIWiringWarning)@inject
def conditional_handler(
service: UserService = Provide[Container.user_service],
debug_mode: bool = Provide[Container.config.debug, as_(bool)]
):
if debug_mode:
print(f"Using service: {service}")
return service.process()@inject
def multi_service_handler(
user_service: UserService = Provide[Container.user_service],
email_service: EmailService = Provide[Container.email_service],
logger: Logger = Provide[Container.logger]
):
user = user_service.get_user(123)
email_service.send_email(user.email, "Welcome!")
logger.info(f"Sent welcome email to {user.email}")@inject
def process_with_resources(
database = Provide[Container.database],
redis = Provide[Container.redis],
db_cleanup = Closing[Container.database],
redis_cleanup = Closing[Container.redis]
):
# Use resources
data = database.query("SELECT * FROM users")
redis.set("cache_key", data)
# Resources will be automatically cleaned up
return dataAdvanced wiring functions for automatic container loading and management.
def register_loader_containers(*containers):
"""Register containers for auto-loading."""
def unregister_loader_containers(*containers):
"""Unregister containers from auto-loading."""
def install_loader():
"""Install auto-wiring loader."""
def uninstall_loader():
"""Uninstall auto-wiring loader."""
def is_loader_installed() -> bool:
"""Check if auto-wiring loader is installed."""
def clear_cache():
"""Clear wiring caches."""Install with Tessl CLI
npx tessl i tessl/pypi-dependency-injector