Overview
The ServiceContainer is a comprehensive dependency injection (DI) container that implements automatic service discovery, lazy loading, circular dependency detection, and performance optimization. It follows the singleton pattern and provides both eager and lazy initialization strategies.
Key Features
- Autowiring: Automatic dependency resolution based on type hints
- Service Discovery: Automatic scanning of modules for service classes
- Caching: Intelligent caching of service definitions and instances
- Background Scanning: Non-blocking service discovery for large projects
- Memory Management: Cleanup capabilities to manage memory usage
- Factory Pattern Support: Custom factories for complex object creation
- Tag-Based Retrieval: Organize and retrieve services by tags
Thread Safety
The container is thread-safe for service retrieval but not for registration. Service registration should be completed during initialization.
Initialization
from framefox.core.di.service_container import ServiceContainer
# Get the singleton instance
container = ServiceContainer()
The ServiceContainer automatically initializes when first instantiated:
- Registers core factories
- Discovers and registers essential services
- Scans for service definitions in framework and application code
- Sets up caching for performance optimization
Core Methods
get()
Retrieve a service instance with automatic dependency injection.
The class type of the service to retrieve
The instantiated service with all dependencies resolved
from src.services.email_service import EmailService
# Get service instance
email_service = container.get(EmailService)
# Automatically resolves dependencies
class UserController:
def __init__(self, email_service: EmailService):
self.email_service = email_service
controller = container.get(UserController)
# EmailService is automatically injected
Behavior:
- Returns cached instance if available
- Creates new instance with autowired dependencies
- Detects and raises
CircularDependencyError for circular dependencies
- Raises
ServiceNotFoundError if service cannot be found or created
- Raises
ServiceInstantiationError if instantiation fails
set_instance()
Manually register a pre-configured service instance.
The class type of the service
The service instance to register
from src.services.cache_service import CacheService
# Create custom instance
cache = CacheService(ttl=3600, max_size=1000)
# Register it in the container
container.set_instance(CacheService, cache)
# Future calls to get() will return this instance
retrieved = container.get(CacheService)
assert retrieved is cache
get_by_name()
Retrieve a service by its class name (string).
The name of the service class
The service instance, or None if not found
# Get service by string name
settings = container.get_by_name("Settings")
session = container.get_by_name("Session")
get_by_tag()
Retrieve the first service associated with a specific tag.
The first matching service instance, or None if not found
# Get first controller service
controller = container.get_by_tag("controller")
# Get first repository service
repo = container.get_by_tag("repository")
Note: If multiple services match the tag, a warning is logged and the first one is returned.
get_all_by_tag()
Retrieve all services associated with a specific tag.
List of all matching service instances
# Get all controllers
controllers = container.get_all_by_tag("controller")
for controller in controllers:
print(f"Registered: {controller.__class__.__name__}")
# Get all middleware
middleware_list = container.get_all_by_tag("middleware")
get_by_tag_prefix()
Retrieve all services where at least one tag starts with the given prefix.
List of all services with matching tag prefixes
# Get all services in the security namespace
security_services = container.get_by_tag_prefix("security")
# Returns services tagged with "security.auth", "security.token", etc.
# Get all ORM-related services
orm_services = container.get_by_tag_prefix("orm")
has()
Check if a service class is registered in the container.
True if the service is registered, False otherwise
from src.services.payment_service import PaymentService
if container.has(PaymentService):
payment = container.get(PaymentService)
else:
print("Payment service not registered")
has_by_name()
Check if a service is registered by its class name.
True if the service is registered, False otherwise
if container.has_by_name("EmailService"):
email = container.get_by_name("EmailService")
register_factory()
Register a custom factory for creating complex service instances.
The factory instance to register
from framefox.core.di.service_factory import ServiceFactory
class DatabaseConnectionFactory(ServiceFactory):
def can_create(self, service_class: Type) -> bool:
return service_class.__name__ == "DatabaseConnection"
def create(self, service_class: Type, container) -> Any:
# Custom creation logic
return DatabaseConnection(
host=container.get_by_name("Settings").db_host,
pool_size=10
)
container.register_factory(DatabaseConnectionFactory())
Memory Management
cleanup_memory()
Clean up container memory by removing non-essential caches.
# Periodic cleanup in long-running applications
container.cleanup_memory()
Behavior:
- Clears resolution cache
- Removes cached module definitions (except essential framework modules)
- Triggers garbage collection
- Logs cleanup statistics
clear_cache()
Clear all caches including service definitions.
# Complete cache reset
container.clear_cache()
rebuild_cache()
Force a complete cache rebuild by rescanning all services.
# Useful after dynamic code changes
container.rebuild_cache()
Advanced Features
freeze_registry()
Freeze the service registry to prevent further modifications.
# After all services are registered
container.freeze_registry()
Note: The container can still force-register services after freezing if needed.
force_complete_scan()
Force immediate synchronous scanning of all source directories.
# Ensure all services are discovered immediately
container.force_complete_scan()
disable_background_scan()
Disable background service scanning.
# For testing or controlled environments
container.disable_background_scan()
get_scan_status()
Get detailed information about the scanning state.
Dictionary containing scan status information
status = container.get_scan_status()
print(status)
# {
# "src_scanned": True,
# "src_scan_in_progress": False,
# "scanned_modules_count": 45,
# "cached_modules_count": 42,
# "src_paths_count": 1,
# "background_scan_enabled": False
# }
get_stats()
Get comprehensive container statistics.
Dictionary containing container statistics
stats = container.get_stats()
print(stats)
# {
# "container_instance": 1,
# "instantiated_services": 23,
# "cached_resolutions": 45,
# "registered_factories": 2,
# "total_definitions": 67,
# "public_services": 34,
# "tagged_services": 28
# }
Dependency Injection Patterns
Constructor Injection
The recommended pattern for dependency injection:
from framefox.core.di.service_container import ServiceContainer
class UserRepository:
def find_by_id(self, user_id: int):
# Repository logic
pass
class EmailService:
def send(self, to: str, subject: str, body: str):
# Email sending logic
pass
class UserService:
def __init__(self, user_repo: UserRepository, email: EmailService):
self.user_repo = user_repo
self.email = email
def register_user(self, email: str, name: str):
# Dependencies are automatically injected
user = self.user_repo.create(email=email, name=name)
self.email.send(email, "Welcome!", f"Hello {name}")
return user
container = ServiceContainer()
user_service = container.get(UserService)
# UserRepository and EmailService are automatically injected
Interface-Based Injection
Using abstract base classes or protocols:
from abc import ABC, abstractmethod
from typing import Protocol
class CacheInterface(Protocol):
def get(self, key: str) -> Any: ...
def set(self, key: str, value: Any): ...
class RedisCache:
def get(self, key: str) -> Any:
# Redis implementation
pass
def set(self, key: str, value: Any):
# Redis implementation
pass
class DataService:
def __init__(self, cache: RedisCache): # Type hint with concrete class
self.cache = cache
# Register the implementation
container.set_instance(RedisCache, RedisCache())
service = container.get(DataService)
Optional Dependencies
Handling optional dependencies with defaults:
from typing import Optional
class LoggingService:
def __init__(self,
email: EmailService,
cache: Optional[CacheService] = None):
self.email = email
self.cache = cache or InMemoryCache()
Error Handling
The container raises specific exceptions for different error conditions:
from framefox.core.debug.exception.di_exception import (
CircularDependencyError,
ServiceInstantiationError,
ServiceNotFoundError,
)
try:
service = container.get(MyService)
except CircularDependencyError as e:
# Circular dependency detected
print(f"Circular dependency: {e}")
except ServiceNotFoundError as e:
# Service not registered
print(f"Service not found: {e}")
except ServiceInstantiationError as e:
# Error during service creation
print(f"Failed to create service: {e}")
Best Practices
-
Use Type Hints: Always use type hints for constructor parameters to enable autowiring
-
Constructor Injection: Prefer constructor injection over property injection for required dependencies
-
Service Tagging: Use tags to organize and retrieve related services
# In service definition or configuration
ServiceDefinition(
UserController,
tags=["controller", "user", "api.v1"]
)
-
Memory Management: Call
cleanup_memory() periodically in long-running applications
-
Testing: Use
set_instance() to inject mock services for testing
# In tests
mock_email = Mock(spec=EmailService)
container.set_instance(EmailService, mock_email)
- Avoid Circular Dependencies: Design services to avoid circular references
- Services are created once and cached (singleton pattern by default)
- Resolution cache provides O(1) lookup for previously resolved services
- Background scanning doesn’t block application startup
- Module scan cache reduces repeated imports
- Memory cleanup removes non-essential caches while preserving core services