Паттерн «Цепочка обязанностей» в Python

Паттерн Цепочка обязанностей — это поведенческий паттерн проектирования, который позволяет передавать запросы по цепочке обработчиков. Получив запрос, каждый обработчик может либо обработать его, либо передать следующему обработчику в цепочке.

Аналогия из реальной жизни

Представьте процесс утверждения отпуска в компании:

1-3 дня→ Руководитель команды одобряет
4-7 дней→ Руководитель отдела одобряет
Более 8 дней→ Генеральный директор одобряет

Когда вы отправляете заявку на отпуск, она последовательно передается по этой цепочке, пока не найдет человека, уполномоченного ее обработать. Это реальное проявление паттерна «Цепочка обязанностей».

Основная идея паттерна

Суть паттерна «Цепочка обязанностей» состоит вразделении отправителя и получателя запроса, что дает нескольким объектам возможность обработать запрос и, таким образом, позволяет избежать жесткой связи между отправителем и получателем.

Структура паттерна «Цепочка обязанностей»

Давайте разберем состав паттерна «Цепочка обязанностей» с помощью UML-диаграммы классов:

Описание основных компонентов

Handler (абстрактный обработчик)

Определяет интерфейс для обработки запросов
Содержит метод для установки преемника
Может реализовать поведение по умолчанию для цепочки преемников

ConcreteHandler (Конкретный обработчик)

Реализует конкретную логику обработки
Обрабатывает запрос, если может его обработать
В противном случае передаёт запрос преемнику

Базовая реализация: система утверждения отпусков

Давайте реализуем простую систему утверждения отпусков на Python, чтобы понять паттерн «Цепочка обязанностей».

Шаг 1: Определите абстрактный обработчик

Пример

from abc import ABC, abstractmethod
from typing import Optional

class Approver(ABC):
"""Абстрактный класс Approver"""

def __init__(self, name: str):
self.name = name
self.successor: Optional['Approver'] = None

def set_successor(self, successor: 'Approver'):
"""Установить следующего утверждающего"""
self.successor = successor

@abstractmethod
def process_request(self, leave_request: 'LeaveRequest'):
"""Абстрактный метод для обработки запросов на отпуск"""
pass

Шаг 2: Определите объект запроса

Пример

from dataclasses import dataclass

@dataclass
class LeaveRequest:
"""Класс данных запроса на отпуск"""
employee_name: str # Имя сотрудника
leave_days: int # Дни отпуска
reason: str # Причина отпуска

def __str__(self):
return f"{self.employee_name} запрашивает отпуск на {self.leave_days} дней, причина: {self.reason}"

Шаг 3: Реализуйте конкретные обработчики

Пример

class TeamLeader(Approver):
"""Руководитель команды — обрабатывает заявки на отпуск на 1-3 дня"""

def process_request(self, leave_request: LeaveRequest):
if leave_request.leave_days <= 3:
print(f"Руководитель команды {self.name} одобрил {leave_request}")
elif self.successor is not None:
# Превышает полномочия, передать на следующий уровень
print(f"Руководитель команды {self.name} не имеет полномочий, передача на более высокий уровень")
self.successor.process_request(leave_request)
else:
print("Невозможно обработать запрос: нет подходящего утверждающего")

class DepartmentManager(Approver):
"""Руководитель отдела — обрабатывает заявки на отпуск на 4-7 дней"""

def process_request(self, leave_request: LeaveRequest):
if 4 <= leave_request.leave_days <= 7:
print(f"Руководитель отдела {self.name} одобрил {leave_request}")
elif self.successor is not None:
# Превышает полномочия, передать на следующий уровень
print(f"Руководитель отдела {self.name} не имеет полномочий, передача на более высокий уровень")
self.successor.process_request(leave_request)
else:
print("Невозможно обработать запрос: нет подходящего утверждающего")

class GeneralManager(Approver):
"""Генеральный директор — обрабатывает заявки на отпуск продолжительностью более 8 дней"""

def process_request(self, leave_request: LeaveRequest):
if leave_request.leave_days >= 8:
print(f"Генеральный директор {self.name} одобрил {leave_request}")
else:
print("Невозможно обработать запрос: количество дней отпуска не соответствует требованиям")

Шаг 4: Построение цепочки и тестирование

Пример

def main():
# Создание утверждающих
team_leader = TeamLeader("Чжан Сань")
dept_manager = DepartmentManager("Ли Сы")
general_manager = GeneralManager("Ван У")

# Построение цепочки: руководитель команды → руководитель отдела → генеральный директор
team_leader.set_successor(dept_manager)
dept_manager.set_successor(general_manager)

# Создать тестовые случаи
test_cases = [
LeaveRequest("Сяо Мин", 2, "Простуда и жар"),
LeaveRequest("Сяо Хун", 5, "Поехать домой навестить семью"),
LeaveRequest("Сяо Ган", 10, "Свадебный отпуск"),
LeaveRequest("Сяо Ли", 15, "Поездка за границу")
]

print("=== Тест одобрения отпусков ==="\n")

# Обработать все запросы на отпуск
for request in test_cases:
print(f"Обработка запроса: {request.employee_name} запрашивает {request.leave_days} дней отпуска")
team_leader.process_request(request)
print("-" * 50)

if __name__ == "__main__":
main()

Вывод

=== 请假审批测试 ===

处理请求: 小明 请假 2 天
组长 张三 批准了 小明 申请请假 2 天，原因：感冒发烧
--------------------------------------------------
处理请求: 小红 请假 5 天
组长 张三 无权限，转交给上级处理
部门经理 李四 批准了 小红 申请请假 5 天，原因：回家探亲
--------------------------------------------------
处理请求: 小刚 请假 10 天
组长 张三 无权限，转交给上级处理
部门经理 李四 无权限，转交给上级处理
总经理 王五 批准了 小刚 申请请假 10 天，原因：结婚休假
--------------------------------------------------
处理请求: 小李 请假 15 天
组长 张三 无权限，转交给上级处理
部门经理 李四 无权限，转交给上级处理
总经理 王五 批准了 小李 申请请假 15 天，原因：出国旅游
--------------------------------------------------

Расширенное применение: Фильтр веб-запросов

Паттерн «Цепочка обязанностей» очень распространён в веб-разработке, особенно в middleware и цепочках фильтров. Давайте реализуем простую цепочку фильтров HTTP-запросов.

Шаг 1: Определите объекты запроса и ответа

Пример

from dataclasses import dataclass
from typing import Dict, Any

@dataclass
class HttpRequest:
"""Объект HTTP-запроса"""
method: str
path: str
headers: Dict[str, str]
body: Any = None
user: Dict[str, Any] = None

@dataclass
class HttpResponse:
"""Объект HTTP-ответа"""
status_code: int
headers: Dict[str, str]
body: Any = None

Шаг 2: Реализация цепочки фильтров

Пример

class Filter(ABC):
"""Абстрактный класс Filter"""

def __init__(self):
self.next_filter: Optional['Filter'] = None

def set_next(self, next_filter: 'Filter'):
"""Установить следующий фильтр"""
self.next_filter = next_filter

def do_filter(self, request: HttpRequest, response: HttpResponse) -> bool:
"""
Выполнить операцию фильтрации
Вернуть True для продолжения, False для прерывания
"""
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:
"""Конкретная логика фильтрации"""
pass

class AuthenticationFilter(Filter):
"""Фильтр аутентификации"""

def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
print("Выполняется аутентификация...")

# Имитация логики аутентификации
token = request.headers.get('Authorization')
if token == "Bearer valid-token":
request.user = {"id": 1, "name": "Zhang San", "role": "user"}
print("✓ Аутентификация прошла успешно")
return True
else:
response.status_code = 401
response.body = {"error": "Неавторизованный доступ"}
print("✗ Аутентификация не удалась")
return False

class LoggingFilter(Filter):
"""Фильтр логирования"""

def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
print(f"Логирование запроса: {request.method} {request.path}")
return True # Всегда продолжать

class PermissionFilter(Filter):
"""Фильтр проверки разрешений"""

def handle(self, request: HttpRequest, response: HttpResponse) -> bool:
print("Проверка разрешений пользователя...")

if request.user and request.user.get('role') == 'admin':
print("✓ Проверка прав доступа пройдена")
return True
else:
response.status_code = 403
response.body = {"error": "Недостаточно прав"}
print("✗ Проверка прав доступа не пройдена")
return False

Шаг 3: Используйте цепочку фильтров для обработки запросов

Пример

def process_http_request(request: HttpRequest) -> HttpResponse:
"""Обработать HTTP-запрос"""

# Создать ответ по умолчанию
response = HttpResponse(
status_code=200,
headers={"Content-Type": "application/json"},
body={"message": "Запрос выполнен успешно"}
)

# Создать цепочку фильтров
auth_filter = AuthenticationFilter()
logging_filter = LoggingFilter()
permission_filter = PermissionFilter()

# Построить цепочку фильтров: Логирование → Аутентификация → Права доступа
logging_filter.set_next(auth_filter)
auth_filter.set_next(permission_filter)

# Выполнить цепочку фильтров
print("Начало обработки HTTP-запроса...")
logging_filter.do_filter(request, response)
print("Обработка запроса завершена")

return response

# Тестирование различных сценариев запросов
def test_filter_chain():
print("=== Тест цепочки HTTP-фильтров ==="\n")

# Тестовый случай 1: Корректный запрос администратора
print("Тестовый случай 1: Корректный запрос администратора")
request1 = HttpRequest(
method="GET",
path="/admin/data",
headers={"Authorization": "Bearer valid-token"}
)
response1 = process_http_request(request1)
print(f"Код состояния ответа: {response1.status_code}")
print(f"Тело ответа: {response1.body}"\n")

# Тестовый случай 2: Неавторизованный запрос
print("Тестовый случай 2: Неавторизованный запрос")
request2 = HttpRequest(
method="GET",
path="/admin/data",
headers={"Authorization": "Bearer invalid-token"}
)
response2 = process_http_request(request2)
print(f"Код состояния ответа: {response2.status_code}")
print(f"Тело ответа: {response2.body}"\n")

if __name__ == "__main__":
test_filter_chain()

Плюсы и минусы паттерна «Цепочка обязанностей»

Преимущества

Недостатки

Отправителю запроса не нужно знать, какой объект будет обрабатывать его запрос, а получателю не нужно знать полную картину запроса.

2. Повышенная гибкость

Обработчики можно динамически добавлять, удалять или переупорядочивать, не влияя на клиентский код.

3. Упрощённые обязанности объектов

Каждому обработчику нужно сосредоточиться только на запросах в пределах своей зоны ответственности, придерживаясь принципа единственной ответственности.

4. Легко расширять

Добавить новый обработчик очень легко; достаточно реализовать интерфейс обработчика и добавить его в цепочку.

Недостатки

1. Запросы могут остаться необработанными

Если цепочка обязанностей настроена неправильно, запрос может достичь конца цепочки, так и не будучи обработан ни одним обработчиком.

2. Соображения производительности

Более длинная цепочка обязанностей может влиять на производительность, особенно при обработке большого количества запросов.

3. Сложная отладка

Путь обработки запроса может быть неясен, что затрудняет отслеживание полного потока обработки при отладке.

Реальные сценарии применения

1. Система обработки событий

Механизм всплытия событий в графических интерфейсах — типичный пример применения паттерна «Цепочка обязанностей».

Пример

class Event:
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):
# Сначала обработайте его самостоятельно; если его невозможно обработать, передайте его родительскому компоненту
if self.on_event(event):
event.handled = True
elif self.parent:
self.parent.handle_event(event)

def on_event(self, event):
# Подклассы переопределяют этот метод для реализации конкретной обработки событий
return False

2. Система логирования

Сообщения журнала разных уровней обрабатываются разными обработчиками.

Пример

import logging

# Создать логгер
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

# Создать цепочку обработчиков
console_handler = logging.StreamHandler()
file_handler = logging.FileHandler('app.log')

# Установить уровни обработчиков
console_handler.setLevel(logging.WARNING)
file_handler.setLevel(logging.DEBUG)

# Добавить в логгер
logger.addHandler(console_handler)
logger.addHandler(file_handler)

# Использование: уровни DEBUG и INFO будут записываться только в файл; WARNING и выше также будут выводиться в консоль.
logger.debug('Отладочная информация') # Записывается только в файл
logger.warning('Предупреждающее сообщение') # Выводится и в консоль, и в файл

Лучшие практики и рекомендации

1. Настройка поведения обработки по умолчанию

Убедитесь, что в конце цепочки обязанностей есть обработчик по умолчанию, чтобы избежать потерянных запросов.

Пример

class DefaultHandler(Approver):
"""Обработчик по умолчанию - обрабатывает все необработанные запросы"""

def process_request(self, leave_request: LeaveRequest):
print(f"Обработчик по умолчанию: не может обработать {leave_request}, пожалуйста, проверьте параметры запроса")

2. Контроль длины цепочки

Избегайте создания слишком длинной цепочки обязанностей, так как это снижает производительность и увеличивает сложность отладки.

3. Четкие правила обработки

Убедитесь, что каждый обработчик имеет четкую границу обработки, чтобы избежать пересечения обязанностей.

4. Рассмотрите использование паттерна Компоновщик

Для сложных цепочек обязанностей рассмотрите использование паттерна Компоновщик для управления связями между обработчиками.

Итог

Паттерн Цепочка обязанностей — это мощный и гибкий паттерн проектирования. Разделяя отправителя и получателя запроса, он предоставляет элегантный способ обработки сценариев, требующих взаимодействия нескольких объектов.

Ключевые моменты

Основная идея: Предоставить нескольким объектам возможность обработать запрос, избегая связывания между отправителем и получателем запроса.
Применимые сценарии: Несколько объектов могут обработать один и тот же запрос, но то, какой объект его обработает, должно определяться во время выполнения.
Моменты реализации: Определите интерфейс обработчика, постройте цепочку обработчиков и передайте запрос.
Возможности Python: Используя динамические возможности Python, паттерн Цепочка обязанностей можно реализовать более гибко.

Другие расширения