Python Chain of Responsibility Pattern
The Chain of Responsibility pattern is a behavioral design pattern that allows you to pass requests along a chain of handlers. Upon receiving a request, each handler can either process the request or pass it to the next handler in the chain.
Real-Life Analogy
Imagine the leave approval process in a company:
- 1-3 days→ Team leader approves
- 4-7 days→ Department manager approves
- More than 8 days→ General manager approves
When you submit a leave request, it is passed along this chain in order until it finds a person authorized to handle it. This is a real-world manifestation of the Chain of Responsibility pattern.
Core Idea of the Pattern
The core of the Chain of Responsibility pattern isdecoupling the sender and receiver of a request, allowing multiple objects the opportunity to handle the request, thus avoiding tight coupling between the sender and receiver.
Structure of the Chain of Responsibility Pattern
Let's understand the composition of the Chain of Responsibility pattern through a UML class diagram:
Core Component Description
Handler (Abstract Handler)
- Defines the interface for handling requests
- Contains a method to set the successor
- Can implement default behavior for the successor chain
ConcreteHandler (Concrete Handler)
- Implements specific handling logic
- Processes the request if it can handle it
- Otherwise forwards the request to the successor
Basic Implementation: Leave Approval System
Let's implement a simple leave approval system in Python to understand the Chain of Responsibility pattern.
Step 1: Define the Abstract Handler
Example
from typing import Optional
class Approver(ABC):
"""Approver abstract class"""
def __init__(self, name: str):
self.name = name
self.successor: Optional['Approver'] = None
def set_successor(self, successor: 'Approver'):
"""Set the successor approver"""
self.successor = successor
@abstractmethod
def process_request(self, leave_request: 'LeaveRequest'):
"""Abstract method for handling leave requests"""
pass
Step 2: Define the Request Object
Example
@dataclass
class LeaveRequest:
"""Leave request data class"""
employee_name: str # Employee name
leave_days: int # Leave days
reason: str # Leave reason
def __str__(self):
return f"{self.employee_name} requests leave for {self.leave_days} days, reason: {self.reason}"
Step 3: Implement Concrete Handlers
Example
"""Team leader - handles leave of 1-3 days"""
def process_request(self, leave_request: LeaveRequest):
if leave_request.leave_days <= 3:
print(f"Team leader {self.name} approved {leave_request}")
elif self.successor is not None:
# Exceeds authority, pass to the next level
print(f"Team leader {self.name} has no authority, transferring to a higher level")
self.successor.process_request(leave_request)
else:
print("Unable to handle the request: no suitable approver")
class DepartmentManager(Approver):
"""Department manager - handles leave of 4-7 days"""
def process_request(self, leave_request: LeaveRequest):
if 4 <= leave_request.leave_days <= 7:
print(f"Department manager {self.name} approved {leave_request}")
elif self.successor is not None:
# Exceeds authority, pass to the next level
print(f"Department manager {self.name} has no authority, transferring to a higher level")
self.successor.process_request(leave_request)
else:
print("Unable to handle the request: no suitable approver")
class GeneralManager(Approver):
"""General manager - handles leave of more than 8 days"""
def process_request(self, leave_request: LeaveRequest):
if leave_request.leave_days >= 8:
print(f"General manager {self.name} approved {leave_request}")
else:
print("Unable to handle the request: the number of leave days does not meet the requirements")
Step 4: Build the Chain and Test
Example
# Create approvers
team_leader = TeamLeader("Zhang San")
dept_manager = DepartmentManager("Li Si")
general_manager = GeneralManager("Wang Wu")
# Build the chain: Team Leader → Department Manager → General Manager
team_leader.set_successor(dept_manager)
dept_manager.set_successor(general_manager)
# Create test cases
test_cases = [
LeaveRequest("Xiao Ming", 2, "Cold and fever"),
LeaveRequest("Xiao Hong", 5, "Going home to visit family"),
LeaveRequest("Xiao Gang", 10, "Marriage leave"),
LeaveRequest("Xiao Li", 15, "Traveling abroad")
]
print("=== Leave Approval Test ==="\n")
# Process all leave requests
for request in test_cases:
print(f"Processing request: {request.employee_name} requests {request.leave_days} days leave")
team_leader.process_request(request)
print("-" * 50)
if __name__ == "__main__":
main()
Output
=== 请假审批测试 === 处理请求: 小明 请假 2 天 组长 张三 批准了 小明 申请请假 2 天，原因：感冒发烧 -------------------------------------------------- 处理请求: 小红 请假 5 天 组长 张三 无权限，转交给上级处理 部门经理 李四 批准了 小红 申请请假 5 天，原因：回家探亲 -------------------------------------------------- 处理请求: 小刚 请假 10 天 组长 张三 无权限，转交给上级处理 部门经理 李四 无权限，转交给上级处理 总经理 王五 批准了 小刚 申请请假 10 天，原因：结婚休假 -------------------------------------------------- 处理请求: 小李 请假 15 天 组长 张三 无权限，转交给上级处理 部门经理 李四 无权限，转交给上级处理 总经理 王五 批准了 小李 申请请假 15 天，原因：出国旅游 --------------------------------------------------
Advanced Application: Web Request Filter
The Chain of Responsibility pattern is very common in Web development, especially in middleware and filter chains. Let's implement a simple HTTP request filter chain.
Step 1: Define Request and Response Objects
Example
from typing import Dict, Any
@dataclass
class HttpRequest:
"""HTTP request object"""
method: str
path: str
headers: Dict[str, str]
body: Any = None
user: Dict[str, Any] = None
@dataclass
class HttpResponse:
"""HTTP response object"""
status_code: int
headers: Dict[str, str]
body: Any = None
Step 2: Implement the Filter Chain
Example
"""Filter abstract class"""
def __init__(self):
self.next_filter: Optional['Filter'] = None
def set_next(self, next_filter: 'Filter'):
"""Set the next filter"""
self.next_filter = next_filter
def do_filter(self, request: HttpRequest, response: HttpResponse) -> bool:
"""
Perform filtering operation
Return True to continue, False to break
"""
if self.handle(request, response):
if self.next_filter is not None:
return self.next_filter.do_filter(request, response)
return True
return False
@abstractmethod
def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
"""Concrete filtering logic"""
pass
class AuthenticationFilter(Filter):
"""Authentication filter"""
def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
print("Performing authentication...")
# Simulate authentication logic
token = request.headers.get('Authorization')
if token == "Bearer valid-token":
request.user = {"id": 1, "name": "Zhang San", "role": "user"}
print("✓ Authentication successful")
return True
else:
response.status_code = 401
response.body = {"error": "Unauthorized access"}
print("✗ Authentication failed")
return False
class LoggingFilter(Filter):
"""Logging filter"""
def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
print(f"Logging request: {request.method} {request.path}")
return True # Always continue
class PermissionFilter(Filter):
"""Permission check filter"""
def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
print("Checking user permissions...")
if request.user and request.user.get('role') == 'admin':
print("✓ Permission check passed")
return True
else:
response.status_code = 403
response.body = {"error": "Insufficient permissions"}
print("✗ Permission check failed")
return False
Step 3: Use the Filter Chain to Handle Requests
Example
"""Handle HTTP request"""
# Create default response
response = HttpResponse(
status_code=200,
headers={"Content-Type": "application/json"},
body={"message": "Request successful"}
)
# Create filter chain
auth_filter = AuthenticationFilter()
logging_filter = LoggingFilter()
permission_filter = PermissionFilter()
# Build the filter chain: Logging → Authentication → Permission
logging_filter.set_next(auth_filter)
auth_filter.set_next(permission_filter)
# Execute the filter chain
print("Start processing HTTP request...")
logging_filter.do_filter(request, response)
print("Request processing complete")
return response
# Test different request scenarios
def test_filter_chain():
print("=== HTTP Filter Chain Test ==="\n")
# Test case 1: Valid admin request
print("Test case 1: Valid admin request")
request1 = HttpRequest(
method="GET",
path="/admin/data",
headers={"Authorization": "Bearer valid-token"}
)
response1 = process_http_request(request1)
print(f"Response status code: {response1.status_code}")
print(f"Response body: {response1.body}"\n")
# Test case 2: Unauthorized request
print("Test case 2: Unauthorized request")
request2 = HttpRequest(
method="GET",
path="/admin/data",
headers={"Authorization": "Bearer invalid-token"}
)
response2 = process_http_request(request2)
print(f"Response status code: {response2.status_code}")
print(f"Response body: {response2.body}"\n")
if __name__ == "__main__":
test_filter_chain()
Pros and Cons of the Chain of Responsibility Pattern
Advantages
Disadvantages
The request sender does not need to know which object will handle its request, and the receiver does not need to know the full picture of the request.
2. Enhanced flexibility
Handlers can be dynamically added, removed, or reordered without affecting client code.
3. Simplified object responsibilities
Each handler only needs to focus on requests within its own responsibility scope, adhering to the Single Responsibility Principle.
4. Easy to extend
Adding a new handler is very easy; you only need to implement the handler interface and add it to the chain.
Disadvantages
1. Requests may go unhandled
If the chain of responsibility is not configured properly, a request may reach the end of the chain without being handled by any handler.
2. Performance considerations
A longer chain of responsibility can affect performance, especially when processing a large number of requests.
3. Difficult debugging
The processing path of a request may be unclear, making it difficult to trace the complete processing flow during debugging.
Real-World Application Scenarios
1. Event Handling System
The event bubbling mechanism in graphical interfaces is a typical application of the Chain of Responsibility pattern.
Example
def __init__(self, name, target):
self.name = name
self.target = target
self.handled = False
class Widget:
def __init__(self, parent=None):
self.parent = parent
def handle_event(self, event):
# Handle it yourself first; if it cannot be handled, pass it to the parent component
if self.on_event(event):
event.handled = True
elif self.parent:
self.parent.handle_event(event)
def on_event(self, event):
# Subclasses override this method to implement specific event handling
return False
2. Logging System
Log messages of different levels are handled by different handlers.
Example
# Create a logger
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)
# Create a handler chain
console_handler = logging.StreamHandler()
file_handler = logging.FileHandler('app.log')
# Set handler levels
console_handler.setLevel(logging.WARNING)
file_handler.setLevel(logging.DEBUG)
# Add to the logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)
# Usage: DEBUG and INFO levels will only be logged to a file; WARNING and above will also be output to the console.
logger.debug('Debug info') # Only logged to file
logger.warning('Warning message') # Output to both console and file
Best Practices and Considerations
1. Set Default Handling Behavior
Ensure there is a default handler at the end of the chain of responsibility to avoid lost requests.
Example
"""Default handler - handles all unhandled requests"""
def process_request(self, leave_request: LeaveRequest):
print(f"Default handler: cannot handle {leave_request}, please check the request parameters")
2. Control the Length of the Chain
Avoid creating an overly long chain of responsibility, as this reduces performance and increases debugging difficulty.
3. Clear Handling Rules
Ensure each handler has a clear handling boundary to avoid overlapping responsibilities.
4. Consider Using the Composite Pattern
For complex chains of responsibility, consider using the Composite pattern to manage the relationships between handlers.
Summary
The Chain of Responsibility pattern is a powerful and flexible design pattern. By decoupling the sender and receiver of a request, it provides an elegant way to handle scenarios that require multiple objects to collaborate.
Key Points
- Core idea: Give multiple objects the opportunity to handle the request, avoiding coupling between the request sender and receiver.
- Applicable scenarios: Multiple objects can handle the same request, but which object handles it needs to be determined at runtime.
- Implementation points: Define a handler interface, build the handler chain, and pass the request.
- Python features: Using Python's dynamic features, the Chain of Responsibility pattern can be implemented more flexibly.