# Injx - Type-Safe Dependency Injection
[](https://python.org)
[](https://github.com/DetachHead/basedpyright)
[](https://www.apache.org/licenses/LICENSE-2.0)
[](https://qriusglobal.github.io/injx/)
> **Status: Alpha** — Ready for early adoption in SDK libraries and greenfield projects. APIs may change. Not recommended for production use without thorough testing.
## Project Status
[](https://pypi.org/project/injx/)
[](https://pypi.org/project/injx/)
[](https://github.com/QriusGlobal/injx/actions/workflows/ci.yml)
[](https://github.com/QriusGlobal/injx/actions/workflows/docs.yml)
[](https://codecov.io/gh/QriusGlobal/injx)
[](https://www.apache.org/licenses/LICENSE-2.0)
[](#)
[](https://docs.astral.sh/ruff/)
[](https://docs.astral.sh/ruff/formatter/)
Type-safe dependency injection container for Python 3.13+.
## Features
- Thread-safe and async-safe resolution using ContextVar isolation
- O(1) token lookups with pre-computed hashes
- O(1) circular dependency detection using set-based tracking
- Automatic resource cleanup in LIFO order
- Protocol-based type safety with static type checking
- Metaclass auto-registration for declarative patterns
- Zero external dependencies
- PEP 561 compliant with py.typed marker
- Memory-efficient singleton management
## Architecture
### Pure Python Implementation
Injx uses pure Python without C extensions:
- No platform-specific compilation requirements
- Standard Python debugging tools work without modification
- No segmentation faults from C extension issues
- Consistent behavior across all Python environments
### Async Handling
Explicit `get()` and `aget()` methods for synchronous and asynchronous resolution:
- Synchronous `get()` raises `AsyncCleanupRequiredError` for async providers
- Asynchronous `aget()` properly awaits async providers
- No implicit async mode switching
- Clear separation of sync and async code paths
### Token System
Strongly-typed `Token[T]` instances as container keys:
- Type information preserved at runtime
- Pre-computed hashes for O(1) lookups
- No string-based token resolution
- Tokens are immutable and hashable
### Registration Model
Explicit provider registration without auto-discovery:
- All dependencies must be explicitly registered
- No module scanning or import hooks
- No decorator-based auto-registration
- Registration happens at container initialization
## Documentation
Full docs: https://qriusglobal.github.io/injx/
## Quick Start
```bash
# Install with UV (recommended)
uv add injx
# Or with pip
pip install injx
```
### Basic Usage (Recommended Pattern)
```python
from typing import Protocol
from injx import Container, Token, Scope, inject
# Define interfaces
class Logger(Protocol):
def info(self, message: str) -> None: ...
class Database(Protocol):
def query(self, sql: str) -> list[dict[str, str]]: ...
# Implementations
class ConsoleLogger:
def info(self, message: str) -> None:
print(f"INFO: {message}")
class PostgreSQLDatabase:
def query(self, sql: str) -> list[dict[str, str]]:
# Implementation here
return [{"result": "data"}]
# Create container and tokens
container = Container()
LOGGER = Token[Logger]("logger", scope=Scope.SINGLETON)
DATABASE = Token[Database]("database", scope=Scope.SINGLETON)
# Register providers
container.register(LOGGER, ConsoleLogger)
container.register(DATABASE, PostgreSQLDatabase)
# Use with @inject decorator (recommended)
@inject
def process_users(logger: Logger, db: Database) -> None:
"""Dependencies injected automatically via type annotations."""
logger.info("Processing users")
users = db.query("SELECT * FROM users")
logger.info(f"Found {len(users)} users")
# Call without arguments - dependencies auto-resolved
process_users()
# Manual resolution also available
logger = container.get(LOGGER)
db = container.get(DATABASE)
```
### Async Support
```python
import asyncio
from typing import Protocol
from injx import Container, Token, Scope, inject
class AsyncDatabase(Protocol):
async def connect(self) -> None: ...
async def query(self, sql: str) -> list[dict[str, str]]: ...
async def aclose(self) -> None: ...
class PostgreSQLAsyncDatabase:
async def connect(self) -> None:
print("Connecting to database...")
async def query(self, sql: str) -> list[dict[str, str]]:
return [{"id": "1", "name": "Alice"}]
async def aclose(self) -> None:
print("Closing database connection...")
# Setup
container = Container()
ASYNC_DB = Token[AsyncDatabase]("async_db", scope=Scope.SINGLETON)
async def create_db() -> AsyncDatabase:
db = PostgreSQLAsyncDatabase()
await db.connect()
return db
container.register(ASYNC_DB, create_db)
# Async injection
@inject
async def process_users_async(db: AsyncDatabase) -> None:
users = await db.query("SELECT * FROM users")
print(f"Processed {len(users)} users")
# Usage
async def main() -> None:
await process_users_async()
await container.aclose() # Proper cleanup
asyncio.run(main())
```
## Type Safety & Static Analysis
injx provides full static type checking support:
### PEP 561 Compliance
injx includes a `py.typed` marker file and provides complete type information:
```bash
# Works with all type checkers
mypy your_code.py
basedpyright your_code.py
pyright your_code.py
```
### Type-Safe Registration
```python
from typing import Protocol
from injx import Container, Token, Scope
class UserService(Protocol):
def get_user(self, id: int) -> dict[str, str]: ...
class DatabaseUserService:
def get_user(self, id: int) -> dict[str, str]:
return {"id": str(id), "name": "User"}
container = Container()
USER_SERVICE = Token[UserService]("user_service", scope=Scope.SINGLETON)
# Type-safe registration - mypy/basedpyright will verify compatibility
container.register(USER_SERVICE, DatabaseUserService) # ✅ OK
# This would fail type checking:
# container.register(USER_SERVICE, str) # ❌ Type error
```
### Protocol-Based Injection
```python
from typing import Protocol, runtime_checkable
from injx import Container, Token, inject
@runtime_checkable
class EmailService(Protocol):
def send_email(self, to: str, subject: str, body: str) -> bool: ...
class SMTPEmailService:
def send_email(self, to: str, subject: str, body: str) -> bool:
print(f"Sending email to {to}: {subject}")
return True
# Registration with runtime protocol validation
container = Container()
EMAIL_SERVICE = Token[EmailService]("email", scope=Scope.SINGLETON)
container.register(EMAIL_SERVICE, SMTPEmailService)
# Type-safe injection
@inject
def send_welcome_email(email_service: EmailService, user_email: str) -> None:
"""email_service parameter is automatically injected."""
email_service.send_email(
to=user_email,
subject="Welcome!",
body="Thanks for joining us."
)
# Usage - only provide non-injected arguments
send_welcome_email(user_email="user@example.com")
```
## Injection Patterns Guide
### Plain Type Annotations with @inject
```python
from injx import inject
@inject # Uses default container
def business_logic(logger: Logger, db: Database, user_id: int) -> None:
"""Dependencies are automatically resolved based on type annotations."""
logger.info(f"Processing user {user_id}")
db.query("SELECT * FROM users WHERE id = ?", user_id)
# Call with regular parameters only
business_logic(user_id=123)
```
### Inject[T] Markers for Custom Providers
```python
from typing import Annotated
from injx import inject, Inject
@inject
def advanced_handler(
# Regular injection
logger: Logger,
# With custom provider
cache: Annotated[Cache, Inject(lambda: MockCache())],
# Regular parameter
request_id: str
) -> None:
logger.info(f"Handling request {request_id}")
cache.set("last_request", request_id)
```
### Anti-Patterns
```python
# Incorrect: Using Inject[T] as type annotation with None default
def bad_handler(logger: Inject[Logger] = None) -> None:
# Type checkers cannot infer the actual type
pass
# Incorrect: Using Inject[T] without custom provider
def confusing_handler(logger: Inject[Logger]) -> None:
# Use plain Logger annotation instead
pass
# Correct: Plain type annotation
@inject
def good_handler(logger: Logger) -> None:
logger.info("Resolved from container")
# Correct: Override in tests
@inject
def handler(logger: Logger) -> None:
logger.info("Logger injected from container")
# Test override:
container.override(LOGGER, MockLogger())
```
## Core Features
### 1. Contextual Scoping
```python
from injx import Container, ContextualContainer, Token, Scope
# Contextual container supports request/session scopes
container = Container() # Inherits from ContextualContainer
USER_TOKEN = Token[User]("current_user", scope=Scope.REQUEST)
SESSION_TOKEN = Token[Session]("session", scope=Scope.SESSION)
def get_current_user() -> User:
return User(id=123, name="Alice")
def get_session() -> Session:
return Session(id="sess_456", user_id=123)
container.register(USER_TOKEN, get_current_user)
container.register(SESSION_TOKEN, get_session)
# Request scope - each request gets isolated dependencies
with container.request_scope():
user1 = container.get(USER_TOKEN)
user2 = container.get(USER_TOKEN)
assert user1 is user2 # Same instance within request scope
with container.request_scope():
user3 = container.get(USER_TOKEN)
assert user1 is not user3 # Different instance in new scope
# Session scope - longer-lived than request
with container.session_scope():
with container.request_scope():
session = container.get(SESSION_TOKEN)
# Session persists across multiple requests
```
### 2. TokenFactory for Convenient Creation
```python
from injx import Container, TokenFactory, Scope
container = Container()
# TokenFactory provides convenient methods
factory = container.tokens # Built-in factory
# Convenient creation methods
LOGGER = factory.singleton("logger", Logger)
CACHE = factory.request("cache", CacheService)
CONFIG = factory.session("config", Configuration)
TEMP_FILE = factory.transient("temp_file", TempFile)
# With qualifiers for multiple instances
PRIMARY_DB = factory.qualified("primary", Database, Scope.SINGLETON)
SECONDARY_DB = factory.qualified("secondary", Database, Scope.SINGLETON)
# Register providers
container.register(PRIMARY_DB, lambda: PostgreSQLDatabase("primary"))
container.register(SECONDARY_DB, lambda: PostgreSQLDatabase("secondary"))
```
### 3. Default Container Support
```python
from injx import get_default_container, set_default_container, inject
# Set up global default container
default_container = Container()
set_default_container(default_container)
# Register global services
LOGGER = Token[Logger]("logger", scope=Scope.SINGLETON)
default_container.register(LOGGER, ConsoleLogger)
# @inject uses default container when none specified
@inject
def handler(logger: Logger) -> None:
logger.info("Using default container")
# Anywhere in your app
current_container = get_default_container()
```
### 4. Resource Cleanup with Context Managers
```python
import asyncio
from contextlib import asynccontextmanager, contextmanager
from typing import AsyncGenerator, Generator
from injx import Container, Token, Scope
# Sync context manager
@contextmanager
def database_connection() -> Generator[Database, None, None]:
print("Opening database connection")
db = PostgreSQLDatabase()
try:
yield db
finally:
print("Closing database connection")
db.close()
# Async context manager
@asynccontextmanager
async def async_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
print("Creating HTTP client")
client = httpx.AsyncClient()
try:
yield client
finally:
print("Closing HTTP client")
await client.aclose()
# Register context managers
container = Container()
DB_TOKEN = Token[Database]("database", scope=Scope.SINGLETON)
HTTP_TOKEN = Token[httpx.AsyncClient]("http_client", scope=Scope.SINGLETON)
container.register_context_sync(DB_TOKEN, database_connection)
container.register_context_async(HTTP_TOKEN, async_http_client)
# Automatic cleanup
async def main() -> None:
# Resources created on first access
db = container.get(DB_TOKEN)
client = await container.aget(HTTP_TOKEN)
# Proper cleanup in LIFO order
await container.aclose()
asyncio.run(main())
```
### 5. Given Instances (Scala-Style)
```python
from injx import Container, Given, inject
class UserService:
def __init__(self, db: Database):
self.db = db
# Scala-inspired given instances
container = Container()
# Register "given" providers
container.given(Database, lambda: PostgreSQLDatabase())
container.given(Logger, lambda: ConsoleLogger())
@inject
def process_request(
user_service: Given[UserService], # Resolved from given instances
request_id: str
) -> None:
# user_service automatically constructed with given Database
pass
```
## Advanced Patterns
### Async-Safe Singleton Initialization
```python
import asyncio
from injx import Container, Token, Scope
# Thread-safe async singleton creation
async def create_expensive_service() -> ExpensiveService:
print("Creating expensive service...")
await asyncio.sleep(0.1) # Simulate expensive initialization
return ExpensiveService()
container = Container()
SERVICE_TOKEN = Token[ExpensiveService]("expensive", scope=Scope.SINGLETON)
container.register(SERVICE_TOKEN, create_expensive_service)
async def concurrent_access() -> None:
# Multiple concurrent accesses - only one instance created
tasks = [container.aget(SERVICE_TOKEN) for _ in range(100)]
services = await asyncio.gather(*tasks)
# All references point to the same instance
assert all(s is services[0] for s in services)
print(f"All {len(services)} references are identical")
asyncio.run(concurrent_access())
```
### Circular Dependency Detection
```python
from injx import Container, Token, CircularDependencyError
class ServiceA:
def __init__(self, service_b: 'ServiceB'):
self.service_b = service_b
class ServiceB:
def __init__(self, service_a: ServiceA):
self.service_a = service_a
container = Container()
SERVICE_A = Token[ServiceA]("service_a")
SERVICE_B = Token[ServiceB]("service_b")
# This creates a circular dependency
container.register(SERVICE_A, lambda: ServiceA(container.get(SERVICE_B)))
container.register(SERVICE_B, lambda: ServiceB(container.get(SERVICE_A)))
try:
container.get(SERVICE_A)
except CircularDependencyError as e:
print(f"Circular dependency detected: {e}")
# Output: Cannot resolve token 'service_a':
# Resolution chain: service_a -> service_b -> service_a
# Cause: Circular dependency detected
```
### Type-Safe Override System
```python
from unittest.mock import Mock
from injx import Container, Token
# Production setup
container = Container()
EMAIL_SERVICE = Token[EmailService]("email", scope=Scope.SINGLETON)
container.register(EMAIL_SERVICE, SMTPEmailService)
# Type-safe testing with overrides
def test_email_functionality() -> None:
# Create type-safe mock
mock_email = Mock(spec=EmailService)
mock_email.send_email.return_value = True
# Override for testing - type checked!
container.override(EMAIL_SERVICE, mock_email)
@inject
def send_notification(email_service: EmailService) -> bool:
return email_service.send_email("test@example.com", "Test", "Body")
# Test uses mock
result = send_notification()
assert result is True
mock_email.send_email.assert_called_once()
# Cleanup override
container.clear_overrides()
```
## Framework Integration
### FastAPI Integration
```python
from typing import Annotated
from fastapi import FastAPI, Depends
from injx import Container, Token, Scope, inject
# Setup DI container
app = FastAPI()
container = Container()
# Register services
USER_SERVICE = Token[UserService]("user_service", scope=Scope.SINGLETON)
EMAIL_SERVICE = Token[EmailService]("email_service", scope=Scope.SINGLETON)
container.register(USER_SERVICE, lambda: DatabaseUserService())
container.register(EMAIL_SERVICE, lambda: SMTPEmailService())
# FastAPI dependency provider
def get_container() -> Container:
return container
# Option 1: FastAPI-style dependencies
@app.post("/users")
async def create_user(
user_data: UserCreateRequest,
user_service: Annotated[UserService, Depends(lambda: container.get(USER_SERVICE))],
email_service: Annotated[EmailService, Depends(lambda: container.get(EMAIL_SERVICE))]
) -> UserResponse:
user = user_service.create_user(user_data)
email_service.send_email(user.email, "Welcome!", "Welcome to our service")
return UserResponse.from_user(user)
# Option 2: injx @inject decorator (cleaner)
@app.post("/users-v2")
@inject(container=container)
async def create_user_v2(
user_data: UserCreateRequest,
user_service: UserService, # Auto-injected
email_service: EmailService # Auto-injected
) -> UserResponse:
user = user_service.create_user(user_data)
email_service.send_email(user.email, "Welcome!", "Welcome to our service")
return UserResponse.from_user(user)
# Request-scoped dependencies
@app.middleware("http")
async def setup_request_scope(request, call_next):
async with container.async_request_scope():
response = await call_next(request)
return response
# Startup/shutdown
@app.on_event("startup")
async def startup():
# Initialize resources
pass
@app.on_event("shutdown")
async def shutdown():
await container.aclose()
```
### Django Integration
```python
# settings.py
from injx import Container, Token, Scope, set_default_container
# Global container setup
DI_CONTAINER = Container()
set_default_container(DI_CONTAINER)
# Register services
USER_SERVICE = Token[UserService]("user_service", scope=Scope.SINGLETON)
EMAIL_SERVICE = Token[EmailService]("email_service", scope=Scope.SINGLETON)
DI_CONTAINER.register(USER_SERVICE, lambda: DjangoUserService())
DI_CONTAINER.register(EMAIL_SERVICE, lambda: DjangoEmailService())
# views.py
from django.http import JsonResponse
from injx import inject
@inject # Uses default container
def create_user_view(
request,
user_service: UserService, # Auto-injected
email_service: EmailService # Auto-injected
) -> JsonResponse:
if request.method == 'POST':
user_data = json.loads(request.body)
user = user_service.create_user(user_data)
email_service.send_welcome_email(user.email)
return JsonResponse({"user_id": user.id})
return JsonResponse({"error": "Method not allowed"}, status=405)
```
### CLI Applications with Click
```python
import click
from injx import Container, Token, Scope, inject
# Setup container
container = Container()
CONFIG_SERVICE = Token[ConfigService]("config", scope=Scope.SINGLETON)
LOGGER = Token[Logger]("logger", scope=Scope.SINGLETON)
container.register(CONFIG_SERVICE, lambda: FileConfigService("config.yml"))
container.register(LOGGER, lambda: ConsoleLogger())
@click.group()
@click.pass_context
def cli(ctx):
"""CLI application with dependency injection."""
ctx.obj = container
@cli.command()
@click.argument('name')
@click.pass_context
@inject # Can access container via click context
def greet(ctx, name: str, logger: Logger) -> None:
"""Greet a user with proper logging."""
logger.info(f"Greeting user: {name}")
click.echo(f"Hello, {name}!")
@cli.command()
@click.pass_context
@inject
def status(ctx, config: ConfigService, logger: Logger) -> None:
"""Show application status."""
logger.info("Checking application status")
version = config.get("version", "unknown")
click.echo(f"Application version: {version}")
if __name__ == "__main__":
cli()
```
## Testing Patterns
### Unit Testing with Dependency Overrides
```python
import pytest
from unittest.mock import Mock, MagicMock
from injx import Container, Token, Scope
class TestUserService:
def setup_method(self):
"""Setup for each test method."""
self.container = Container()
# Register production dependencies
self.db_token = Token[Database]("database", scope=Scope.SINGLETON)
self.email_token = Token[EmailService]("email", scope=Scope.SINGLETON)
self.user_service_token = Token[UserService]("user_service")
self.container.register(self.db_token, PostgreSQLDatabase)
self.container.register(self.email_token, SMTPEmailService)
self.container.register(
self.user_service_token,
lambda: UserService(
db=self.container.get(self.db_token),
email=self.container.get(self.email_token)
)
)
def test_create_user_success(self):
"""Test successful user creation with mocked dependencies."""
# Create type-safe mocks
mock_db = Mock(spec=Database)
mock_email = Mock(spec=EmailService)
mock_db.create_user.return_value = User(id=1, email="test@example.com")
mock_email.send_welcome_email.return_value = True
# Override dependencies for this test
self.container.override(self.db_token, mock_db)
self.container.override(self.email_token, mock_email)
# Get service with mocked dependencies
user_service = self.container.get(self.user_service_token)
# Test
user = user_service.create_user("test@example.com")
# Verify
assert user.id == 1
mock_db.create_user.assert_called_once_with("test@example.com")
mock_email.send_welcome_email.assert_called_once_with("test@example.com")
def teardown_method(self):
"""Cleanup after each test."""
self.container.clear_overrides()
### Async Testing
```python
import asyncio
import pytest
from unittest.mock import AsyncMock
from injx import Container, Token, Scope
@pytest.mark.asyncio
async def test_async_user_service():
"""Test async service with async mocked dependencies."""
container = Container()
# Setup tokens
async_db_token = Token[AsyncDatabase]("async_db", scope=Scope.SINGLETON)
async_email_token = Token[AsyncEmailService]("async_email", scope=Scope.SINGLETON)
# Create async mocks
mock_async_db = AsyncMock(spec=AsyncDatabase)
mock_async_email = AsyncMock(spec=AsyncEmailService)
mock_async_db.create_user.return_value = User(id=1, email="test@example.com")
mock_async_email.send_welcome_email.return_value = True
# Override with mocks
container.override(async_db_token, mock_async_db)
container.override(async_email_token, mock_async_email)
# Test with @inject decorator
@inject(container=container)
async def create_user_workflow(
email: str,
db: AsyncDatabase,
email_service: AsyncEmailService
) -> User:
user = await db.create_user(email)
await email_service.send_welcome_email(email)
return user
# Execute test
user = await create_user_workflow("test@example.com")
# Verify
assert user.id == 1
mock_async_db.create_user.assert_called_once_with("test@example.com")
mock_async_email.send_welcome_email.assert_called_once_with("test@example.com")
### Request-Scoped Testing
```python
def test_request_scoped_dependencies():
"""Test request-scoped dependency isolation."""
container = Container()
request_service_token = Token[RequestService]("request_service", scope=Scope.REQUEST)
container.register(request_service_token, lambda: RequestService())
# Request 1
with container.request_scope():
service1a = container.get(request_service_token)
service1b = container.get(request_service_token)
assert service1a is service1b # Same instance within scope
# Request 2
with container.request_scope():
service2 = container.get(request_service_token)
assert service1a is not service2 # Different instance in new scope
```
## Performance Optimizations
### Performance Characteristics
- Token lookups: O(1) with pre-computed hashes
- Cycle detection: O(1) using set-based tracking
- Memory overhead: ~500 bytes per registered service
- Singleton access: Constant time after initial creation
- Transient scope: No caching, new instance per resolution
### O(1) Token Lookups
```python
from injx import Container, Token, TokenFactory
# Tokens use pre-computed hashes for O(1) lookups
container = Container()
factory = TokenFactory()
# Create many tokens - lookups remain constant time
tokens = [
factory.singleton(f"service_{i}", type(f"Service{i}", (), {}))
for i in range(1000)
]
# Register all services
for i, token in enumerate(tokens):
container.register(token, lambda i=i: f"Service instance {i}")
# Resolution time is O(1) regardless of container size
service_500 = container.get(tokens[500]) # Same speed as tokens[0]
```
### Cached Injection Metadata
```python
from injx import inject, InjectionAnalyzer
# Function signature analysis is cached automatically
@inject # Analysis cached on first call
def expensive_handler(
service1: Service1,
service2: Service2,
service3: Service3,
regular_param: str
) -> None:
pass
# Subsequent calls use cached metadata - no re-analysis
expensive_handler(regular_param="test") # Fast
expensive_handler(regular_param="test2") # Fast
```
### Memory-Efficient Resource Management
```python
from weakref import WeakValueDictionary
from injx import Container, Token, Scope
# Transient dependencies use weak references for automatic cleanup
container = Container()
# These don't prevent garbage collection
TEMP_TOKEN = Token[TempService]("temp", scope=Scope.TRANSIENT)
container.register(TEMP_TOKEN, lambda: TempService())
temp_service = container.get(TEMP_TOKEN)
# When temp_service goes out of scope, it can be garbage collected
# Container doesn't hold strong references to transient instances
```
## Error Handling and Debugging
### Detailed Error Messages
```python
from injx import Container, Token, ResolutionError
container = Container()
SERVICE_A = Token[ServiceA]("service_a")
SERVICE_B = Token[ServiceB]("missing_service")
# Register only SERVICE_A, not SERVICE_B
container.register(SERVICE_A, lambda: ServiceA(container.get(SERVICE_B)))
try:
container.get(SERVICE_A)
except ResolutionError as e:
print(f"Resolution error: {e}")
# Output:
# Cannot resolve token 'missing_service':
# Resolution chain: service_a -> missing_service
# Cause: No provider registered for token 'missing_service'
# Access structured error data
print(f"Failed token: {e.token.name}")
print(f"Resolution chain: {[t.name for t in e.chain]}")
print(f"Root cause: {e.cause}")
```
### Debug Mode
```python
import logging
from injx import Container, Token
# Enable debug logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("injx")
container = Container()
SERVICE_TOKEN = Token[DebugService]("debug_service", scope=Scope.SINGLETON)
container.register(SERVICE_TOKEN, lambda: DebugService())
# Resolution steps are logged in debug mode
service = container.get(SERVICE_TOKEN)
```
## Migration Guides
### From dependency-injector
```python
# Before (dependency-injector)
from dependency_injector import containers, providers
class ApplicationContainer(containers.DeclarativeContainer):
config = providers.Configuration()
database = providers.Singleton(
Database,
host=config.database.host,
port=config.database.port
)
user_service = providers.Factory(
UserService,
db=database
)
# After (injx)
from injx import Container, Token, Scope
container = Container()
# Define tokens
DATABASE = Token[Database]("database", scope=Scope.SINGLETON)
USER_SERVICE = Token[UserService]("user_service", scope=Scope.TRANSIENT)
# Register providers
container.register(
DATABASE,
lambda: Database(
host=config.get("database.host"),
port=config.get("database.port")
)
)
container.register(
USER_SERVICE,
lambda: UserService(db=container.get(DATABASE))
)
```
### From injector
```python
# Before (injector)
from injector import Injector, inject, singleton
injector = Injector()
injector.binder.bind(Database, to=PostgreSQLDatabase, scope=singleton)
@inject
def user_handler(db: Database) -> None:
pass
# After (injx)
from injx import Container, Token, Scope, inject
container = Container()
DATABASE = Token[Database]("database", scope=Scope.SINGLETON)
container.register(DATABASE, PostgreSQLDatabase)
@inject(container=container) # or set as default container
def user_handler(db: Database) -> None:
pass
```
## Development Setup
```bash
# Clone repository
git clone https://github.com/qriusglobal/injx.git
cd injx
# Install with development dependencies
uv sync
# Run tests with coverage
uv run pytest --cov=injx --cov-report=html
# Type checking (strict mode)
uvx basedpyright src
# Format and lint
uvx ruff format .
uvx ruff check . --fix
# Run all quality checks
uvx ruff check . && uvx basedpyright src && uv run pytest -q
```
### Running Tests
```bash
# All tests
uv run pytest
# Specific test categories
uv run pytest tests/test_container.py # Core container tests
uv run pytest tests/test_injection.py # Injection decorator tests
uv run pytest tests/test_contextual.py # Scoping tests
uv run pytest tests/test_async.py # Async tests
uv run pytest tests/test_performance.py # Performance benchmarks
uv run pytest tests/integration/ # Integration tests
# With coverage
uv run pytest --cov=injx --cov-report=term-missing
```
## Best Practices
### 1. Token Organization
```python
# tokens.py - Centralize token definitions
from injx import TokenFactory, Token
from typing import Protocol
# Use factory for consistency
factory = TokenFactory()
# Group related tokens
class DatabaseTokens:
PRIMARY = factory.singleton("primary_db", Database)
SECONDARY = factory.singleton("secondary_db", Database)
CACHE = factory.request("cache", CacheService)
class ServiceTokens:
USER_SERVICE = factory.singleton("user_service", UserService)
EMAIL_SERVICE = factory.singleton("email_service", EmailService)
AUTH_SERVICE = factory.request("auth_service", AuthService)
# Use protocols for flexibility
class Tokens:
LOGGER = factory.singleton("logger", Logger) # Protocol
CONFIG = factory.singleton("config", Configuration) # Concrete
```
### 2. Container Lifecycle Management
```python
# app.py - Application lifecycle
import asyncio
from contextlib import asynccontextmanager
from injx import Container, set_default_container
@asynccontextmanager
async def lifespan():
"""Manage container lifecycle."""
# Startup
container = Container()
await setup_dependencies(container)
set_default_container(container)
try:
yield container
finally:
# Shutdown - cleanup resources
await container.aclose()
async def setup_dependencies(container: Container) -> None:
"""Register all application dependencies."""
# Register database connections
container.register(DatabaseTokens.PRIMARY, create_primary_db)
container.register(DatabaseTokens.SECONDARY, create_secondary_db)
# Register services
container.register(ServiceTokens.USER_SERVICE, create_user_service)
container.register(ServiceTokens.EMAIL_SERVICE, create_email_service)
async def main():
async with lifespan() as container:
# Application code here
await run_application()
if __name__ == "__main__":
asyncio.run(main())
```
### 3. Testing Strategy
```python
# test_base.py - Shared testing infrastructure
import pytest
from injx import Container
from unittest.mock import Mock
class DITestCase:
"""Base class for DI-enabled tests."""
def setup_method(self):
self.container = Container()
self.mocks = {}
def mock_service(self, token: Token[T], **kwargs) -> Mock:
"""Create and register a type-safe mock."""
mock = Mock(spec=token.type_, **kwargs)
self.container.override(token, mock)
self.mocks[token.name] = mock
return mock
def teardown_method(self):
self.container.clear_overrides()
# test_user_service.py - Concrete test
class TestUserService(DITestCase):
def test_create_user(self):
# Setup mocks
mock_db = self.mock_service(DatabaseTokens.PRIMARY)
mock_email = self.mock_service(ServiceTokens.EMAIL_SERVICE)
mock_db.create_user.return_value = User(id=1, email="test@example.com")
mock_email.send_welcome_email.return_value = True
# Test
user_service = self.container.get(ServiceTokens.USER_SERVICE)
user = user_service.create_user("test@example.com")
# Verify
assert user.id == 1
mock_db.create_user.assert_called_once()
mock_email.send_welcome_email.assert_called_once()
```
## Troubleshooting
### Common Issues
**1. Circular Dependencies**
```python
# Problem: Services depend on each other
# Solution: Use lazy injection or redesign
# Instead of:
class ServiceA:
def __init__(self, service_b: ServiceB): ...
class ServiceB:
def __init__(self, service_a: ServiceA): ...
# Do:
class ServiceA:
def __init__(self, get_service_b: Callable[[], ServiceB]): ...
# Or redesign to avoid circular dependency
```
**2. Type Checker Issues**
```python
# Problem: mypy/basedpyright can't infer types
# Solution: Use explicit type annotations
# Instead of:
@inject
def handler(service): # Type unknown
pass
# Do:
@inject
def handler(service: UserService) -> None: # Explicit types
pass
```
**3. Async/Sync Mixing**
```python
# Problem: Using async resources in sync context
# Solution: Use appropriate container methods
# Instead of:
async def create_service():
return AsyncService()
container.register(TOKEN, create_service)
service = container.get(TOKEN) # ❌ Error: async provider in sync context
# Do:
service = await container.aget(TOKEN) # ✅ Correct async usage
```
**4. Resource Leaks**
```python
# Problem: Resources not properly cleaned up
# Solution: Use context managers or proper cleanup
# Instead of:
def create_connection():
return DatabaseConnection() # May leak
# Do:
@contextmanager
def database_connection():
conn = DatabaseConnection()
try:
yield conn
finally:
conn.close()
container.register_context_sync(DB_TOKEN, database_connection)
```
## Contributing
1. Fork the repository
2. Create a feature branch (`git checkout -b feature/amazing-feature`)
3. Write tests for your changes
4. Ensure all quality checks pass (`uvx ruff check . && uvx basedpyright src && uv run pytest`)
5. Submit a pull request
### Code Standards
- **Type safety**: All code must pass `basedpyright --strict`
- **Testing**: Maintain 90%+ test coverage
- **Documentation**: Update README for user-facing changes
- **Formatting**: Use `ruff format` (88 character lines)
- **Performance**: Maintain O(1) lookup guarantees
## License
injx is licensed under the **Apache License 2.0**.
This is a permissive open source license that allows you to use injx in both open source and proprietary projects. The Apache 2.0 license provides:
- **Freedom to use commercially**: Use injx in your commercial products without restrictions
- **Patent protection**: Explicit patent grant protects you from patent claims
- **Simple attribution**: Just include the license and copyright notice
- **Compatible with most licenses**: Works well with MIT, BSD, and other permissive licenses
See [LICENSE](LICENSE) for the full license text.
Copyright 2025 Qrius Global - Licensed under Apache License 2.0
Raw data
{
"_id": null,
"home_page": null,
"name": "injx",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.13",
"maintainer_email": null,
"keywords": "async, dependency-injection, di, ioc, python3.13, type-safe",
"author": null,
"author_email": "Qrius Global <mishal@qrius.global>",
"download_url": "https://files.pythonhosted.org/packages/fe/42/1dc4f9c75ae8af7026302be085ec9f54aeae1ae759527a24be445fa44230/injx-0.1.0.tar.gz",
"platform": null,
"description": "# Injx - Type-Safe Dependency Injection\n\n[](https://python.org)\n[](https://github.com/DetachHead/basedpyright)\n[](https://www.apache.org/licenses/LICENSE-2.0)\n[](https://qriusglobal.github.io/injx/)\n\n> **Status: Alpha** \u2014 Ready for early adoption in SDK libraries and greenfield projects. APIs may change. Not recommended for production use without thorough testing.\n\n## Project Status\n\n[](https://pypi.org/project/injx/)\n[](https://pypi.org/project/injx/)\n[](https://github.com/QriusGlobal/injx/actions/workflows/ci.yml)\n[](https://github.com/QriusGlobal/injx/actions/workflows/docs.yml)\n[](https://codecov.io/gh/QriusGlobal/injx)\n[](https://www.apache.org/licenses/LICENSE-2.0)\n[](#)\n[](https://docs.astral.sh/ruff/)\n[](https://docs.astral.sh/ruff/formatter/)\n\nType-safe dependency injection container for Python 3.13+.\n\n## Features\n\n- Thread-safe and async-safe resolution using ContextVar isolation\n- O(1) token lookups with pre-computed hashes\n- O(1) circular dependency detection using set-based tracking\n- Automatic resource cleanup in LIFO order\n- Protocol-based type safety with static type checking\n- Metaclass auto-registration for declarative patterns\n- Zero external dependencies\n- PEP 561 compliant with py.typed marker\n- Memory-efficient singleton management\n\n## Architecture\n\n### Pure Python Implementation\nInjx uses pure Python without C extensions:\n- No platform-specific compilation requirements\n- Standard Python debugging tools work without modification\n- No segmentation faults from C extension issues\n- Consistent behavior across all Python environments\n\n### Async Handling\nExplicit `get()` and `aget()` methods for synchronous and asynchronous resolution:\n- Synchronous `get()` raises `AsyncCleanupRequiredError` for async providers\n- Asynchronous `aget()` properly awaits async providers\n- No implicit async mode switching\n- Clear separation of sync and async code paths\n\n### Token System\nStrongly-typed `Token[T]` instances as container keys:\n- Type information preserved at runtime\n- Pre-computed hashes for O(1) lookups\n- No string-based token resolution\n- Tokens are immutable and hashable\n\n### Registration Model\nExplicit provider registration without auto-discovery:\n- All dependencies must be explicitly registered\n- No module scanning or import hooks\n- No decorator-based auto-registration\n- Registration happens at container initialization\n\n## Documentation\n\nFull docs: https://qriusglobal.github.io/injx/\n\n## Quick Start\n\n```bash\n# Install with UV (recommended)\nuv add injx\n\n# Or with pip\npip install injx\n```\n\n### Basic Usage (Recommended Pattern)\n\n```python\nfrom typing import Protocol\nfrom injx import Container, Token, Scope, inject\n\n# Define interfaces\nclass Logger(Protocol):\n def info(self, message: str) -> None: ...\n\nclass Database(Protocol):\n def query(self, sql: str) -> list[dict[str, str]]: ...\n\n# Implementations\nclass ConsoleLogger:\n def info(self, message: str) -> None:\n print(f\"INFO: {message}\")\n\nclass PostgreSQLDatabase:\n def query(self, sql: str) -> list[dict[str, str]]:\n # Implementation here\n return [{\"result\": \"data\"}]\n\n# Create container and tokens\ncontainer = Container()\nLOGGER = Token[Logger](\"logger\", scope=Scope.SINGLETON)\nDATABASE = Token[Database](\"database\", scope=Scope.SINGLETON)\n\n# Register providers\ncontainer.register(LOGGER, ConsoleLogger)\ncontainer.register(DATABASE, PostgreSQLDatabase)\n\n# Use with @inject decorator (recommended)\n@inject\ndef process_users(logger: Logger, db: Database) -> None:\n \"\"\"Dependencies injected automatically via type annotations.\"\"\"\n logger.info(\"Processing users\")\n users = db.query(\"SELECT * FROM users\")\n logger.info(f\"Found {len(users)} users\")\n\n# Call without arguments - dependencies auto-resolved\nprocess_users()\n\n# Manual resolution also available\nlogger = container.get(LOGGER)\ndb = container.get(DATABASE)\n```\n\n### Async Support\n\n```python\nimport asyncio\nfrom typing import Protocol\nfrom injx import Container, Token, Scope, inject\n\nclass AsyncDatabase(Protocol):\n async def connect(self) -> None: ...\n async def query(self, sql: str) -> list[dict[str, str]]: ...\n async def aclose(self) -> None: ...\n\nclass PostgreSQLAsyncDatabase:\n async def connect(self) -> None:\n print(\"Connecting to database...\")\n \n async def query(self, sql: str) -> list[dict[str, str]]:\n return [{\"id\": \"1\", \"name\": \"Alice\"}]\n \n async def aclose(self) -> None:\n print(\"Closing database connection...\")\n\n# Setup\ncontainer = Container()\nASYNC_DB = Token[AsyncDatabase](\"async_db\", scope=Scope.SINGLETON)\n\nasync def create_db() -> AsyncDatabase:\n db = PostgreSQLAsyncDatabase()\n await db.connect()\n return db\n\ncontainer.register(ASYNC_DB, create_db)\n\n# Async injection\n@inject\nasync def process_users_async(db: AsyncDatabase) -> None:\n users = await db.query(\"SELECT * FROM users\")\n print(f\"Processed {len(users)} users\")\n\n# Usage\nasync def main() -> None:\n await process_users_async()\n await container.aclose() # Proper cleanup\n\nasyncio.run(main())\n```\n\n## Type Safety & Static Analysis\n\ninjx provides full static type checking support:\n\n### PEP 561 Compliance\n\ninjx includes a `py.typed` marker file and provides complete type information:\n\n```bash\n# Works with all type checkers\nmypy your_code.py\nbasedpyright your_code.py\npyright your_code.py\n```\n\n### Type-Safe Registration\n\n```python\nfrom typing import Protocol\nfrom injx import Container, Token, Scope\n\nclass UserService(Protocol):\n def get_user(self, id: int) -> dict[str, str]: ...\n\nclass DatabaseUserService:\n def get_user(self, id: int) -> dict[str, str]:\n return {\"id\": str(id), \"name\": \"User\"}\n\ncontainer = Container()\nUSER_SERVICE = Token[UserService](\"user_service\", scope=Scope.SINGLETON)\n\n# Type-safe registration - mypy/basedpyright will verify compatibility\ncontainer.register(USER_SERVICE, DatabaseUserService) # \u2705 OK\n\n# This would fail type checking:\n# container.register(USER_SERVICE, str) # \u274c Type error\n```\n\n### Protocol-Based Injection\n\n```python\nfrom typing import Protocol, runtime_checkable\nfrom injx import Container, Token, inject\n\n@runtime_checkable\nclass EmailService(Protocol):\n def send_email(self, to: str, subject: str, body: str) -> bool: ...\n\nclass SMTPEmailService:\n def send_email(self, to: str, subject: str, body: str) -> bool:\n print(f\"Sending email to {to}: {subject}\")\n return True\n\n# Registration with runtime protocol validation\ncontainer = Container()\nEMAIL_SERVICE = Token[EmailService](\"email\", scope=Scope.SINGLETON)\ncontainer.register(EMAIL_SERVICE, SMTPEmailService)\n\n# Type-safe injection\n@inject\ndef send_welcome_email(email_service: EmailService, user_email: str) -> None:\n \"\"\"email_service parameter is automatically injected.\"\"\"\n email_service.send_email(\n to=user_email,\n subject=\"Welcome!\",\n body=\"Thanks for joining us.\"\n )\n\n# Usage - only provide non-injected arguments\nsend_welcome_email(user_email=\"user@example.com\")\n```\n\n## Injection Patterns Guide\n\n### Plain Type Annotations with @inject\n\n```python\nfrom injx import inject\n\n@inject # Uses default container\ndef business_logic(logger: Logger, db: Database, user_id: int) -> None:\n \"\"\"Dependencies are automatically resolved based on type annotations.\"\"\"\n logger.info(f\"Processing user {user_id}\")\n db.query(\"SELECT * FROM users WHERE id = ?\", user_id)\n\n# Call with regular parameters only\nbusiness_logic(user_id=123)\n```\n\n### Inject[T] Markers for Custom Providers\n\n```python\nfrom typing import Annotated\nfrom injx import inject, Inject\n\n@inject\ndef advanced_handler(\n # Regular injection\n logger: Logger,\n \n # With custom provider\n cache: Annotated[Cache, Inject(lambda: MockCache())],\n \n # Regular parameter\n request_id: str\n) -> None:\n logger.info(f\"Handling request {request_id}\")\n cache.set(\"last_request\", request_id)\n```\n\n### Anti-Patterns\n\n```python\n# Incorrect: Using Inject[T] as type annotation with None default\ndef bad_handler(logger: Inject[Logger] = None) -> None:\n # Type checkers cannot infer the actual type\n pass\n\n# Incorrect: Using Inject[T] without custom provider\ndef confusing_handler(logger: Inject[Logger]) -> None:\n # Use plain Logger annotation instead\n pass\n\n# Correct: Plain type annotation\n@inject\ndef good_handler(logger: Logger) -> None:\n logger.info(\"Resolved from container\")\n\n# Correct: Override in tests\n@inject \ndef handler(logger: Logger) -> None:\n logger.info(\"Logger injected from container\")\n\n# Test override:\ncontainer.override(LOGGER, MockLogger())\n```\n\n## Core Features\n\n### 1. Contextual Scoping\n\n```python\nfrom injx import Container, ContextualContainer, Token, Scope\n\n# Contextual container supports request/session scopes\ncontainer = Container() # Inherits from ContextualContainer\n\nUSER_TOKEN = Token[User](\"current_user\", scope=Scope.REQUEST)\nSESSION_TOKEN = Token[Session](\"session\", scope=Scope.SESSION)\n\ndef get_current_user() -> User:\n return User(id=123, name=\"Alice\")\n\ndef get_session() -> Session:\n return Session(id=\"sess_456\", user_id=123)\n\ncontainer.register(USER_TOKEN, get_current_user)\ncontainer.register(SESSION_TOKEN, get_session)\n\n# Request scope - each request gets isolated dependencies\nwith container.request_scope():\n user1 = container.get(USER_TOKEN)\n user2 = container.get(USER_TOKEN)\n assert user1 is user2 # Same instance within request scope\n\nwith container.request_scope():\n user3 = container.get(USER_TOKEN)\n assert user1 is not user3 # Different instance in new scope\n\n# Session scope - longer-lived than request\nwith container.session_scope():\n with container.request_scope():\n session = container.get(SESSION_TOKEN)\n # Session persists across multiple requests\n```\n\n### 2. TokenFactory for Convenient Creation\n\n```python\nfrom injx import Container, TokenFactory, Scope\n\ncontainer = Container()\n# TokenFactory provides convenient methods\nfactory = container.tokens # Built-in factory\n\n# Convenient creation methods\nLOGGER = factory.singleton(\"logger\", Logger)\nCACHE = factory.request(\"cache\", CacheService) \nCONFIG = factory.session(\"config\", Configuration)\nTEMP_FILE = factory.transient(\"temp_file\", TempFile)\n\n# With qualifiers for multiple instances\nPRIMARY_DB = factory.qualified(\"primary\", Database, Scope.SINGLETON)\nSECONDARY_DB = factory.qualified(\"secondary\", Database, Scope.SINGLETON)\n\n# Register providers\ncontainer.register(PRIMARY_DB, lambda: PostgreSQLDatabase(\"primary\"))\ncontainer.register(SECONDARY_DB, lambda: PostgreSQLDatabase(\"secondary\"))\n```\n\n### 3. Default Container Support\n\n```python\nfrom injx import get_default_container, set_default_container, inject\n\n# Set up global default container\ndefault_container = Container()\nset_default_container(default_container)\n\n# Register global services\nLOGGER = Token[Logger](\"logger\", scope=Scope.SINGLETON)\ndefault_container.register(LOGGER, ConsoleLogger)\n\n# @inject uses default container when none specified\n@inject\ndef handler(logger: Logger) -> None:\n logger.info(\"Using default container\")\n\n# Anywhere in your app\ncurrent_container = get_default_container()\n```\n\n### 4. Resource Cleanup with Context Managers\n\n```python\nimport asyncio\nfrom contextlib import asynccontextmanager, contextmanager\nfrom typing import AsyncGenerator, Generator\nfrom injx import Container, Token, Scope\n\n# Sync context manager\n@contextmanager \ndef database_connection() -> Generator[Database, None, None]:\n print(\"Opening database connection\")\n db = PostgreSQLDatabase()\n try:\n yield db\n finally:\n print(\"Closing database connection\")\n db.close()\n\n# Async context manager\n@asynccontextmanager\nasync def async_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:\n print(\"Creating HTTP client\")\n client = httpx.AsyncClient()\n try:\n yield client\n finally:\n print(\"Closing HTTP client\")\n await client.aclose()\n\n# Register context managers\ncontainer = Container()\nDB_TOKEN = Token[Database](\"database\", scope=Scope.SINGLETON)\nHTTP_TOKEN = Token[httpx.AsyncClient](\"http_client\", scope=Scope.SINGLETON)\n\ncontainer.register_context_sync(DB_TOKEN, database_connection)\ncontainer.register_context_async(HTTP_TOKEN, async_http_client)\n\n# Automatic cleanup\nasync def main() -> None:\n # Resources created on first access\n db = container.get(DB_TOKEN)\n client = await container.aget(HTTP_TOKEN)\n \n # Proper cleanup in LIFO order\n await container.aclose()\n\nasyncio.run(main())\n```\n\n### 5. Given Instances (Scala-Style)\n\n```python\nfrom injx import Container, Given, inject\n\nclass UserService:\n def __init__(self, db: Database):\n self.db = db\n\n# Scala-inspired given instances\ncontainer = Container()\n\n# Register \"given\" providers\ncontainer.given(Database, lambda: PostgreSQLDatabase())\ncontainer.given(Logger, lambda: ConsoleLogger())\n\n@inject\ndef process_request(\n user_service: Given[UserService], # Resolved from given instances\n request_id: str\n) -> None:\n # user_service automatically constructed with given Database\n pass\n```\n\n## Advanced Patterns\n\n### Async-Safe Singleton Initialization\n\n```python\nimport asyncio\nfrom injx import Container, Token, Scope\n\n# Thread-safe async singleton creation\nasync def create_expensive_service() -> ExpensiveService:\n print(\"Creating expensive service...\")\n await asyncio.sleep(0.1) # Simulate expensive initialization\n return ExpensiveService()\n\ncontainer = Container()\nSERVICE_TOKEN = Token[ExpensiveService](\"expensive\", scope=Scope.SINGLETON)\ncontainer.register(SERVICE_TOKEN, create_expensive_service)\n\nasync def concurrent_access() -> None:\n # Multiple concurrent accesses - only one instance created\n tasks = [container.aget(SERVICE_TOKEN) for _ in range(100)]\n services = await asyncio.gather(*tasks)\n \n # All references point to the same instance\n assert all(s is services[0] for s in services)\n print(f\"All {len(services)} references are identical\")\n\nasyncio.run(concurrent_access())\n```\n\n### Circular Dependency Detection\n\n```python\nfrom injx import Container, Token, CircularDependencyError\n\nclass ServiceA:\n def __init__(self, service_b: 'ServiceB'):\n self.service_b = service_b\n\nclass ServiceB:\n def __init__(self, service_a: ServiceA):\n self.service_a = service_a\n\ncontainer = Container()\nSERVICE_A = Token[ServiceA](\"service_a\")\nSERVICE_B = Token[ServiceB](\"service_b\")\n\n# This creates a circular dependency\ncontainer.register(SERVICE_A, lambda: ServiceA(container.get(SERVICE_B)))\ncontainer.register(SERVICE_B, lambda: ServiceB(container.get(SERVICE_A)))\n\ntry:\n container.get(SERVICE_A)\nexcept CircularDependencyError as e:\n print(f\"Circular dependency detected: {e}\")\n # Output: Cannot resolve token 'service_a':\n # Resolution chain: service_a -> service_b -> service_a\n # Cause: Circular dependency detected\n```\n\n### Type-Safe Override System\n\n```python\nfrom unittest.mock import Mock\nfrom injx import Container, Token\n\n# Production setup\ncontainer = Container()\nEMAIL_SERVICE = Token[EmailService](\"email\", scope=Scope.SINGLETON)\ncontainer.register(EMAIL_SERVICE, SMTPEmailService)\n\n# Type-safe testing with overrides\ndef test_email_functionality() -> None:\n # Create type-safe mock\n mock_email = Mock(spec=EmailService)\n mock_email.send_email.return_value = True\n \n # Override for testing - type checked!\n container.override(EMAIL_SERVICE, mock_email)\n \n @inject\n def send_notification(email_service: EmailService) -> bool:\n return email_service.send_email(\"test@example.com\", \"Test\", \"Body\")\n \n # Test uses mock\n result = send_notification()\n assert result is True\n mock_email.send_email.assert_called_once()\n \n # Cleanup override\n container.clear_overrides()\n```\n\n## Framework Integration\n\n### FastAPI Integration\n\n```python\nfrom typing import Annotated\nfrom fastapi import FastAPI, Depends\nfrom injx import Container, Token, Scope, inject\n\n# Setup DI container\napp = FastAPI()\ncontainer = Container()\n\n# Register services\nUSER_SERVICE = Token[UserService](\"user_service\", scope=Scope.SINGLETON)\nEMAIL_SERVICE = Token[EmailService](\"email_service\", scope=Scope.SINGLETON)\n\ncontainer.register(USER_SERVICE, lambda: DatabaseUserService())\ncontainer.register(EMAIL_SERVICE, lambda: SMTPEmailService())\n\n# FastAPI dependency provider\ndef get_container() -> Container:\n return container\n\n# Option 1: FastAPI-style dependencies\n@app.post(\"/users\")\nasync def create_user(\n user_data: UserCreateRequest,\n user_service: Annotated[UserService, Depends(lambda: container.get(USER_SERVICE))],\n email_service: Annotated[EmailService, Depends(lambda: container.get(EMAIL_SERVICE))]\n) -> UserResponse:\n user = user_service.create_user(user_data)\n email_service.send_email(user.email, \"Welcome!\", \"Welcome to our service\")\n return UserResponse.from_user(user)\n\n# Option 2: injx @inject decorator (cleaner)\n@app.post(\"/users-v2\")\n@inject(container=container)\nasync def create_user_v2(\n user_data: UserCreateRequest,\n user_service: UserService, # Auto-injected\n email_service: EmailService # Auto-injected\n) -> UserResponse:\n user = user_service.create_user(user_data)\n email_service.send_email(user.email, \"Welcome!\", \"Welcome to our service\")\n return UserResponse.from_user(user)\n\n# Request-scoped dependencies\n@app.middleware(\"http\")\nasync def setup_request_scope(request, call_next):\n async with container.async_request_scope():\n response = await call_next(request)\n return response\n\n# Startup/shutdown\n@app.on_event(\"startup\")\nasync def startup():\n # Initialize resources\n pass\n\n@app.on_event(\"shutdown\") \nasync def shutdown():\n await container.aclose()\n```\n\n### Django Integration\n\n```python\n# settings.py\nfrom injx import Container, Token, Scope, set_default_container\n\n# Global container setup\nDI_CONTAINER = Container()\nset_default_container(DI_CONTAINER)\n\n# Register services\nUSER_SERVICE = Token[UserService](\"user_service\", scope=Scope.SINGLETON)\nEMAIL_SERVICE = Token[EmailService](\"email_service\", scope=Scope.SINGLETON)\n\nDI_CONTAINER.register(USER_SERVICE, lambda: DjangoUserService())\nDI_CONTAINER.register(EMAIL_SERVICE, lambda: DjangoEmailService())\n\n# views.py\nfrom django.http import JsonResponse\nfrom injx import inject\n\n@inject # Uses default container\ndef create_user_view(\n request,\n user_service: UserService, # Auto-injected\n email_service: EmailService # Auto-injected\n) -> JsonResponse:\n if request.method == 'POST':\n user_data = json.loads(request.body)\n user = user_service.create_user(user_data)\n email_service.send_welcome_email(user.email)\n return JsonResponse({\"user_id\": user.id})\n \n return JsonResponse({\"error\": \"Method not allowed\"}, status=405)\n```\n\n### CLI Applications with Click\n\n```python\nimport click\nfrom injx import Container, Token, Scope, inject\n\n# Setup container\ncontainer = Container()\nCONFIG_SERVICE = Token[ConfigService](\"config\", scope=Scope.SINGLETON)\nLOGGER = Token[Logger](\"logger\", scope=Scope.SINGLETON)\n\ncontainer.register(CONFIG_SERVICE, lambda: FileConfigService(\"config.yml\"))\ncontainer.register(LOGGER, lambda: ConsoleLogger())\n\n@click.group()\n@click.pass_context\ndef cli(ctx):\n \"\"\"CLI application with dependency injection.\"\"\"\n ctx.obj = container\n\n@cli.command()\n@click.argument('name')\n@click.pass_context\n@inject # Can access container via click context\ndef greet(ctx, name: str, logger: Logger) -> None:\n \"\"\"Greet a user with proper logging.\"\"\"\n logger.info(f\"Greeting user: {name}\")\n click.echo(f\"Hello, {name}!\")\n\n@cli.command()\n@click.pass_context\n@inject\ndef status(ctx, config: ConfigService, logger: Logger) -> None:\n \"\"\"Show application status.\"\"\"\n logger.info(\"Checking application status\")\n version = config.get(\"version\", \"unknown\")\n click.echo(f\"Application version: {version}\")\n\nif __name__ == \"__main__\":\n cli()\n```\n\n## Testing Patterns\n\n### Unit Testing with Dependency Overrides\n\n```python\nimport pytest\nfrom unittest.mock import Mock, MagicMock\nfrom injx import Container, Token, Scope\n\nclass TestUserService:\n def setup_method(self):\n \"\"\"Setup for each test method.\"\"\"\n self.container = Container()\n \n # Register production dependencies\n self.db_token = Token[Database](\"database\", scope=Scope.SINGLETON)\n self.email_token = Token[EmailService](\"email\", scope=Scope.SINGLETON)\n self.user_service_token = Token[UserService](\"user_service\")\n \n self.container.register(self.db_token, PostgreSQLDatabase)\n self.container.register(self.email_token, SMTPEmailService)\n self.container.register(\n self.user_service_token,\n lambda: UserService(\n db=self.container.get(self.db_token),\n email=self.container.get(self.email_token)\n )\n )\n \n def test_create_user_success(self):\n \"\"\"Test successful user creation with mocked dependencies.\"\"\"\n # Create type-safe mocks\n mock_db = Mock(spec=Database)\n mock_email = Mock(spec=EmailService)\n \n mock_db.create_user.return_value = User(id=1, email=\"test@example.com\")\n mock_email.send_welcome_email.return_value = True\n \n # Override dependencies for this test\n self.container.override(self.db_token, mock_db)\n self.container.override(self.email_token, mock_email)\n \n # Get service with mocked dependencies\n user_service = self.container.get(self.user_service_token)\n \n # Test\n user = user_service.create_user(\"test@example.com\")\n \n # Verify\n assert user.id == 1\n mock_db.create_user.assert_called_once_with(\"test@example.com\")\n mock_email.send_welcome_email.assert_called_once_with(\"test@example.com\")\n \n def teardown_method(self):\n \"\"\"Cleanup after each test.\"\"\"\n self.container.clear_overrides()\n\n### Async Testing\n\n```python\nimport asyncio\nimport pytest\nfrom unittest.mock import AsyncMock\nfrom injx import Container, Token, Scope\n\n@pytest.mark.asyncio\nasync def test_async_user_service():\n \"\"\"Test async service with async mocked dependencies.\"\"\"\n container = Container()\n \n # Setup tokens\n async_db_token = Token[AsyncDatabase](\"async_db\", scope=Scope.SINGLETON)\n async_email_token = Token[AsyncEmailService](\"async_email\", scope=Scope.SINGLETON)\n \n # Create async mocks\n mock_async_db = AsyncMock(spec=AsyncDatabase)\n mock_async_email = AsyncMock(spec=AsyncEmailService)\n \n mock_async_db.create_user.return_value = User(id=1, email=\"test@example.com\")\n mock_async_email.send_welcome_email.return_value = True\n \n # Override with mocks\n container.override(async_db_token, mock_async_db)\n container.override(async_email_token, mock_async_email)\n \n # Test with @inject decorator\n @inject(container=container)\n async def create_user_workflow(\n email: str,\n db: AsyncDatabase,\n email_service: AsyncEmailService\n ) -> User:\n user = await db.create_user(email)\n await email_service.send_welcome_email(email)\n return user\n \n # Execute test\n user = await create_user_workflow(\"test@example.com\")\n \n # Verify\n assert user.id == 1\n mock_async_db.create_user.assert_called_once_with(\"test@example.com\")\n mock_async_email.send_welcome_email.assert_called_once_with(\"test@example.com\")\n\n### Request-Scoped Testing\n\n```python\ndef test_request_scoped_dependencies():\n \"\"\"Test request-scoped dependency isolation.\"\"\"\n container = Container()\n request_service_token = Token[RequestService](\"request_service\", scope=Scope.REQUEST)\n \n container.register(request_service_token, lambda: RequestService())\n \n # Request 1\n with container.request_scope():\n service1a = container.get(request_service_token) \n service1b = container.get(request_service_token)\n assert service1a is service1b # Same instance within scope\n \n # Request 2\n with container.request_scope():\n service2 = container.get(request_service_token)\n assert service1a is not service2 # Different instance in new scope\n```\n\n## Performance Optimizations\n\n### Performance Characteristics\n\n- Token lookups: O(1) with pre-computed hashes\n- Cycle detection: O(1) using set-based tracking \n- Memory overhead: ~500 bytes per registered service\n- Singleton access: Constant time after initial creation\n- Transient scope: No caching, new instance per resolution\n\n### O(1) Token Lookups\n\n```python\nfrom injx import Container, Token, TokenFactory\n\n# Tokens use pre-computed hashes for O(1) lookups\ncontainer = Container()\nfactory = TokenFactory()\n\n# Create many tokens - lookups remain constant time\ntokens = [\n factory.singleton(f\"service_{i}\", type(f\"Service{i}\", (), {}))\n for i in range(1000)\n]\n\n# Register all services\nfor i, token in enumerate(tokens):\n container.register(token, lambda i=i: f\"Service instance {i}\")\n\n# Resolution time is O(1) regardless of container size\nservice_500 = container.get(tokens[500]) # Same speed as tokens[0]\n```\n\n### Cached Injection Metadata\n\n```python\nfrom injx import inject, InjectionAnalyzer\n\n# Function signature analysis is cached automatically\n@inject # Analysis cached on first call\ndef expensive_handler(\n service1: Service1,\n service2: Service2,\n service3: Service3,\n regular_param: str\n) -> None:\n pass\n\n# Subsequent calls use cached metadata - no re-analysis\nexpensive_handler(regular_param=\"test\") # Fast\nexpensive_handler(regular_param=\"test2\") # Fast\n```\n\n### Memory-Efficient Resource Management\n\n```python\nfrom weakref import WeakValueDictionary\nfrom injx import Container, Token, Scope\n\n# Transient dependencies use weak references for automatic cleanup\ncontainer = Container()\n\n# These don't prevent garbage collection\nTEMP_TOKEN = Token[TempService](\"temp\", scope=Scope.TRANSIENT)\ncontainer.register(TEMP_TOKEN, lambda: TempService())\n\ntemp_service = container.get(TEMP_TOKEN)\n# When temp_service goes out of scope, it can be garbage collected\n# Container doesn't hold strong references to transient instances\n```\n\n## Error Handling and Debugging\n\n### Detailed Error Messages\n\n```python\nfrom injx import Container, Token, ResolutionError\n\ncontainer = Container()\nSERVICE_A = Token[ServiceA](\"service_a\")\nSERVICE_B = Token[ServiceB](\"missing_service\")\n\n# Register only SERVICE_A, not SERVICE_B\ncontainer.register(SERVICE_A, lambda: ServiceA(container.get(SERVICE_B)))\n\ntry:\n container.get(SERVICE_A)\nexcept ResolutionError as e:\n print(f\"Resolution error: {e}\")\n # Output:\n # Cannot resolve token 'missing_service':\n # Resolution chain: service_a -> missing_service\n # Cause: No provider registered for token 'missing_service'\n \n # Access structured error data\n print(f\"Failed token: {e.token.name}\")\n print(f\"Resolution chain: {[t.name for t in e.chain]}\")\n print(f\"Root cause: {e.cause}\")\n```\n\n### Debug Mode\n\n```python\nimport logging\nfrom injx import Container, Token\n\n# Enable debug logging\nlogging.basicConfig(level=logging.DEBUG)\nlogger = logging.getLogger(\"injx\")\n\ncontainer = Container()\nSERVICE_TOKEN = Token[DebugService](\"debug_service\", scope=Scope.SINGLETON)\n\ncontainer.register(SERVICE_TOKEN, lambda: DebugService())\n\n# Resolution steps are logged in debug mode\nservice = container.get(SERVICE_TOKEN)\n```\n\n## Migration Guides\n\n### From dependency-injector\n\n```python\n# Before (dependency-injector)\nfrom dependency_injector import containers, providers\n\nclass ApplicationContainer(containers.DeclarativeContainer):\n config = providers.Configuration()\n \n database = providers.Singleton(\n Database,\n host=config.database.host,\n port=config.database.port\n )\n \n user_service = providers.Factory(\n UserService,\n db=database\n )\n\n# After (injx)\nfrom injx import Container, Token, Scope\n\ncontainer = Container()\n\n# Define tokens\nDATABASE = Token[Database](\"database\", scope=Scope.SINGLETON)\nUSER_SERVICE = Token[UserService](\"user_service\", scope=Scope.TRANSIENT)\n\n# Register providers\ncontainer.register(\n DATABASE, \n lambda: Database(\n host=config.get(\"database.host\"),\n port=config.get(\"database.port\")\n )\n)\n\ncontainer.register(\n USER_SERVICE,\n lambda: UserService(db=container.get(DATABASE))\n)\n```\n\n### From injector\n\n```python\n# Before (injector)\nfrom injector import Injector, inject, singleton\n\ninjector = Injector()\ninjector.binder.bind(Database, to=PostgreSQLDatabase, scope=singleton)\n\n@inject\ndef user_handler(db: Database) -> None:\n pass\n\n# After (injx) \nfrom injx import Container, Token, Scope, inject\n\ncontainer = Container()\nDATABASE = Token[Database](\"database\", scope=Scope.SINGLETON)\ncontainer.register(DATABASE, PostgreSQLDatabase)\n\n@inject(container=container) # or set as default container\ndef user_handler(db: Database) -> None:\n pass\n```\n\n## Development Setup\n\n```bash\n# Clone repository\ngit clone https://github.com/qriusglobal/injx.git\ncd injx\n\n# Install with development dependencies\nuv sync\n\n# Run tests with coverage\nuv run pytest --cov=injx --cov-report=html\n\n# Type checking (strict mode)\nuvx basedpyright src\n\n# Format and lint\nuvx ruff format .\nuvx ruff check . --fix\n\n# Run all quality checks\nuvx ruff check . && uvx basedpyright src && uv run pytest -q\n```\n\n### Running Tests\n\n```bash\n# All tests\nuv run pytest\n\n# Specific test categories\nuv run pytest tests/test_container.py # Core container tests\nuv run pytest tests/test_injection.py # Injection decorator tests \nuv run pytest tests/test_contextual.py # Scoping tests\nuv run pytest tests/test_async.py # Async tests\nuv run pytest tests/test_performance.py # Performance benchmarks\nuv run pytest tests/integration/ # Integration tests\n\n# With coverage\nuv run pytest --cov=injx --cov-report=term-missing\n```\n\n## Best Practices\n\n### 1. Token Organization\n\n```python\n# tokens.py - Centralize token definitions\nfrom injx import TokenFactory, Token\nfrom typing import Protocol\n\n# Use factory for consistency\nfactory = TokenFactory()\n\n# Group related tokens\nclass DatabaseTokens:\n PRIMARY = factory.singleton(\"primary_db\", Database)\n SECONDARY = factory.singleton(\"secondary_db\", Database)\n CACHE = factory.request(\"cache\", CacheService)\n\nclass ServiceTokens:\n USER_SERVICE = factory.singleton(\"user_service\", UserService)\n EMAIL_SERVICE = factory.singleton(\"email_service\", EmailService)\n AUTH_SERVICE = factory.request(\"auth_service\", AuthService)\n\n# Use protocols for flexibility\nclass Tokens:\n LOGGER = factory.singleton(\"logger\", Logger) # Protocol\n CONFIG = factory.singleton(\"config\", Configuration) # Concrete\n```\n\n### 2. Container Lifecycle Management\n\n```python\n# app.py - Application lifecycle\nimport asyncio\nfrom contextlib import asynccontextmanager\nfrom injx import Container, set_default_container\n\n@asynccontextmanager\nasync def lifespan():\n \"\"\"Manage container lifecycle.\"\"\"\n # Startup\n container = Container()\n await setup_dependencies(container)\n set_default_container(container)\n \n try:\n yield container\n finally:\n # Shutdown - cleanup resources\n await container.aclose()\n\nasync def setup_dependencies(container: Container) -> None:\n \"\"\"Register all application dependencies.\"\"\"\n # Register database connections\n container.register(DatabaseTokens.PRIMARY, create_primary_db)\n container.register(DatabaseTokens.SECONDARY, create_secondary_db)\n \n # Register services\n container.register(ServiceTokens.USER_SERVICE, create_user_service)\n container.register(ServiceTokens.EMAIL_SERVICE, create_email_service)\n\nasync def main():\n async with lifespan() as container:\n # Application code here\n await run_application()\n\nif __name__ == \"__main__\":\n asyncio.run(main())\n```\n\n### 3. Testing Strategy\n\n```python\n# test_base.py - Shared testing infrastructure\nimport pytest\nfrom injx import Container\nfrom unittest.mock import Mock\n\nclass DITestCase:\n \"\"\"Base class for DI-enabled tests.\"\"\"\n \n def setup_method(self):\n self.container = Container()\n self.mocks = {}\n \n def mock_service(self, token: Token[T], **kwargs) -> Mock:\n \"\"\"Create and register a type-safe mock.\"\"\"\n mock = Mock(spec=token.type_, **kwargs)\n self.container.override(token, mock)\n self.mocks[token.name] = mock\n return mock\n \n def teardown_method(self):\n self.container.clear_overrides()\n\n# test_user_service.py - Concrete test\nclass TestUserService(DITestCase):\n \n def test_create_user(self):\n # Setup mocks\n mock_db = self.mock_service(DatabaseTokens.PRIMARY)\n mock_email = self.mock_service(ServiceTokens.EMAIL_SERVICE)\n \n mock_db.create_user.return_value = User(id=1, email=\"test@example.com\")\n mock_email.send_welcome_email.return_value = True\n \n # Test\n user_service = self.container.get(ServiceTokens.USER_SERVICE)\n user = user_service.create_user(\"test@example.com\")\n \n # Verify\n assert user.id == 1\n mock_db.create_user.assert_called_once()\n mock_email.send_welcome_email.assert_called_once()\n```\n\n## Troubleshooting\n\n### Common Issues\n\n**1. Circular Dependencies**\n```python\n# Problem: Services depend on each other\n# Solution: Use lazy injection or redesign\n\n# Instead of:\nclass ServiceA:\n def __init__(self, service_b: ServiceB): ...\n\nclass ServiceB:\n def __init__(self, service_a: ServiceA): ...\n\n# Do:\nclass ServiceA:\n def __init__(self, get_service_b: Callable[[], ServiceB]): ...\n\n# Or redesign to avoid circular dependency\n```\n\n**2. Type Checker Issues**\n```python\n# Problem: mypy/basedpyright can't infer types\n# Solution: Use explicit type annotations\n\n# Instead of:\n@inject\ndef handler(service): # Type unknown\n pass\n\n# Do:\n@inject \ndef handler(service: UserService) -> None: # Explicit types\n pass\n```\n\n**3. Async/Sync Mixing**\n```python\n# Problem: Using async resources in sync context\n# Solution: Use appropriate container methods\n\n# Instead of:\nasync def create_service():\n return AsyncService()\n\ncontainer.register(TOKEN, create_service)\nservice = container.get(TOKEN) # \u274c Error: async provider in sync context\n\n# Do:\nservice = await container.aget(TOKEN) # \u2705 Correct async usage\n```\n\n**4. Resource Leaks**\n```python\n# Problem: Resources not properly cleaned up\n# Solution: Use context managers or proper cleanup\n\n# Instead of:\ndef create_connection():\n return DatabaseConnection() # May leak\n\n# Do:\n@contextmanager\ndef database_connection():\n conn = DatabaseConnection()\n try:\n yield conn\n finally:\n conn.close()\n\ncontainer.register_context_sync(DB_TOKEN, database_connection)\n```\n\n## Contributing\n\n1. Fork the repository\n2. Create a feature branch (`git checkout -b feature/amazing-feature`)\n3. Write tests for your changes\n4. Ensure all quality checks pass (`uvx ruff check . && uvx basedpyright src && uv run pytest`)\n5. Submit a pull request\n\n### Code Standards\n\n- **Type safety**: All code must pass `basedpyright --strict`\n- **Testing**: Maintain 90%+ test coverage\n- **Documentation**: Update README for user-facing changes\n- **Formatting**: Use `ruff format` (88 character lines)\n- **Performance**: Maintain O(1) lookup guarantees\n\n## License\n\ninjx is licensed under the **Apache License 2.0**.\n\nThis is a permissive open source license that allows you to use injx in both open source and proprietary projects. The Apache 2.0 license provides:\n\n- **Freedom to use commercially**: Use injx in your commercial products without restrictions\n- **Patent protection**: Explicit patent grant protects you from patent claims\n- **Simple attribution**: Just include the license and copyright notice\n- **Compatible with most licenses**: Works well with MIT, BSD, and other permissive licenses\n\nSee [LICENSE](LICENSE) for the full license text.\n\nCopyright 2025 Qrius Global - Licensed under Apache License 2.0\n\n",
"bugtrack_url": null,
"license": "Apache-2.0",
"summary": "Type-safe dependency injection for Python 3.13+",
"version": "0.1.0",
"project_urls": {
"Bug Tracker": "https://github.com/qriusglobal/injx/issues",
"CI": "https://github.com/qriusglobal/injx/actions",
"Changelog": "https://github.com/qriusglobal/injx/releases",
"Documentation": "https://github.com/qriusglobal/injx#readme",
"Homepage": "https://github.com/qriusglobal/injx",
"Repository": "https://github.com/qriusglobal/injx"
},
"split_keywords": [
"async",
" dependency-injection",
" di",
" ioc",
" python3.13",
" type-safe"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "1a24a3846f67e3141c0a14314d6d3fd18b1ba8250e4ddc8996dca9330bdb0a48",
"md5": "95137ae64275f8c9e0cecdb36b08f13d",
"sha256": "d36dddf226d18e77c0ca3c27617af1aebab27536e8f06dae89593362e3863c18"
},
"downloads": -1,
"filename": "injx-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "95137ae64275f8c9e0cecdb36b08f13d",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.13",
"size": 51066,
"upload_time": "2025-09-15T08:37:18",
"upload_time_iso_8601": "2025-09-15T08:37:18.362729Z",
"url": "https://files.pythonhosted.org/packages/1a/24/a3846f67e3141c0a14314d6d3fd18b1ba8250e4ddc8996dca9330bdb0a48/injx-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "fe421dc4f9c75ae8af7026302be085ec9f54aeae1ae759527a24be445fa44230",
"md5": "6b98555de0117da01880bda7eb76d760",
"sha256": "5acb5e45ec58629188c86dfd6017bbaae6f4b6ed401af739364d0c5dfbdecc57"
},
"downloads": -1,
"filename": "injx-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "6b98555de0117da01880bda7eb76d760",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.13",
"size": 52313,
"upload_time": "2025-09-15T08:37:19",
"upload_time_iso_8601": "2025-09-15T08:37:19.563083Z",
"url": "https://files.pythonhosted.org/packages/fe/42/1dc4f9c75ae8af7026302be085ec9f54aeae1ae759527a24be445fa44230/injx-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-15 08:37:19",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "qriusglobal",
"github_project": "injx",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "injx"
}
JSON
