Circular Dependencies¶
Circular dependencies occur when two or more services depend on each other, creating a dependency loop that can cause resolution failures or infinite loops.
🔄 Understanding Circular Dependencies¶
What are Circular Dependencies?¶
# ❌ Circular dependency example
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
# This creates a circular dependency:
# ServiceA -> ServiceB -> ServiceA
Types of Circular Dependencies¶
# 1. Direct circular dependency
class DirectCircularA:
def __init__(self, b: DirectCircularB):
self.b = b
class DirectCircularB:
def __init__(self, a: DirectCircularA):
self.a = a
# 2. Indirect circular dependency
class IndirectCircularA:
def __init__(self, b: IndirectCircularB):
self.b = b
class IndirectCircularB:
def __init__(self, c: IndirectCircularC):
self.c = c
class IndirectCircularC:
def __init__(self, a: IndirectCircularA):
self.a = a
# 3. Self-dependency (rare but possible)
class SelfDependency:
def __init__(self, self_ref):
self.self_ref = self_ref
🔍 Circular Dependency Detection¶
Automatic Detection¶
from injectq.core.circulardeps import CircularDependencyDetector
# Automatic circular dependency detection
detector = CircularDependencyDetector(container)
# Detect circular dependencies
circular_deps = detector.detect_circular_dependencies()
print("Circular dependencies found:")
for dep_chain in circular_deps:
print(f"- {' -> '.join(cls.__name__ for cls in dep_chain)}")
# Check if specific service has circular dependency
has_circular = detector.has_circular_dependency(SomeService)
print(f"SomeService has circular dependency: {has_circular}")
# Get circular dependency chains for specific service
chains = detector.get_circular_chains(SomeService)
for chain in chains:
print(f"Circular chain: {' -> '.join(cls.__name__ for cls in chain)}")
Dependency Graph Analysis¶
from injectq.core.circulardeps import DependencyGraphAnalyzer
# Analyze dependency graph for circular dependencies
analyzer = DependencyGraphAnalyzer(container)
# Build dependency graph
graph = analyzer.build_dependency_graph()
# Find all circular dependencies
circular_paths = analyzer.find_circular_paths()
print("Circular dependency paths:")
for path in circular_paths:
print(f"- {' -> '.join(cls.__name__ for cls in path)}")
# Get strongly connected components
scc = analyzer.get_strongly_connected_components()
print("Strongly connected components:")
for component in scc:
if len(component) > 1: # Only show components with cycles
print(f"- {', '.join(cls.__name__ for cls in component)}")
Runtime Detection¶
# Runtime circular dependency detection
class RuntimeCircularDetector:
"""Detect circular dependencies at runtime."""
def __init__(self):
self.resolution_stack = []
self.visited = set()
def detect_during_resolution(self, service_type):
"""Detect circular dependency during service resolution."""
if service_type in self.resolution_stack:
# Circular dependency found
cycle_start = self.resolution_stack.index(service_type)
cycle = self.resolution_stack[cycle_start:] + [service_type]
raise CircularDependencyError(f"Circular dependency detected: {' -> '.join(cls.__name__ for cls in cycle)}")
if service_type in self.visited:
return # Already processed
self.resolution_stack.append(service_type)
try:
# Get dependencies of this service
dependencies = self.get_dependencies(service_type)
for dep in dependencies:
self.detect_during_resolution(dep)
self.visited.add(service_type)
finally:
self.resolution_stack.pop()
def get_dependencies(self, service_type):
"""Get dependencies of a service type."""
# This would integrate with container's dependency resolution
return []
# Usage
detector = RuntimeCircularDetector()
try:
detector.detect_during_resolution(SomeService)
except CircularDependencyError as e:
print(f"Circular dependency: {e}")
🛠️ Resolving Circular Dependencies¶
Method 1: Property Injection¶
# ✅ Solution: Use property injection to break circular dependency
class PropertyInjectionA:
def __init__(self):
self.service_b = None # Injected later
def set_service_b(self, service_b: PropertyInjectionB):
self.service_b = service_b
class PropertyInjectionB:
def __init__(self, service_a: PropertyInjectionA):
self.service_a = service_a
# Setup with property injection
def setup_property_injection(container):
# Create instances
service_a = PropertyInjectionA()
service_b = PropertyInjectionB(service_a)
# Break circular dependency with property injection
service_a.set_service_b(service_b)
# Bind to container
container.bind(PropertyInjectionA, service_a)
container.bind(PropertyInjectionB, service_b)
# Usage
setup_property_injection(container)
service_a = container.get(PropertyInjectionA)
service_b = container.get(PropertyInjectionB)
Method 2: Interface Segregation¶
# ✅ Solution: Use interfaces to break circular dependency
from abc import ABC, abstractmethod
class IServiceA(ABC):
@abstractmethod
def method_a(self):
pass
class IServiceB(ABC):
@abstractmethod
def method_b(self):
pass
class InterfaceSegregationA(IServiceA):
def __init__(self, service_b: IServiceB):
self.service_b = service_b
def method_a(self):
return f"A calling B: {self.service_b.method_b()}"
class InterfaceSegregationB(IServiceB):
def __init__(self):
self.service_a = None # Will be set later
def set_service_a(self, service_a: IServiceA):
self.service_a = service_a
def method_b(self):
if self.service_a:
return f"B calling A: {self.service_a.method_a()}"
return "B: No service A available"
# Setup with interface segregation
def setup_interface_segregation(container):
# Bind interface to implementation
container.bind(IServiceA, InterfaceSegregationA)
container.bind(IServiceB, InterfaceSegregationB)
# Create instances
service_b = container.get(IServiceB)
service_a = container.get(IServiceA)
# Set the circular reference
service_b.set_service_a(service_a)
# Usage
setup_interface_segregation(container)
service_a = container.get(IServiceA)
result = service_a.method_a() # This will work without circular dependency
Method 3: Factory Pattern¶
# ✅ Solution: Use factory pattern to break circular dependency
class FactoryA:
def __init__(self, value: str):
self.value = value
def get_service_b(self, container):
"""Lazy creation of ServiceB."""
return container.get(FactoryB)
class FactoryB:
def __init__(self, value: str):
self.value = value
def get_service_a(self, container):
"""Lazy creation of ServiceA."""
return container.get(FactoryA)
# Setup with factory pattern
def setup_factory_pattern(container):
container.bind(FactoryA, lambda: FactoryA("from_a"))
container.bind(FactoryB, lambda: FactoryB("from_b"))
# Usage
setup_factory_pattern(container)
service_a = container.get(FactoryA)
service_b = service_a.get_service_b(container) # Lazy resolution
service_a_again = service_b.get_service_a(container) # Lazy resolution
Method 4: Service Locator Pattern¶
# ✅ Solution: Use service locator to break circular dependency
class ServiceLocator:
"""Simple service locator to break circular dependencies."""
def __init__(self):
self.services = {}
def register(self, service_type, service_instance):
self.services[service_type] = service_instance
def get(self, service_type):
return self.services.get(service_type)
class ServiceLocatorA:
def __init__(self, locator: ServiceLocator):
self.locator = locator
def call_service_b(self):
service_b = self.locator.get(ServiceLocatorB)
return service_b.do_something()
class ServiceLocatorB:
def __init__(self, locator: ServiceLocator):
self.locator = locator
def call_service_a(self):
service_a = self.locator.get(ServiceLocatorA)
return service_a.do_something()
def do_something(self):
return "ServiceB result"
# Setup with service locator
def setup_service_locator(container):
locator = ServiceLocator()
# Create services with locator
service_a = ServiceLocatorA(locator)
service_b = ServiceLocatorB(locator)
# Register services
locator.register(ServiceLocatorA, service_a)
locator.register(ServiceLocatorB, service_b)
# Bind to container
container.bind(ServiceLocatorA, service_a)
container.bind(ServiceLocatorB, service_b)
# Usage
setup_service_locator(container)
service_a = container.get(ServiceLocatorA)
result = service_a.call_service_b()
Method 5: Event-Driven Architecture¶
# ✅ Solution: Use events to break circular dependency
from typing import Callable
class EventBus:
"""Simple event bus for decoupling services."""
def __init__(self):
self.listeners = {}
def subscribe(self, event_type: str, listener: Callable):
if event_type not in self.listeners:
self.listeners[event_type] = []
self.listeners[event_type].append(listener)
def publish(self, event_type: str, data=None):
if event_type in self.listeners:
for listener in self.listeners[event_type]:
listener(data)
class EventDrivenA:
def __init__(self, event_bus: EventBus):
self.event_bus = event_bus
self.event_bus.subscribe("service_b_event", self.handle_service_b_event)
def do_something(self):
# Instead of calling ServiceB directly, publish an event
self.event_bus.publish("service_a_event", {"data": "from_a"})
return "ServiceA result"
def handle_service_b_event(self, data):
print(f"ServiceA received event from ServiceB: {data}")
class EventDrivenB:
def __init__(self, event_bus: EventBus):
self.event_bus = event_bus
self.event_bus.subscribe("service_a_event", self.handle_service_a_event)
def do_something(self):
# Instead of calling ServiceA directly, publish an event
self.event_bus.publish("service_b_event", {"data": "from_b"})
return "ServiceB result"
def handle_service_a_event(self, data):
print(f"ServiceB received event from ServiceA: {data}")
# Setup with event-driven architecture
def setup_event_driven(container):
event_bus = EventBus()
container.bind(EventBus, event_bus)
container.bind(EventDrivenA, EventDrivenA)
container.bind(EventDrivenB, EventDrivenB)
# Usage
setup_event_driven(container)
service_a = container.get(EventDrivenA)
service_b = container.get(EventDrivenB)
result_a = service_a.do_something() # Triggers event to ServiceB
result_b = service_b.do_something() # Triggers event to ServiceA
🏗️ Advanced Circular Dependency Resolution¶
Lazy Resolution¶
# Lazy resolution to break circular dependencies
class LazyResolver:
"""Lazy dependency resolver."""
def __init__(self, container):
self.container = container
self._cache = {}
def get_lazy(self, service_type):
"""Get lazy resolver for service type."""
if service_type not in self._cache:
self._cache[service_type] = LazyService(service_type, self.container)
return self._cache[service_type]
class LazyService:
"""Lazy service wrapper."""
def __init__(self, service_type, container):
self.service_type = service_type
self.container = container
self._instance = None
def __call__(self):
"""Resolve service when called."""
if self._instance is None:
self._instance = self.container.get(self.service_type)
return self._instance
class LazyA:
def __init__(self, lazy_resolver: LazyResolver):
self.lazy_b = lazy_resolver.get_lazy(LazyB)
def call_b(self):
service_b = self.lazy_b() # Resolved when called
return service_b.do_something()
class LazyB:
def __init__(self, lazy_resolver: LazyResolver):
self.lazy_a = lazy_resolver.get_lazy(LazyA)
def call_a(self):
service_a = self.lazy_a() # Resolved when called
return service_a.do_something()
def do_something(self):
return "LazyB result"
# Setup with lazy resolution
def setup_lazy_resolution(container):
lazy_resolver = LazyResolver(container)
container.bind(LazyResolver, lazy_resolver)
container.bind(LazyA, LazyA)
container.bind(LazyB, LazyB)
# Usage
setup_lazy_resolution(container)
service_a = container.get(LazyA)
result = service_a.call_b() # ServiceB resolved lazily
Proxy Pattern¶
# Proxy pattern to break circular dependencies
class ServiceProxy:
"""Proxy for delayed service resolution."""
def __init__(self, service_type, container):
self.service_type = service_type
self.container = container
self._real_service = None
def _get_real_service(self):
if self._real_service is None:
self._real_service = self.container.get(self.service_type)
return self._real_service
def __getattr__(self, name):
"""Delegate attribute access to real service."""
return getattr(self._get_real_service(), name)
class ProxyA:
def __init__(self, proxy_b: ServiceProxy):
self.proxy_b = proxy_b
def call_b(self):
return self.proxy_b.do_something()
class ProxyB:
def __init__(self, proxy_a: ServiceProxy):
self.proxy_a = proxy_a
def call_a(self):
return self.proxy_a.call_b()
def do_something(self):
return "ProxyB result"
# Setup with proxy pattern
def setup_proxy_pattern(container):
# Create proxies
proxy_a = ServiceProxy(ProxyA, container)
proxy_b = ServiceProxy(ProxyB, container)
# Bind proxies
container.bind(ServiceProxy, proxy_a, name="proxy_a")
container.bind(ServiceProxy, proxy_b, name="proxy_b")
# Bind real services
container.bind(ProxyA, lambda: ProxyA(proxy_b))
container.bind(ProxyB, lambda: ProxyB(proxy_a))
# Usage
setup_proxy_pattern(container)
service_a = container.get(ProxyA)
result = service_a.call_b() # Uses proxy to access ServiceB
Dependency Injection Container Features¶
# Using InjectQ's circular dependency resolution features
from injectq.core.circulardeps import CircularDependencyResolver
resolver = CircularDependencyResolver(container)
# Automatic circular dependency resolution
@resolver.resolve_circular
class AutoResolvedA:
def __init__(self, b: AutoResolvedB):
self.b = b
def call_b(self):
return self.b.do_something()
@resolver.resolve_circular
class AutoResolvedB:
def __init__(self, a: AutoResolvedA):
self.a = a
def call_a(self):
return self.a.call_b()
def do_something(self):
return "AutoResolvedB result"
# Setup with automatic resolution
def setup_automatic_resolution(container):
container.bind(AutoResolvedA, AutoResolvedA)
container.bind(AutoResolvedB, AutoResolvedB)
# Usage
setup_automatic_resolution(container)
service_a = container.get(AutoResolvedA)
result = service_a.call_b() # Works despite circular dependency
📊 Circular Dependency Analysis¶
Dependency Graph Visualization¶
from injectq.core.circulardeps import DependencyGraphVisualizer
# Visualize dependency graph
visualizer = DependencyGraphVisualizer(container)
# Generate dependency graph
graph_data = visualizer.generate_graph()
# Save as different formats
visualizer.save_as_dot("dependencies.dot")
visualizer.save_as_png("dependencies.png")
visualizer.save_as_svg("dependencies.svg")
# Highlight circular dependencies
circular_graph = visualizer.highlight_circular_dependencies()
visualizer.save_as_png("circular_dependencies.png", graph=circular_graph)
Impact Analysis¶
from injectq.core.circulardeps import CircularDependencyAnalyzer
# Analyze impact of circular dependencies
analyzer = CircularDependencyAnalyzer(container)
# Analyze specific circular dependency
impact = analyzer.analyze_impact(SomeService)
print("Circular Dependency Impact Analysis:")
print(f"- Involved services: {impact.involved_services}")
print(f"- Resolution depth: {impact.resolution_depth}")
print(f"- Performance impact: {impact.performance_impact}")
print(f"- Maintenance complexity: {impact.maintenance_complexity}")
# Get resolution recommendations
recommendations = analyzer.get_recommendations(SomeService)
print("Resolution Recommendations:")
for rec in recommendations:
print(f"- {rec.strategy}: {rec.description}")
print(f" Difficulty: {rec.difficulty}")
print(f" Benefits: {rec.benefits}")
Metrics and Monitoring¶
from injectq.core.circulardeps import CircularDependencyMonitor
# Monitor circular dependencies
monitor = CircularDependencyMonitor(container)
# Get circular dependency metrics
metrics = monitor.get_metrics()
print("Circular Dependency Metrics:")
print(f"- Total circular dependencies: {metrics.total_circular_deps}")
print(f"- Most complex cycle: {metrics.most_complex_cycle}")
print(f"- Average cycle length: {metrics.avg_cycle_length}")
print(f"- Resolution success rate: {metrics.resolution_success_rate}%")
# Monitor resolution attempts
with monitor.track_resolution(SomeService) as tracking:
service = container.get(SomeService)
resolution_metrics = tracking.get_metrics()
print("Resolution Metrics:")
print(f"- Resolution time: {resolution_metrics.resolution_time}ms")
print(f"- Cycle detected: {resolution_metrics.cycle_detected}")
print(f"- Resolution strategy used: {resolution_metrics.strategy_used}")
🎯 Best Practices¶
✅ Good Practices¶
1. Design for Testability¶
# ✅ Good: Design interfaces to avoid circular dependencies
class RepositoryInterface:
def get_data(self, id: str):
pass
class ServiceInterface:
def process_data(self, data: dict):
pass
class RepositoryImpl(RepositoryInterface):
def __init__(self, config: dict):
self.config = config
def get_data(self, id: str):
# Implementation
return {"id": id, "data": "from_repository"}
class ServiceImpl(ServiceInterface):
def __init__(self, repository: RepositoryInterface):
self.repository = repository
def process_data(self, data: dict):
# No circular dependency
repo_data = self.repository.get_data(data["id"])
return {"processed": True, "data": repo_data}
# Usage
container.bind(RepositoryInterface, RepositoryImpl)
container.bind(ServiceInterface, ServiceImpl)
2. Use Dependency Inversion Principle¶
# ✅ Good: Use dependency inversion to break circular dependencies
from abc import ABC, abstractmethod
class NotificationServiceInterface(ABC):
@abstractmethod
def send_notification(self, message: str):
pass
class UserServiceInterface(ABC):
@abstractmethod
def get_user(self, user_id: str):
pass
class NotificationService(NotificationServiceInterface):
def __init__(self, user_service: UserServiceInterface):
self.user_service = user_service
def send_notification(self, message: str):
# Can get user information without circular dependency
user = self.user_service.get_user("current_user")
print(f"Sending notification to {user['name']}: {message}")
class UserService(UserServiceInterface):
def __init__(self, notification_service: NotificationServiceInterface = None):
self.notification_service = notification_service
def get_user(self, user_id: str):
user = {"id": user_id, "name": "John Doe"}
# Optional notification (no circular dependency)
if self.notification_service:
self.notification_service.send_notification(f"User {user_id} accessed")
return user
# Usage
container.bind(NotificationServiceInterface, NotificationService)
container.bind(UserServiceInterface, UserService)
3. Event-Driven Communication¶
# ✅ Good: Use events for cross-service communication
class EventDrivenRepository:
def __init__(self, event_publisher):
self.event_publisher = event_publisher
def save_data(self, data: dict):
# Save data
saved_data = {"id": "123", **data}
# Publish event instead of calling service directly
self.event_publisher.publish("data_saved", saved_data)
return saved_data
class EventDrivenProcessor:
def __init__(self, event_subscriber):
self.event_subscriber = event_subscriber
self.event_subscriber.subscribe("data_saved", self.process_saved_data)
def process_saved_data(self, data):
# Process the saved data
print(f"Processing saved data: {data}")
# Usage
event_bus = EventBus()
container.bind(EventBus, event_bus)
container.bind(EventDrivenRepository, EventDrivenRepository)
container.bind(EventDrivenProcessor, EventDrivenProcessor)
❌ Bad Practices¶
1. Tight Coupling¶
# ❌ Bad: Tight coupling leads to circular dependencies
class TightlyCoupledA:
def __init__(self, b: TightlyCoupledB):
self.b = b
def process(self):
return self.b.process()
class TightlyCoupledB:
def __init__(self, a: TightlyCoupledA):
self.a = a
def process(self):
return self.a.process() # Circular call
# ✅ Better: Use interfaces and proper separation
class ProcessingInterface:
def process(self):
pass
class DecoupledA(ProcessingInterface):
def __init__(self, processor: ProcessingInterface):
self.processor = processor
def process(self):
return f"A: {self.processor.process()}"
class DecoupledB(ProcessingInterface):
def __init__(self):
pass # No circular dependency
def process(self):
return "B processed"
2. Service Locator Anti-Pattern¶
# ❌ Bad: Global service locator can hide circular dependencies
class GlobalServiceLocator:
_instance = None
services = {}
@classmethod
def get_instance(cls):
if cls._instance is None:
cls._instance = cls()
return cls._instance
def get_service(self, service_type):
return self.services.get(service_type)
# ✅ Better: Explicit dependency injection
class ExplicitDependenciesA:
def __init__(self, dependency_b):
self.dependency_b = dependency_b
class ExplicitDependenciesB:
def __init__(self, dependency_a):
self.dependency_a = dependency_a
🎯 Summary¶
Circular dependencies create resolution challenges:
- Detection - Automatic and runtime circular dependency detection
- Resolution strategies - Property injection, interfaces, factories, service locators, events
- Advanced techniques - Lazy resolution, proxy pattern, automatic resolution
- Analysis tools - Graph visualization, impact analysis, monitoring
- Best practices - Design for testability, dependency inversion, event-driven communication
Key features: - Comprehensive circular dependency detection - Multiple resolution strategies - Dependency graph analysis and visualization - Impact analysis and recommendations - Performance monitoring and metrics
Best practices: - Design with interfaces to prevent circular dependencies - Use dependency inversion principle - Implement event-driven communication - Avoid tight coupling between services - Use lazy resolution when necessary - Monitor and analyze dependency graphs
Common resolution patterns: - Property injection for breaking direct cycles - Interface segregation for loose coupling - Factory pattern for lazy instantiation - Service locator for centralized access - Event-driven architecture for decoupling - Proxy pattern for delayed resolution
Ready to explore profiling?