Understanding Chain of Responsability pattern

Introduction

On many occasions on our projects, we find ourselves writing pieces of code that stop entirely the program or to continue given a specific condition, it can be a validation of data that was sent, a query on the database, or a logic validation of our business rules, to solve each validation we tend to write simple ifs, in many cases it fits well but it also starts to generate a bigger problem that is having a bunch of ifs on the same place will make the code harder to debug and read.

The Problem

I'm going to use as an example an e-commerce that is selling a product with a discount on a specific date, the steps that we need to validate are:
Is the product still in stock?
Is the discount still valid (valid period)?
Does the discount value exceed the product price?

To solve this issue we usually write many ifs, like this:

def show_discount(product):
    if product.stock == 0:
        return f"there's no stock for {product}"

    discount = Discount.find(product.id)
    if discount.expire_at < today:
        return f"discount for {product} is expired"

    product_price = product.price - discount_amount.value
    if product_price <= 0:
        return "cannot apply discount, discount value is greater than the product price"

    return f"total with discount: {product_price}"

The code above looks fine, but as time pass we will need to change some rule or add new ones, the problems with this approach may include:

1. Coupling
2. Complexity
3. Organization

Now let's say that by accident someone applied a discount for a product that shouldn't be applied and from now on we need to create a mechanism to block those certain products to get a discount

blocked_brands = ("apple",)
def show_discount(product):
    if product.stock == 0:
        return f"there's no stock for {product}"

    if product.brand in blocked_brands:
        return "cannot apply discount for blocked brands"

    # + previous code...

Add a new if can solve the problem in the quickest and easiest way but it will start to create another problem of "code smell", our function is now even longer and more complex, this situation can make the code harder to maintain, but if the "simplest" solution is not the right solution, how can we improve/fix this situation?

Chain of Responsiability

Chain of Responsibility is a behavioral design pattern that lets you pass requests along a chain of handlers. Upon receiving a call, each handler decides either to process the request or to pass it to the next handler in the chain.

Benefits

We can control the order of the validation easier
Single responsibility principle
Easy to maintain
Open/Closed principle, we can add a new handler without changing the ones that already work

How to implement

Declare the handler interface and describe the signature of a method for handling requests. To remove duplicated code we can write abstract classes derived from the handler interface.

class Handler(ABC):
    @abstractmethod
    def set_next(self, handler: Handler) -> Handler:
        pass

    @abstractmethod
    def handle(self, request) -> Optional[str]:
        pass


class AbstractHandler(Handler):
    _next_handler: Handler = None

    def set_next(self, handler: Handler) -> Handler:
        self._next_handler = handler
        # Returning a handler from here will let us link handlers in a
        # convenient way:
        # product.set_next(brand).set_next(discount)
        return handler

    @abstractmethod
    def handle(self, product: Any) -> str:
        if self._next_handler:
            return self._next_handler.handle(product)
        return None

For each handler, we create a subclass handler that implements the interface methods. Each handler should make their own decisions separately.

class ProductHandler(AbstractHandler):
    def handle(product) -> str:
        if product.stock == 0:
            return "there's no stock for {product}"

        return super().handle(product)

class BrandHandler(AbstractHandler):
    _blocked_brands = ("apple",)

    def handle(product) -> str:
        if product.brand in self._blocked_brands:
            return f"{product.name} is not allowed to have discount, blocked by brand"

        return super().handle(product)

class DiscountHandler(AbstractHandler):
    def handle(product) -> str:
        discount = Discount.find(product.id)
        if not has_valid_period(discount):
            return "the discount is expired"

        if has_valid_amount(product.price, discount_amount.value):
            return "cannot apply discount, discount value is greater than the product price"

        return super().handle(product)

    def has_valid_period(discount):
        return discount.expires_at < datetime.now()

    def has_valid_amount(product_price, discount_value):
        return discount_value < product_price

What's happening?

Every class handler has an attribute that tells what is the next step

ProductHandler._next_handler will be BrandHandler 
BrandHandler._next_handler will be DiscountHandler 
DiscountHandler.next_handler will be None

Refactoring

Now that we've separated the responsibilities using classes that implement the interface we can change our code:

product_handler = ProductHandler()
brand_handler = BrandHandler()
discount_handler = DiscountHandler()

product_handler.set_next(brand_handler).set_next(discount_handler)

# product sample
product = Product(id=1, name="mobile phone", stock=0, brand="sansumg", price=1000)

print(product_handler.handle(product))

After the change, the code is simpler to change because every class has his own responsibility and is also more flexible because if we need to change the handler order or add a new one we just need to change the set_next order, we can also start the chain from any point

# will execute only BrandHandler and DiscountHandler
print(brand_handler.handle(product))

If we need to add a new handler we can simply create another class and add another set_next method.

class NotificationHandler(AbstractHandler):
    def handle(product) -> str:
        if product.stock <= 10 and product.price >= 1000:
            logger.info(f"{product.name} is running out of stock, current stock: {product.stock}")

        return super().handle(product)

product_handler = ProductHandler()
brand_handler = BrandHandler()
discount_handler = DiscountHandler()
notification_handler = NotificationHandler()

product_handler.set_next(
    brand_handler
).set_next(
    discount_handler
).set_next(
    notification_handler
)

References

https://refactoring.guru/pt-br/design-patterns/chain-of-responsibility