A dependency injection library for Python with async/await support using explicit scopes instead of global state
A dependency injection library for Python that uses explicit scopes instead of global state. Provides type-safe dependency resolution with circular dependency detection.
@ainject decoratorpip install injectipy
from injectipy import inject, Inject, DependencyScope
# Create a dependency scope
scope = DependencyScope()
# Register a simple value
scope.register_value("database_url", "postgresql://localhost/mydb")
# Register a factory function
def create_database_connection(database_url: str = Inject["database_url"]):
return f"Connected to {database_url}"
scope.register_resolver("db_connection", create_database_connection)
# Use dependency injection in your functions within a scope context
@inject
def get_user(user_id: int, db_connection: str = Inject["db_connection"]):
return f"User {user_id} from {db_connection}"
# Use the scope as a context manager
with scope:
user = get_user(123)
print(user) # "User 123 from Connected to postgresql://localhost/mydb"
@ainjectInjectipy provides strict separation between sync and async dependency injection:
@inject: Only works with sync dependencies. Rejects async dependencies with clear error messages.@ainject: Designed for async functions, automatically awaits async dependencies before function execution.The @ainject decorator provides clean async dependency injection by automatically awaiting async dependencies:
import asyncio
from injectipy import ainject, Inject, DependencyScope
# Create a scope with async dependencies
scope = DependencyScope()
scope.register_value("base_url", "https://api.example.com")
scope.register_value("api_key", "secret-key")
# Register an async factory
async def create_api_client(base_url: str = Inject["base_url"], api_key: str = Inject["api_key"]):
await asyncio.sleep(0.1) # Simulate async initialization
return {"url": base_url, "key": api_key}
scope.register_async_resolver("api_client", create_api_client)
# ❌ WRONG: @inject rejects async dependencies
@inject
async def wrong_way(endpoint: str, client = Inject["api_client"]):
# This will raise AsyncDependencyError!
return await client.fetch(endpoint)
# ✅ CORRECT: Use @ainject for async dependencies
@ainject
async def correct_way(endpoint: str, client = Inject["api_client"]):
# @ainject pre-awaits async dependencies - client is ready to use!
return await client.fetch(endpoint)
async def main():
async with scope:
try:
await wrong_way("/users") # Raises AsyncDependencyError
except Exception as e:
print(f"Error: {e}")
print("Use @ainject instead!")
# This works correctly
data = await correct_way("/users")
print(data)
asyncio.run(main())
Key Benefits:
@inject guides you to use @ainject when neededhasattr(..., '__await__') checks@ainject pre-resolves all dependencies before function executionfrom injectipy import inject, Inject, DependencyScope
# Create and configure a scope
scope = DependencyScope()
scope.register_value("db_connection", "PostgreSQL://localhost")
scope.register_value("config", "production_config")
scope.register_value("helper", "UtilityHelper")
class UserService:
@inject
def __init__(self, db_connection: str = Inject["db_connection"]):
self.db = db_connection
def get_user(self, user_id: int):
return f"User {user_id} from {self.db}"
@inject
@classmethod
def create_service(cls, config: str = Inject["config"]):
return cls()
@inject
@staticmethod
def utility_function(helper: str = Inject["helper"]):
return f"Helper: {helper}"
# Use within scope context for dependency injection
with scope:
service = UserService()
print(service.get_user(456))
from injectipy import inject, Inject, DependencyScope
# Create and configure a scope
scope = DependencyScope()
# Register configuration
scope.register_value("api_key", "secret123")
scope.register_value("base_url", "https://api.example.com")
# Factory function that depends on other registered dependencies
def create_api_client(
api_key: str = Inject["api_key"],
base_url: str = Inject["base_url"]
):
return f"APIClient(key={api_key}, url={base_url})"
# Register the factory
scope.register_resolver("api_client", create_api_client)
# Use in your code within scope context
@inject
def fetch_data(client = Inject["api_client"]):
return f"Fetching data with {client}"
with scope:
print(fetch_data())
evaluate_oncefrom injectipy import DependencyScope
import time
def expensive_resource():
print("Creating expensive resource...")
time.sleep(1) # Simulate expensive operation
return "ExpensiveResource"
# Create scope and register with evaluate_once=True for singleton behavior
scope = DependencyScope()
scope.register_resolver(
"expensive_resource",
expensive_resource,
evaluate_once=True
)
with scope:
# First access creates the resource
resource1 = scope["expensive_resource"] # Prints "Creating..."
resource2 = scope["expensive_resource"] # No print, reuses cached
assert resource1 is resource2 # Same instance
The @inject decorator supports keyword-only parameters:
from injectipy import inject, Inject, DependencyScope
# Create scope and register dependencies
scope = DependencyScope()
scope.register_value("database", "ProductionDB")
scope.register_value("cache", "RedisCache")
@inject
def process_data(data: str, *, db=Inject["database"], cache=Inject["cache"], debug=False):
return f"Processing {data} with {db}, {cache}, debug={debug}"
with scope:
# Keyword-only parameters work seamlessly
result = process_data("user_data")
print(result) # "Processing user_data with ProductionDB, RedisCache, debug=False"
# Override specific parameters
result = process_data("user_data", cache="MemoryCache", debug=True)
print(result) # "Processing user_data with ProductionDB, MemoryCache, debug=True"
The @inject decorator works with other Python decorators. Order matters:
from injectipy import inject, Inject, DependencyScope
# Create scope and register dependencies
scope = DependencyScope()
scope.register_value("logger", "ProductionLogger")
class APIService:
# ✅ Recommended order: @inject comes after @classmethod/@staticmethod
@inject
@classmethod
def create_from_config(cls, logger=Inject["logger"]):
return cls(logger)
@inject
@staticmethod
def validate_data(data, logger=Inject["logger"]):
print(f"Validating with {logger}")
return True
# Works with other decorators too
def timer_decorator(func):
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
return f"timed({result})"
return wrapper
@timer_decorator
@inject
def process_data(data, logger=Inject["logger"]):
return f"Processed {data} with {logger}"
with scope:
result = process_data("user_data")
print(result) # "timed(Processed user_data with ProductionLogger)"
Decorator ordering rules:
@inject comes after @classmethod or @staticmethod@inject comes after other decorators (@contextmanager, @lru_cache, @property)@inject last (closest to the function definition)# Correct order
@classmethod
@inject
def create_service(cls, dep=Inject["service"]): ...
@lru_cache(maxsize=128)
@inject
def cached_func(dep=Inject["service"]): ...
Use types directly as keys for enhanced type safety:
from typing import Protocol
from injectipy import inject, Inject, DependencyScope
class DatabaseProtocol(Protocol):
def query(self, sql: str) -> list: ...
class PostgreSQLDatabase:
def query(self, sql: str) -> list:
return ["result1", "result2"]
class CacheService:
def get(self, key: str) -> str | None:
return f"cached_{key}"
class ConfigService:
def __init__(self, env: str):
self.env = env
def get_database_url(self) -> str:
return f"postgresql://localhost/{self.env}"
# Register dependencies using types as keys
scope = DependencyScope()
scope.register_value(DatabaseProtocol, PostgreSQLDatabase())
scope.register_value(CacheService, CacheService())
scope.register_value(ConfigService, ConfigService("production"))
@inject
def process_user(
user_id: int,
db: DatabaseProtocol = Inject[DatabaseProtocol],
cache: CacheService = Inject[CacheService],
config: ConfigService = Inject[ConfigService]
) -> str:
users = db.query("SELECT * FROM users WHERE id = ?")
cached_data = cache.get(f"user_{user_id}")
db_url = config.get_database_url()
return f"User data: {users}, cached: {cached_data}, db: {db_url}"
with scope:
# Full type safety - mypy knows exact types
result = process_user(123)
You can also use string keys for more flexible scenarios:
from typing import Protocol
from injectipy import inject, Inject, DependencyScope
class DatabaseProtocol(Protocol):
def query(self, sql: str) -> list: ...
class PostgreSQLDatabase:
def query(self, sql: str) -> list:
return ["result1", "result2"]
# Create scope and register with string keys
scope = DependencyScope()
scope.register_value("database", PostgreSQLDatabase())
scope.register_value("app_name", "MyApp")
@inject
def get_users(db: DatabaseProtocol = Inject["database"], app: str = Inject["app_name"]) -> list:
print(f"Querying from {app}")
return db.query("SELECT * FROM users")
with scope:
# mypy will verify types correctly
users: list = get_users()
You can create multiple isolated scopes and even nest them:
from injectipy import DependencyScope, inject, Inject
# Create separate scopes for different contexts
production_scope = DependencyScope()
production_scope.register_value("config", {"env": "production"})
production_scope.register_value("db_url", "postgresql://prod-server/db")
test_scope = DependencyScope()
test_scope.register_value("config", {"env": "test"})
test_scope.register_value("db_url", "sqlite:///:memory:")
@inject
def get_environment(config: dict = Inject["config"]):
return config["env"]
# Use different scopes for different contexts
with production_scope:
print(get_environment()) # "production"
with test_scope:
print(get_environment()) # "test"
# Scopes can also be nested - inner scope takes precedence
with production_scope:
with test_scope:
print(get_environment()) # "test" (inner scope wins)
The library raises clear error messages for common issues:
from injectipy import inject, ainject, Inject, DependencyScope
from injectipy import (
DependencyNotFoundError,
CircularDependencyError,
DuplicateRegistrationError,
AsyncDependencyError # New!
)
# Missing dependency
@inject
def missing_dep(value: str = Inject["nonexistent"]):
return value
try:
missing_dep()
except DependencyNotFoundError as e:
print(e) # "Dependency 'nonexistent' not found"
# Async dependency with @inject (NEW!)
async def async_service():
return "AsyncService"
scope = DependencyScope()
scope.register_async_resolver("async_service", async_service)
@inject
def wrong_decorator(service = Inject["async_service"]):
return service
with scope:
try:
wrong_decorator()
except AsyncDependencyError as e:
print(e) # "Cannot use @inject with async dependency 'async_service'. Use @ainject instead."
# Circular dependency (detected at registration)
def service_a(b = Inject["service_b"]):
return f"A: {b}"
def service_b(a = Inject["service_a"]):
return f"B: {a}"
scope = DependencyScope()
scope.register_resolver("service_a", service_a)
try:
scope.register_resolver("service_b", service_b)
except CircularDependencyError as e:
print(e) # "Circular dependency detected"
# Duplicate registration
scope = DependencyScope()
scope.register_value("config", "prod")
try:
scope.register_value("config", "dev")
except DuplicateRegistrationError as e:
print(e) # "Key 'config' already registered"
Use separate scopes for test isolation:
import pytest
from injectipy import DependencyScope, inject, Inject
@pytest.fixture
def test_scope():
"""Provide a clean scope for each test"""
return DependencyScope()
def test_dependency_injection(test_scope):
test_scope.register_value("test_value", "hello")
@inject
def test_function(value: str = Inject["test_value"]):
return value
with test_scope:
assert test_function() == "hello"
def test_isolation(test_scope):
# Each test gets a fresh scope, so dependencies are isolated
test_scope.register_value("isolated_value", "test_specific")
@inject
def isolated_function(value: str = Inject["isolated_value"]):
return value
with test_scope:
assert isolated_function() == "test_specific"
def test_scoped_mocking(test_scope):
# Easy to mock dependencies per test
test_scope.register_value("database", "MockDatabase")
test_scope.register_value("cache", "MockCache")
@inject
def service_function(db=Inject["database"], cache=Inject["cache"]):
return f"Using {db} and {cache}"
with test_scope:
result = service_function()
assert result == "Using MockDatabase and MockCache"
Scopes are thread-safe and can be shared between threads:
import threading
from injectipy import DependencyScope, inject, Inject
# Create a shared scope
shared_scope = DependencyScope()
shared_scope.register_value("shared_resource", "ThreadSafeResource")
@inject
def worker_function(resource: str = Inject["shared_resource"]):
return f"Worker using {resource}"
def worker():
with shared_scope: # Each thread uses the same scope safely
print(worker_function())
# Safe to use across multiple threads
threads = []
for i in range(10):
thread = threading.Thread(target=worker)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
DependencyScope supports both sync and async context managers:
import asyncio
from injectipy import DependencyScope, inject, Inject
scope = DependencyScope()
scope.register_value("api_key", "secret-key")
@inject
async def fetch_data(endpoint: str, api_key: str = Inject["api_key"]) -> dict:
# Simulate async API call
await asyncio.sleep(0.1)
return {"endpoint": endpoint, "authenticated": bool(api_key)}
async def main():
async with scope: # Use async context manager
data = await fetch_data("/users")
print(data)
asyncio.run(main())
Each task gets proper context isolation:
async def concurrent_example():
async def task_with_scope(task_id: int):
task_scope = DependencyScope()
task_scope.register_value("task_id", task_id)
async with task_scope:
@inject
async def process_task(task_id: int = Inject["task_id"]) -> str:
await asyncio.sleep(0.1)
return f"Processed task {task_id}"
return await process_task()
# Run multiple tasks concurrently with proper isolation
results = await asyncio.gather(
task_with_scope(1),
task_with_scope(2),
task_with_scope(3)
)
print(results) # ['Processed task 1', 'Processed task 2', 'Processed task 3']
asyncio.run(concurrent_example())
@inject decoratorDecorates functions/methods to enable automatic dependency injection within active scopes. Only works with sync dependencies - rejects async dependencies with AsyncDependencyError.
@ainject decoratorDecorates async functions to enable automatic dependency injection with proper async/await handling. Automatically awaits async dependencies before function execution.
Inject[key]Type-safe dependency marker for function parameters.
DependencyScopeContext manager for managing dependency lifecycles and isolation.
register_value(key, value)Register a static value as a dependency. Returns self for method chaining.
register_resolver(key, resolver, *, evaluate_once=False)Register a sync factory function as a dependency. Returns self for method chaining.
evaluate_once=True: Cache the result after first evaluation (singleton pattern)register_async_resolver(key, async_resolver, *, evaluate_once=False)Register an async factory function as a dependency. Returns self for method chaining.
evaluate_once=True: Cache the result after first evaluation (singleton pattern)@ainject decorator for clean async dependency injection[key] (getitem)Resolve and return a dependency by key. Only works within active scope context.
contains(key)Check if a dependency key is registered in this scope.
is_active()Check if this scope is currently active (within a with block).
__enter__(): Activate the scope__exit__(): Deactivate the scope and clean upFull documentation is available at wimonder.github.io/injectipy.
See CONTRIBUTING.md for development setup and contribution guidelines.
# Clone the repository
git clone https://github.com/Wimonder/injectipy.git
cd injectipy
# Install dependencies
poetry install
# Run tests
poetry run pytest
# Run type checking
poetry run mypy injectipy/
# Run linting
poetry run ruff check .
# Run all tests
poetry run pytest
# Run with coverage
poetry run pytest --cov=injectipy
# Run specific test files
poetry run pytest tests/test_core_inject.py
poetry run pytest tests/test_scope_functionality.py
This project is licensed under the MIT License - see the LICENSE file for details.
@ainject decorator for clean async dependency injectionAsyncDependencyError with helpful error messages guiding users to correct decorator@inject now strictly rejects async dependencies (use @ainject instead)@inject decorator, Inject[key] markers, DependencyScope context managersSee CHANGELOG.md for complete details.
FAQs
A dependency injection library for Python with async/await support using explicit scopes instead of global state
We found that injectipy demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Socket CTO Ahmad Nassri discusses why supply chain attacks now target developer machines and what AI means for the future of enterprise security.
Security News
Learn the essential steps every developer should take to stay secure on npm and reduce exposure to supply chain attacks.
Security News
Experts push back on new claims about AI-driven ransomware, warning that hype and sponsored research are distorting how the threat is understood.