# Injectipy
A dependency injection library for Python that uses explicit scopes instead of global state. Provides type-safe dependency resolution with circular dependency detection.
[](https://badge.fury.io/py/injectipy)
[](https://www.python.org/downloads/)
[](https://opensource.org/licenses/MIT)
[](https://mypy.readthedocs.io/)
## Key Features
- **Explicit scopes**: Dependencies managed within context managers, no global state
- **Async/await support**: Clean async dependency injection with `@ainject` decorator
- **Type safety**: Works with mypy and provides runtime type checking
- **Circular dependency detection**: Detects dependency cycles at registration time
- **Thread safety**: Each scope is isolated, safe for concurrent use
- **Lazy evaluation**: Dependencies resolved only when accessed
- **Test isolation**: Each test can use its own scope with different dependencies
## Installation
```bash
pip install injectipy
```
## Quick Start
### Basic Usage
```python
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"
```
### Async/Await Support with `@ainject`
Injectipy 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:
```python
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:**
- **Clear separation**: No confusion about which decorator to use
- **Better error messages**: `@inject` guides you to use `@ainject` when needed
- **Type safety**: Eliminates manual `hasattr(..., '__await__')` checks
- **Clean code**: `@ainject` pre-resolves all dependencies before function execution
### Class-based Injection
```python
from 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))
```
### Factory Functions with Dependencies
```python
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())
```
### Singleton Pattern with `evaluate_once`
```python
from 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
```
## Advanced Usage
### Keyword-Only Parameters
The `@inject` decorator supports keyword-only parameters:
```python
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"
```
### Decorator Compatibility
The `@inject` decorator works with other Python decorators. Order matters:
```python
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`)
- Apply `@inject` last (closest to the function definition)
```python
# Correct order
@classmethod
@inject
def create_service(cls, dep=Inject["service"]): ...
@lru_cache(maxsize=128)
@inject
def cached_func(dep=Inject["service"]): ...
```
### Type-Based Registration and Injection
Use types directly as keys for enhanced type safety:
```python
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)
```
### String-Based Registration
You can also use string keys for more flexible scenarios:
```python
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()
```
### Scope Isolation and Nesting
You can create multiple isolated scopes and even nest them:
```python
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)
```
## Error Handling
The library raises clear error messages for common issues:
```python
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"
```
## Testing
Use separate scopes for test isolation:
```python
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"
```
## Thread Safety
Scopes are thread-safe and can be shared between threads:
```python
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()
```
## Async/Await Support
DependencyScope supports both sync and async context managers:
```python
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())
```
### Concurrent Async Tasks
Each task gets proper context isolation:
```python
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())
```
## API Reference
### Core Components
#### `@inject` decorator
Decorates functions/methods to enable automatic dependency injection within active scopes. **Only works with sync dependencies** - rejects async dependencies with `AsyncDependencyError`.
#### `@ainject` decorator
Decorates 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.
#### `DependencyScope`
Context manager for managing dependency lifecycles and isolation.
### DependencyScope Methods
#### `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)
- Use with `@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).
#### Context Manager Protocol
- `__enter__()`: Activate the scope
- `__exit__()`: Deactivate the scope and clean up
## Documentation
Full documentation is available at [wimonder.github.io/injectipy](https://wimonder.github.io/injectipy/).
## Development
See [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and contribution guidelines.
### Development Setup
```bash
# 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 .
```
### Testing
```bash
# 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
```
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Changelog
### Version 0.3.0 (2025-01-03)
- **NEW**: `@ainject` decorator for clean async dependency injection
- **NEW**: `AsyncDependencyError` with helpful error messages guiding users to correct decorator
- **BREAKING**: `@inject` now strictly rejects async dependencies (use `@ainject` instead)
- **Enhanced**: Strict separation between sync and async dependency injection
- **Performance**: Optimized async resolver detection with caching
- **Documentation**: Updated README with comprehensive async/await examples
### Version 0.1.0 (2024-01-20)
- **Initial release** of Injectipy dependency injection library
- **Core features**: `@inject` decorator, `Inject[key]` markers, `DependencyScope` context managers
- **Advanced capabilities**: Thread safety, circular dependency detection, lazy evaluation
- **Modern Python**: Python 3.11+ support with native union types
- **Developer experience**: Type safety with mypy, comprehensive testing
See [CHANGELOG.md](CHANGELOG.md) for complete details.
Raw data
{
"_id": null,
"home_page": "https://github.com/Wimonder/injectipy",
"name": "injectipy",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.11",
"maintainer_email": null,
"keywords": "dependency injection, async, asyncio, context manager, type safety",
"author": "Wim Onderbeke",
"author_email": "wim.onderbeke@hotmail.com",
"download_url": "https://files.pythonhosted.org/packages/af/b3/cab62f69e72765e67ab8c7ffc21d47c695d9b614c4ecd514cc1a89ea3ced/injectipy-0.3.0.tar.gz",
"platform": null,
"description": "# Injectipy\n\nA dependency injection library for Python that uses explicit scopes instead of global state. Provides type-safe dependency resolution with circular dependency detection.\n\n[](https://badge.fury.io/py/injectipy)\n[](https://www.python.org/downloads/)\n[](https://opensource.org/licenses/MIT)\n[](https://mypy.readthedocs.io/)\n\n## Key Features\n\n- **Explicit scopes**: Dependencies managed within context managers, no global state\n- **Async/await support**: Clean async dependency injection with `@ainject` decorator\n- **Type safety**: Works with mypy and provides runtime type checking\n- **Circular dependency detection**: Detects dependency cycles at registration time\n- **Thread safety**: Each scope is isolated, safe for concurrent use\n- **Lazy evaluation**: Dependencies resolved only when accessed\n- **Test isolation**: Each test can use its own scope with different dependencies\n\n## Installation\n\n```bash\npip install injectipy\n```\n\n## Quick Start\n\n### Basic Usage\n\n```python\nfrom injectipy import inject, Inject, DependencyScope\n\n# Create a dependency scope\nscope = DependencyScope()\n\n# Register a simple value\nscope.register_value(\"database_url\", \"postgresql://localhost/mydb\")\n\n# Register a factory function\ndef create_database_connection(database_url: str = Inject[\"database_url\"]):\n return f\"Connected to {database_url}\"\n\nscope.register_resolver(\"db_connection\", create_database_connection)\n\n# Use dependency injection in your functions within a scope context\n@inject\ndef get_user(user_id: int, db_connection: str = Inject[\"db_connection\"]):\n return f\"User {user_id} from {db_connection}\"\n\n# Use the scope as a context manager\nwith scope:\n user = get_user(123)\n print(user) # \"User 123 from Connected to postgresql://localhost/mydb\"\n```\n\n### Async/Await Support with `@ainject`\n\nInjectipy provides strict separation between sync and async dependency injection:\n\n- **`@inject`**: Only works with sync dependencies. **Rejects async dependencies with clear error messages.**\n- **`@ainject`**: Designed for async functions, automatically awaits async dependencies before function execution.\n\nThe `@ainject` decorator provides clean async dependency injection by automatically awaiting async dependencies:\n\n```python\nimport asyncio\nfrom injectipy import ainject, Inject, DependencyScope\n\n# Create a scope with async dependencies\nscope = DependencyScope()\nscope.register_value(\"base_url\", \"https://api.example.com\")\nscope.register_value(\"api_key\", \"secret-key\")\n\n# Register an async factory\nasync def create_api_client(base_url: str = Inject[\"base_url\"], api_key: str = Inject[\"api_key\"]):\n await asyncio.sleep(0.1) # Simulate async initialization\n return {\"url\": base_url, \"key\": api_key}\n\nscope.register_async_resolver(\"api_client\", create_api_client)\n\n# \u274c WRONG: @inject rejects async dependencies\n@inject\nasync def wrong_way(endpoint: str, client = Inject[\"api_client\"]):\n # This will raise AsyncDependencyError!\n return await client.fetch(endpoint)\n\n# \u2705 CORRECT: Use @ainject for async dependencies\n@ainject\nasync def correct_way(endpoint: str, client = Inject[\"api_client\"]):\n # @ainject pre-awaits async dependencies - client is ready to use!\n return await client.fetch(endpoint)\n\nasync def main():\n async with scope:\n try:\n await wrong_way(\"/users\") # Raises AsyncDependencyError\n except Exception as e:\n print(f\"Error: {e}\")\n print(\"Use @ainject instead!\")\n\n # This works correctly\n data = await correct_way(\"/users\")\n print(data)\n\nasyncio.run(main())\n```\n\n**Key Benefits:**\n- **Clear separation**: No confusion about which decorator to use\n- **Better error messages**: `@inject` guides you to use `@ainject` when needed\n- **Type safety**: Eliminates manual `hasattr(..., '__await__')` checks\n- **Clean code**: `@ainject` pre-resolves all dependencies before function execution\n\n### Class-based Injection\n\n```python\nfrom injectipy import inject, Inject, DependencyScope\n\n# Create and configure a scope\nscope = DependencyScope()\nscope.register_value(\"db_connection\", \"PostgreSQL://localhost\")\nscope.register_value(\"config\", \"production_config\")\nscope.register_value(\"helper\", \"UtilityHelper\")\n\nclass UserService:\n @inject\n def __init__(self, db_connection: str = Inject[\"db_connection\"]):\n self.db = db_connection\n\n def get_user(self, user_id: int):\n return f\"User {user_id} from {self.db}\"\n\n @inject\n @classmethod\n def create_service(cls, config: str = Inject[\"config\"]):\n return cls()\n\n @inject\n @staticmethod\n def utility_function(helper: str = Inject[\"helper\"]):\n return f\"Helper: {helper}\"\n\n# Use within scope context for dependency injection\nwith scope:\n service = UserService()\n print(service.get_user(456))\n```\n\n### Factory Functions with Dependencies\n\n```python\nfrom injectipy import inject, Inject, DependencyScope\n\n# Create and configure a scope\nscope = DependencyScope()\n\n# Register configuration\nscope.register_value(\"api_key\", \"secret123\")\nscope.register_value(\"base_url\", \"https://api.example.com\")\n\n# Factory function that depends on other registered dependencies\ndef create_api_client(\n api_key: str = Inject[\"api_key\"],\n base_url: str = Inject[\"base_url\"]\n):\n return f\"APIClient(key={api_key}, url={base_url})\"\n\n# Register the factory\nscope.register_resolver(\"api_client\", create_api_client)\n\n# Use in your code within scope context\n@inject\ndef fetch_data(client = Inject[\"api_client\"]):\n return f\"Fetching data with {client}\"\n\nwith scope:\n print(fetch_data())\n```\n\n### Singleton Pattern with `evaluate_once`\n\n```python\nfrom injectipy import DependencyScope\nimport time\n\ndef expensive_resource():\n print(\"Creating expensive resource...\")\n time.sleep(1) # Simulate expensive operation\n return \"ExpensiveResource\"\n\n# Create scope and register with evaluate_once=True for singleton behavior\nscope = DependencyScope()\nscope.register_resolver(\n \"expensive_resource\",\n expensive_resource,\n evaluate_once=True\n)\n\nwith scope:\n # First access creates the resource\n resource1 = scope[\"expensive_resource\"] # Prints \"Creating...\"\n resource2 = scope[\"expensive_resource\"] # No print, reuses cached\n\n assert resource1 is resource2 # Same instance\n```\n\n## Advanced Usage\n\n### Keyword-Only Parameters\n\nThe `@inject` decorator supports keyword-only parameters:\n\n```python\nfrom injectipy import inject, Inject, DependencyScope\n\n# Create scope and register dependencies\nscope = DependencyScope()\nscope.register_value(\"database\", \"ProductionDB\")\nscope.register_value(\"cache\", \"RedisCache\")\n\n@inject\ndef process_data(data: str, *, db=Inject[\"database\"], cache=Inject[\"cache\"], debug=False):\n return f\"Processing {data} with {db}, {cache}, debug={debug}\"\n\nwith scope:\n # Keyword-only parameters work seamlessly\n result = process_data(\"user_data\")\n print(result) # \"Processing user_data with ProductionDB, RedisCache, debug=False\"\n\n # Override specific parameters\n result = process_data(\"user_data\", cache=\"MemoryCache\", debug=True)\n print(result) # \"Processing user_data with ProductionDB, MemoryCache, debug=True\"\n```\n\n### Decorator Compatibility\n\nThe `@inject` decorator works with other Python decorators. Order matters:\n\n```python\nfrom injectipy import inject, Inject, DependencyScope\n\n# Create scope and register dependencies\nscope = DependencyScope()\nscope.register_value(\"logger\", \"ProductionLogger\")\n\nclass APIService:\n # \u2705 Recommended order: @inject comes after @classmethod/@staticmethod\n @inject\n @classmethod\n def create_from_config(cls, logger=Inject[\"logger\"]):\n return cls(logger)\n\n @inject\n @staticmethod\n def validate_data(data, logger=Inject[\"logger\"]):\n print(f\"Validating with {logger}\")\n return True\n\n# Works with other decorators too\ndef timer_decorator(func):\n def wrapper(*args, **kwargs):\n result = func(*args, **kwargs)\n return f\"timed({result})\"\n return wrapper\n\n@timer_decorator\n@inject\ndef process_data(data, logger=Inject[\"logger\"]):\n return f\"Processed {data} with {logger}\"\n\nwith scope:\n result = process_data(\"user_data\")\n print(result) # \"timed(Processed user_data with ProductionLogger)\"\n```\n\n**Decorator ordering rules:**\n- `@inject` comes after `@classmethod` or `@staticmethod`\n- `@inject` comes after other decorators (`@contextmanager`, `@lru_cache`, `@property`)\n- Apply `@inject` last (closest to the function definition)\n\n```python\n# Correct order\n@classmethod\n@inject\ndef create_service(cls, dep=Inject[\"service\"]): ...\n\n@lru_cache(maxsize=128)\n@inject\ndef cached_func(dep=Inject[\"service\"]): ...\n```\n\n### Type-Based Registration and Injection\n\nUse types directly as keys for enhanced type safety:\n\n```python\nfrom typing import Protocol\nfrom injectipy import inject, Inject, DependencyScope\n\nclass DatabaseProtocol(Protocol):\n def query(self, sql: str) -> list: ...\n\nclass PostgreSQLDatabase:\n def query(self, sql: str) -> list:\n return [\"result1\", \"result2\"]\n\nclass CacheService:\n def get(self, key: str) -> str | None:\n return f\"cached_{key}\"\n\nclass ConfigService:\n def __init__(self, env: str):\n self.env = env\n\n def get_database_url(self) -> str:\n return f\"postgresql://localhost/{self.env}\"\n\n# Register dependencies using types as keys\nscope = DependencyScope()\nscope.register_value(DatabaseProtocol, PostgreSQLDatabase())\nscope.register_value(CacheService, CacheService())\nscope.register_value(ConfigService, ConfigService(\"production\"))\n\n@inject\ndef process_user(\n user_id: int,\n db: DatabaseProtocol = Inject[DatabaseProtocol],\n cache: CacheService = Inject[CacheService],\n config: ConfigService = Inject[ConfigService]\n) -> str:\n users = db.query(\"SELECT * FROM users WHERE id = ?\")\n cached_data = cache.get(f\"user_{user_id}\")\n db_url = config.get_database_url()\n return f\"User data: {users}, cached: {cached_data}, db: {db_url}\"\n\nwith scope:\n # Full type safety - mypy knows exact types\n result = process_user(123)\n```\n\n### String-Based Registration\n\nYou can also use string keys for more flexible scenarios:\n\n```python\nfrom typing import Protocol\nfrom injectipy import inject, Inject, DependencyScope\n\nclass DatabaseProtocol(Protocol):\n def query(self, sql: str) -> list: ...\n\nclass PostgreSQLDatabase:\n def query(self, sql: str) -> list:\n return [\"result1\", \"result2\"]\n\n# Create scope and register with string keys\nscope = DependencyScope()\nscope.register_value(\"database\", PostgreSQLDatabase())\nscope.register_value(\"app_name\", \"MyApp\")\n\n@inject\ndef get_users(db: DatabaseProtocol = Inject[\"database\"], app: str = Inject[\"app_name\"]) -> list:\n print(f\"Querying from {app}\")\n return db.query(\"SELECT * FROM users\")\n\nwith scope:\n # mypy will verify types correctly\n users: list = get_users()\n```\n\n\n### Scope Isolation and Nesting\n\nYou can create multiple isolated scopes and even nest them:\n\n```python\nfrom injectipy import DependencyScope, inject, Inject\n\n# Create separate scopes for different contexts\nproduction_scope = DependencyScope()\nproduction_scope.register_value(\"config\", {\"env\": \"production\"})\nproduction_scope.register_value(\"db_url\", \"postgresql://prod-server/db\")\n\ntest_scope = DependencyScope()\ntest_scope.register_value(\"config\", {\"env\": \"test\"})\ntest_scope.register_value(\"db_url\", \"sqlite:///:memory:\")\n\n@inject\ndef get_environment(config: dict = Inject[\"config\"]):\n return config[\"env\"]\n\n# Use different scopes for different contexts\nwith production_scope:\n print(get_environment()) # \"production\"\n\nwith test_scope:\n print(get_environment()) # \"test\"\n\n# Scopes can also be nested - inner scope takes precedence\nwith production_scope:\n with test_scope:\n print(get_environment()) # \"test\" (inner scope wins)\n```\n\n## Error Handling\n\nThe library raises clear error messages for common issues:\n\n```python\nfrom injectipy import inject, ainject, Inject, DependencyScope\nfrom injectipy import (\n DependencyNotFoundError,\n CircularDependencyError,\n DuplicateRegistrationError,\n AsyncDependencyError # New!\n)\n\n# Missing dependency\n@inject\ndef missing_dep(value: str = Inject[\"nonexistent\"]):\n return value\n\ntry:\n missing_dep()\nexcept DependencyNotFoundError as e:\n print(e) # \"Dependency 'nonexistent' not found\"\n\n# Async dependency with @inject (NEW!)\nasync def async_service():\n return \"AsyncService\"\n\nscope = DependencyScope()\nscope.register_async_resolver(\"async_service\", async_service)\n\n@inject\ndef wrong_decorator(service = Inject[\"async_service\"]):\n return service\n\nwith scope:\n try:\n wrong_decorator()\n except AsyncDependencyError as e:\n print(e) # \"Cannot use @inject with async dependency 'async_service'. Use @ainject instead.\"\n\n# Circular dependency (detected at registration)\ndef service_a(b = Inject[\"service_b\"]):\n return f\"A: {b}\"\n\ndef service_b(a = Inject[\"service_a\"]):\n return f\"B: {a}\"\n\nscope = DependencyScope()\nscope.register_resolver(\"service_a\", service_a)\n\ntry:\n scope.register_resolver(\"service_b\", service_b)\nexcept CircularDependencyError as e:\n print(e) # \"Circular dependency detected\"\n\n# Duplicate registration\nscope = DependencyScope()\nscope.register_value(\"config\", \"prod\")\n\ntry:\n scope.register_value(\"config\", \"dev\")\nexcept DuplicateRegistrationError as e:\n print(e) # \"Key 'config' already registered\"\n```\n\n## Testing\n\nUse separate scopes for test isolation:\n\n```python\nimport pytest\nfrom injectipy import DependencyScope, inject, Inject\n\n@pytest.fixture\ndef test_scope():\n \"\"\"Provide a clean scope for each test\"\"\"\n return DependencyScope()\n\ndef test_dependency_injection(test_scope):\n test_scope.register_value(\"test_value\", \"hello\")\n\n @inject\n def test_function(value: str = Inject[\"test_value\"]):\n return value\n\n with test_scope:\n assert test_function() == \"hello\"\n\ndef test_isolation(test_scope):\n # Each test gets a fresh scope, so dependencies are isolated\n test_scope.register_value(\"isolated_value\", \"test_specific\")\n\n @inject\n def isolated_function(value: str = Inject[\"isolated_value\"]):\n return value\n\n with test_scope:\n assert isolated_function() == \"test_specific\"\n\ndef test_scoped_mocking(test_scope):\n # Easy to mock dependencies per test\n test_scope.register_value(\"database\", \"MockDatabase\")\n test_scope.register_value(\"cache\", \"MockCache\")\n\n @inject\n def service_function(db=Inject[\"database\"], cache=Inject[\"cache\"]):\n return f\"Using {db} and {cache}\"\n\n with test_scope:\n result = service_function()\n assert result == \"Using MockDatabase and MockCache\"\n```\n\n## Thread Safety\n\nScopes are thread-safe and can be shared between threads:\n\n```python\nimport threading\nfrom injectipy import DependencyScope, inject, Inject\n\n# Create a shared scope\nshared_scope = DependencyScope()\nshared_scope.register_value(\"shared_resource\", \"ThreadSafeResource\")\n\n@inject\ndef worker_function(resource: str = Inject[\"shared_resource\"]):\n return f\"Worker using {resource}\"\n\ndef worker():\n with shared_scope: # Each thread uses the same scope safely\n print(worker_function())\n\n# Safe to use across multiple threads\nthreads = []\nfor i in range(10):\n thread = threading.Thread(target=worker)\n threads.append(thread)\n thread.start()\n\nfor thread in threads:\n thread.join()\n```\n\n## Async/Await Support\n\nDependencyScope supports both sync and async context managers:\n\n```python\nimport asyncio\nfrom injectipy import DependencyScope, inject, Inject\n\nscope = DependencyScope()\nscope.register_value(\"api_key\", \"secret-key\")\n\n@inject\nasync def fetch_data(endpoint: str, api_key: str = Inject[\"api_key\"]) -> dict:\n # Simulate async API call\n await asyncio.sleep(0.1)\n return {\"endpoint\": endpoint, \"authenticated\": bool(api_key)}\n\nasync def main():\n async with scope: # Use async context manager\n data = await fetch_data(\"/users\")\n print(data)\n\nasyncio.run(main())\n```\n\n### Concurrent Async Tasks\n\nEach task gets proper context isolation:\n\n```python\nasync def concurrent_example():\n async def task_with_scope(task_id: int):\n task_scope = DependencyScope()\n task_scope.register_value(\"task_id\", task_id)\n\n async with task_scope:\n @inject\n async def process_task(task_id: int = Inject[\"task_id\"]) -> str:\n await asyncio.sleep(0.1)\n return f\"Processed task {task_id}\"\n\n return await process_task()\n\n # Run multiple tasks concurrently with proper isolation\n results = await asyncio.gather(\n task_with_scope(1),\n task_with_scope(2),\n task_with_scope(3)\n )\n print(results) # ['Processed task 1', 'Processed task 2', 'Processed task 3']\n\nasyncio.run(concurrent_example())\n```\n\n## API Reference\n\n### Core Components\n\n#### `@inject` decorator\nDecorates functions/methods to enable automatic dependency injection within active scopes. **Only works with sync dependencies** - rejects async dependencies with `AsyncDependencyError`.\n\n#### `@ainject` decorator\nDecorates async functions to enable automatic dependency injection with proper async/await handling. Automatically awaits async dependencies before function execution.\n\n#### `Inject[key]`\nType-safe dependency marker for function parameters.\n\n#### `DependencyScope`\nContext manager for managing dependency lifecycles and isolation.\n\n### DependencyScope Methods\n\n#### `register_value(key, value)`\nRegister a static value as a dependency. Returns self for method chaining.\n\n#### `register_resolver(key, resolver, *, evaluate_once=False)`\nRegister a sync factory function as a dependency. Returns self for method chaining.\n- `evaluate_once=True`: Cache the result after first evaluation (singleton pattern)\n\n#### `register_async_resolver(key, async_resolver, *, evaluate_once=False)`\nRegister an async factory function as a dependency. Returns self for method chaining.\n- `evaluate_once=True`: Cache the result after first evaluation (singleton pattern)\n- Use with `@ainject` decorator for clean async dependency injection\n\n#### `[key]` (getitem)\nResolve and return a dependency by key. Only works within active scope context.\n\n#### `contains(key)`\nCheck if a dependency key is registered in this scope.\n\n#### `is_active()`\nCheck if this scope is currently active (within a `with` block).\n\n#### Context Manager Protocol\n- `__enter__()`: Activate the scope\n- `__exit__()`: Deactivate the scope and clean up\n\n## Documentation\n\nFull documentation is available at [wimonder.github.io/injectipy](https://wimonder.github.io/injectipy/).\n\n## Development\n\nSee [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and contribution guidelines.\n\n### Development Setup\n\n```bash\n# Clone the repository\ngit clone https://github.com/Wimonder/injectipy.git\ncd injectipy\n\n# Install dependencies\npoetry install\n\n# Run tests\npoetry run pytest\n\n# Run type checking\npoetry run mypy injectipy/\n\n# Run linting\npoetry run ruff check .\n```\n\n### Testing\n\n```bash\n# Run all tests\npoetry run pytest\n\n# Run with coverage\npoetry run pytest --cov=injectipy\n\n# Run specific test files\npoetry run pytest tests/test_core_inject.py\npoetry run pytest tests/test_scope_functionality.py\n```\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n## Changelog\n\n### Version 0.3.0 (2025-01-03)\n- **NEW**: `@ainject` decorator for clean async dependency injection\n- **NEW**: `AsyncDependencyError` with helpful error messages guiding users to correct decorator\n- **BREAKING**: `@inject` now strictly rejects async dependencies (use `@ainject` instead)\n- **Enhanced**: Strict separation between sync and async dependency injection\n- **Performance**: Optimized async resolver detection with caching\n- **Documentation**: Updated README with comprehensive async/await examples\n\n### Version 0.1.0 (2024-01-20)\n- **Initial release** of Injectipy dependency injection library\n- **Core features**: `@inject` decorator, `Inject[key]` markers, `DependencyScope` context managers\n- **Advanced capabilities**: Thread safety, circular dependency detection, lazy evaluation\n- **Modern Python**: Python 3.11+ support with native union types\n- **Developer experience**: Type safety with mypy, comprehensive testing\n\nSee [CHANGELOG.md](CHANGELOG.md) for complete details.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A dependency injection library for Python with async/await support using explicit scopes instead of global state",
"version": "0.3.0",
"project_urls": {
"Documentation": "https://wimonder.github.io/injectipy/",
"Homepage": "https://github.com/Wimonder/injectipy",
"Repository": "https://github.com/Wimonder/injectipy"
},
"split_keywords": [
"dependency injection",
" async",
" asyncio",
" context manager",
" type safety"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "11e3c22e4db4c200bb9ab9d83c3b4f33fe670bc1c0a5c9f3ec53063cb3fe05a0",
"md5": "7641462def01b2f25091f4062fa44181",
"sha256": "6fffe878fd313688d1685ca26a6fba7435a102f195fbf846580cbaf5d5de613e"
},
"downloads": -1,
"filename": "injectipy-0.3.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7641462def01b2f25091f4062fa44181",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.11",
"size": 19814,
"upload_time": "2025-08-03T14:52:09",
"upload_time_iso_8601": "2025-08-03T14:52:09.712384Z",
"url": "https://files.pythonhosted.org/packages/11/e3/c22e4db4c200bb9ab9d83c3b4f33fe670bc1c0a5c9f3ec53063cb3fe05a0/injectipy-0.3.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "afb3cab62f69e72765e67ab8c7ffc21d47c695d9b614c4ecd514cc1a89ea3ced",
"md5": "859c91175f0dd12284a6d31c614ff7ec",
"sha256": "60b84feaae9c8b822b173002cf01304c642c240f368488d07c379d4388fe8d3a"
},
"downloads": -1,
"filename": "injectipy-0.3.0.tar.gz",
"has_sig": false,
"md5_digest": "859c91175f0dd12284a6d31c614ff7ec",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.11",
"size": 22717,
"upload_time": "2025-08-03T14:52:11",
"upload_time_iso_8601": "2025-08-03T14:52:11.053775Z",
"url": "https://files.pythonhosted.org/packages/af/b3/cab62f69e72765e67ab8c7ffc21d47c695d9b614c4ecd514cc1a89ea3ced/injectipy-0.3.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-03 14:52:11",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "Wimonder",
"github_project": "injectipy",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "injectipy"
}
JSON
