YouTipPython Chain of Responsibility Pattern
\n\nThe 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.
\n\nReal-life Analogy
\n\nImagine the leave approval process in a company:
\n\n- \n
- 1-3 days → Team Leader approval \n
- 4-7 days → Department Manager approval \n
- 8+ days → General Manager approval \n
When you submit a leave request, it passes through this chain until it finds someone with the authority to handle it. This is the real-world manifestation of the Chain of Responsibility pattern.
\n\nCore Concept of the Pattern
\n\nThe core of the Chain of Responsibility pattern is to decouple the sender of a request from its receivers, allowing multiple objects to have the opportunity to handle the request, thereby avoiding tight coupling between the request sender and receivers.
\n\n\n\n
Structure of the Chain of Responsibility Pattern
\n\nLet's understand the composition of the Chain of Responsibility pattern through a UML class diagram:
\n\nCore Component Descriptions
\n\nHandler (Abstract Handler)
\n\n- \n
- Defines the interface for handling requests \n
- Includes methods for setting the successor \n
- Can implement default behavior for the successor chain \n
ConcreteHandler (Concrete Handler)
\n\n- \n
- Implements specific processing logic \n
- If it can handle the request, it processes it \n
- Otherwise, forwards the request to the successor \n
\n\n
Basic Implementation: Leave Approval System
\n\nLet's implement a simple leave approval system in Python to understand the Chain of Responsibility pattern.
\n\nStep 1: Define the Abstract Handler
\n\nfrom abc import ABC, abstractmethod\nfrom typing import Optional\n\nclass Approver(ABC):\n """Abstract approver class"""\n def __init__(self, name: str):\n self.name = name\n self.successor: Optional['Approver'] = None\n \n def set_successor(self, successor: 'Approver'):\n """Set successor approver"""\n self.successor = successor\n \n @abstractmethod\n def process_request(self, leave_request: 'LeaveRequest'):\n """Abstract method to process leave requests"""\n pass\nStep 2: Define the Request Object
\n\nfrom dataclasses import dataclass\n\n@dataclass\nclass LeaveRequest:\n """Leave request data class"""\n employee_name: str # Employee Name\n leave_days: int # Leave Days\n reason: str # Leave reason\n \n def __str__(self):\n return f"{self.employee_name} Apply for Leave {self.leave_days} days, reason:{self.reason}"\nStep 3: Implement Concrete Handlers
\n\nclass TeamLeader(Approver):\n """Team Leader - Handle 1-3 days of leave"""\n def process_request(self, leave_request: LeaveRequest):\n if leave_request.leave_days <= 3:\n print(f"Team Leader {self.name} Approved {leave_request}")\n elif self.successor is not None:\n # Exceeds Permissions, pass to the next level\n print(f"Team Leader {self.name} No permission, forwarding to superior")\n self.successor.process_request(leave_request)\n else:\n print("Unable to process request: No suitable approver")\n\nclass DepartmentManager(Approver):\n """Department Manager - Handle 4-7 Leave request for days"""\n def process_request(self, leave_request: LeaveRequest):\n if 4 <= leave_request.leave_days = 8:\n print(f"General Manager {self.name} Approved {leave_request}")\n else:\n print("Cannot process request: Leave Days do not meet requirements")\nStep 4: Build the Chain of Responsibility and Test
\n\ndef main():\n # Create approver\n team_leader = TeamLeader("Zhang San")\n dept_manager = DepartmentManager("Li Si")\n general_manager = GeneralManager("Wang Wu")\n \n # Build Chain of Responsibility: Team Leader → Department Manager → General Manager\n team_leader.set_successor(dept_manager)\n dept_manager.set_successor(general_manager)\n \n # Create test cases\n test_cases = [\n LeaveRequest("Xiaoming", 2, "Cold and fever"),\n LeaveRequest("Xiaohong", 5, "Visit family"),\n LeaveRequest("Xiao Gang", 10, "Marriage leave"),\n LeaveRequest("Xiao Li", 15, "Overseas travel")\n ]\n \n print("=== Leave Approval Test ===n")\n \n # Handle All Leave Requests\n for request in test_cases:\n print(f"Process Request: {request.employee_name} Leave request {request.leave_days} days")\n team_leader.process_request(request)\n print("-" * 50)\n\nif __name__ == "__main__":\n main()\nOutput
\n\n=== Leave Approval Test ===\nProcess Request: Xiao Ming Leave request 2 days\nTeam Leader Zhang San Approved Xiao Ming Apply for Leave 2 days, reason:Cold and fever\n--------------------------------------------------\nProcess Request: Xiao Hong requests 5 days of leave\nTeam Leader Zhang San No permission, forwarding to superior\nDepartment Manager Li Si Approved Xiao Hong Apply for Leave 5 days, reason:Return home to visit family\n--------------------------------------------------\nProcess Request: Xiao Gang requests 10 days of leave\nTeam Leader Zhang San No permission, forwarding to superior\nDepartment Manager Li Si lacks permission, forwarding to superior\nGeneral Manager Wang Wu approved Xiao Gang's 10-day leave request, reason: Wedding leave\n--------------------------------------------------\nProcess Request: Xiao Li requests 15 days of leave\nTeam Leader Zhang San No permission, forwarding to superior\nDepartment Manager Li Si lacks permission, forwarding to superior\nGeneral Manager Wang Wu Approved Xiao Li Apply for Leave 15 days, reason:Travel abroad\n--------------------------------------------------\n\n\n
Advanced Application: Web Request Filter
\n\nThe 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.
\n\nStep 1: Define Request and Response Objects
\n\nfrom dataclasses import dataclass\nfrom typing import Dict, Any\n\n@dataclass\nclass HttpRequest:\n """HTTP Request Object"""\n method: str\n path: str\n headers: Dict[str, str]\n body: Any = None\n user: Dict[str, Any] = None\n\n@dataclass\nclass HttpResponse:\n """HTTP Response Object"""\n status_code: int\n headers: Dict[str, str]\n body: Any = None\nStep 2: Implement the Filter Chain
\n\nclass Filter(ABC):\n """Abstract filter class"""\n def __init__(self):\n self.next_filter: Optional['Filter'] = None\n \n def set_next(self, next_filter: 'Filter'):\n """Set Next Filter"""\n self.next_filter = next_filter\n \n def do_filter(self, request: HttpRequest, response: HttpResponse) -> bool:\n """\n Execute filter operation\n Return True to continue execution, False to interrupt\n """\n if self.handle(request, response):\n if self.next_filter is not None:\n return self.next_filter.do_filter(request, response)\n return True\n return False\n \n @abstractmethod\n def handle(self, request: HttpRequest, response: HttpResponse) -> bool:\n """Specific filter processing logic"""\n pass\n\nclass AuthenticationFilter(Filter):\n """Identity Authentication Filter"""\n def handle(self, request: HttpRequest, response: HttpResponse) -> bool:\n print("Execute Authentication...")\n # Simulate Authentication logic\n token = request.headers.get('Authorization')\n if token == "Bearer valid-token":\n request.user = {"id": 1, "name": "Zhang San", "role": "user"}\n print("✓ Identity Authentication successful")\n return True\n else:\n response.status_code = 401\n response.body = {"error": "Unauthorized access"}\n print("✗ Identity Authentication failed")\n return False\n\nclass LoggingFilter(Filter):\n """Logging Filter"""\n def handle(self, request: HttpRequest, response: HttpResponse) -> bool:\n print(f"Log Request: {request.method} {request.path}")\n return True # Always continue execution\n\nclass PermissionFilter(Filter):\n """Permission Check Filter"""\n def handle(self, request: HttpRequest, response: HttpResponse) -> bool:\n print("Check user permissions...")\n if request.user and request.user.get('role') == 'admin':\n print("✓ PermissionsCheck passed")\n return True\n else:\n response.status_code = 403\n response.body = {"error": "Insufficient permissions"}\n print("✗ PermissionsCheck failed")\n return False\nStep 3: Use the Filter Chain to Process Requests
\n\ndef process_http_request(request: HttpRequest) -> HttpResponse:\n """Process HTTP Request"""\n # Create default response\n response = HttpResponse(\n status_code=200,\n headers={"Content-Type": "application/json"},\n body={"message": "Request successful"}\n )\n \n # Create filter chain\n auth_filter = AuthenticationFilter()\n logging_filter = LoggingFilter()\n permission_filter = PermissionFilter()\n \n # Build Filter Chain: Logging → Authentication → Permissions\n logging_filter.set_next(auth_filter)\n auth_filter.set_next(permission_filter)\n \n # Execute Filter Chain\n print("Start Processing HTTP Request...")\n logging_filter.do_filter(request, response)\n print("Request processing completed")\n \n return response\n\n# Test Different Request Scenarios\ndef test_filter_chain():\n print("=== HTTP Filter Chain Test ===n")\n \n # Test Case 1: Valid Admin Request\n print("Test Case 1: Valid Admin Request")\n request1 = HttpRequest(\n method="GET",\n path="/admin/data",\n headers={"Authorization": "Bearer valid-token"}\n )\n response1 = process_http_request(request1)\n print(f"Response Status Code: {response1.status_code}")\n print(f"Response content: {response1.body}n")\n \n # Test Case 2: Unauthorized Request\n print("Test Case 2: Unauthorized Request")\n request2 = HttpRequest(\n method="GET",\n path="/admin/data",\n headers={"Authorization": "Bearer invalid-token"}\n )\n response2 = process_http_request(request2)\n print(f"Response Status Code: {response2.status_code}")\n print(f"Response content: {response2.body}n")\n\nif __name__ == "__main__":\n test_filter_chain()\n\n\n
Advantages and Disadvantages of the Chain of Responsibility Pattern
\n\nAdvantages
\n\n1. Reduced Coupling
\n\nThe request sender doesn't need to know which object will handle its request, and receivers don't need to know the full picture of the request.
\n\n2. Enhanced Flexibility
\n\nHandlers can be dynamically added, removed, or rearranged without affecting client code.
\n\n3. Simplified Object Responsibilities
\n\nEach handler only needs to focus on requests within its scope of responsibility, following the Single Responsibility Principle.
\n\n4. Easy to Extend
\n\nAdding new handlers is very easy—just implement the handler interface and add it to the chain.
\n\nDisadvantages
\n\n1. Requests May Go Unhandled
\n\nIf the chain of responsibility is improperly configured, a request may reach the end of the chain without being handled by any handler.
\n\n2. Performance Considerations
\n\nLonger chains of responsibility may impact performance, especially when processing large numbers of requests.
\n\n3. Difficult Debugging
\n\nThe processing path of requests may not be clear, making it difficult to track the complete processing flow during debugging.
\n\n\n\n
Practical Application Scenarios
\n\n1. Event Handling Systems
\n\nThe event bubbling mechanism in graphical interfaces is a typical application of the Chain of Responsibility pattern.
\n\nclass Event:\n def __init__(self, name, target):\n self.name = name\n self.target = target\n self.handled = False\n\nclass Widget:\n def __init__(self, parent=None):\n self.parent = parent\n \n def handle_event(self, event):\n # Handle it first; if unable, pass to the parent component\n if self.on_event(event):\n event.handled = True\n elif self.parent:\n self.parent.handle_event(event)\n \n def on_event(self, event):\n # Subclasses override this method to implement specific event handling\n return False\n2. Logging Systems
\n\nLog messages of different levels are handled by different handlers.
\n\nimport logging\n\n# Create Logger\nlogger = logging.getLogger('my_app')\nlogger.setLevel(logging.DEBUG)\n\n# Create Handler Chain\nconsole_handler = logging.StreamHandler()\nfile_handler = logging.FileHandler('app.log')\n\n# Set Handler Level\nconsole_handler.setLevel(logging.WARNIN