Chain of Responsibility

The Chain of Responsibility lets you pass requests along a chain of handlers; each handler can decide whether to process the request or pass it on. This way, it ensures that the sender of a request is not coupled to the receiver, since the request is passed on until an object chooses to handle it.

TL;DR

Allows a request to be passed along different handlers until there is one that deals with the request. Often used if you will deal with many requests, but you don’t know what they look like beforehand.

Problem

Suppose you are designing a payment processing connector that can connect several accounts to pay for an item. The functionality is design such that different banks can have different priorities, but payments must be done in full.

If you don’t know how much a specific purchase will be, you will need different checks to ensure that it can be processed. For example, lets say you have Bank1(100), Bank2(200), and Bank3(300) (meaning how much money there is), and that is the same order of priority. If a purchase is made for $290, the first two cannot pay, so the request must arrive at Bank3.

A naive solution is to add if/else statements and complex logic, but this quickly becomes unmaintainable and the priorities would then be hard-coded into the application.

Solution

The solution is to use CoR, where we instantiate each bank as a concrete handler that can either make the payment or pass the request on to the next bank.

The main part of this pattern is in the abstract handler class, which implements set_next() and also provides a default implementation of handle(). The former is what allows each handler to know what the next step is in case of failure, while the handle() method will return the next instance of a handler when needed (see Builder and Linked Lists).

Note

This pattern could be combined with Builder by creating a SuportChainBuilder

from abc import ABC, abstractmethod
 
# Abstract handler class
class SupportHandler(ABC):
    def __init__(self):
        self._next_handler = None
    
    def set_next(self, handler):
        """Set the next handler in the chain"""
        self._next_handler = handler
        return handler  # Allow chaining: handler1.set_next(handler2).set_next(handler3)
    
    @abstractmethod
    def handle(self, request):
        """Handle the request or pass it to the next handler"""
        if self._next_handler:
            return self._next_handler.handle(request)
        return None
 
# Concrete handlers
class Level1Support(SupportHandler):
    def handle(self, request):
        if request.priority <= 1:
            return f"Level 1 Support: Handled '{request.issue}' (Priority {request.priority})"
        else:
            print(f"Level 1 Support: Cannot handle priority {request.priority}, passing to next level")
            return super().handle(request)
 
class Level2Support(SupportHandler):
    def handle(self, request):
        if request.priority <= 2:
            return f"Level 2 Support: Handled '{request.issue}' (Priority {request.priority})"
        else:
            print(f"Level 2 Support: Cannot handle priority {request.priority}, passing to next level")
            return super().handle(request)
 
class Level3Support(SupportHandler):
    def handle(self, request):
        if request.priority <= 3:
            return f"Level 3 Support: Handled '{request.issue}' (Priority {request.priority})"
        else:
            print(f"Level 3 Support: Cannot handle priority {request.priority}, passing to next level")
            return super().handle(request)
 
# Request class
class SupportRequest:
    def __init__(self, issue, priority):
        self.issue = issue
        self.priority = priority
 
# Example usage
if __name__ == "__main__":
    # Create handlers
    level1 = Level1Support()
    level2 = Level2Support()
    level3 = Level3Support()
    
    # Build the chain
    level1.set_next(level2).set_next(level3)
    
    # Create test requests
    requests = [
        SupportRequest("Password reset", 1),
        SupportRequest("Software installation", 2),
        SupportRequest("Server crash", 3),
        SupportRequest("Data center fire", 4)
    ]
    
    # Process requests through the chain
    for request in requests:
        print(f"\nProcessing: {request.issue} (Priority {request.priority})")
        result = level1.handle(request)
        if result:
            print(f"Result: {result}")
        else:
            print("No handler could process this request")
        print("-" * 50)