Расширяем Python-приложения с помощью хуков: гибкость без изменений

class HookManager:
def __init__(self):
self.hooks = {}

def register(self, hook_name, callback):
"""Регистрирует функцию-обработчик для указанного хука"""
if hook_name not in self.hooks:
self.hooks[hook_name] = []
self.hooks[hook_name].append(callback)

def trigger(self, hook_name, *args, **kwargs):
"""Вызывает все зарегистрированные обработчики для хука"""
if hook_name not in self.hooks:
return []

results = []
for callback in self.hooks[hook_name]:
results.append(callback(*args, **kwargs))
return results

# Создаем экземпляр менеджера хуков
hook_manager = HookManager()

# Определяем функции-обработчики
def log_user_login(user_id, timestamp):
print(f"User {user_id} logged in at {timestamp}")
return True

def check_user_permissions(user_id, timestamp):
# Здесь могла бы быть проверка прав доступа
print(f"Checking permissions for user {user_id}")
return user_id % 2 == 0 # Допустим, разрешаем вход только пользователям с четными ID

# Регистрируем обработчики для хука "user_login"
hook_manager.register("user_login", log_user_login)
hook_manager.register("user_login", check_user_permissions)

# В соответствующем месте приложения вызываем хук
def process_login(user_id):
import time
timestamp = time.time()

# Вызываем все обработчики хука user_login
results = hook_manager.trigger("user_login", user_id, timestamp)

# Проверяем результаты обработчиков
if all(results):
print(f"User {user_id} successfully logged in")
else:
print(f"Login denied for user {user_id}")

# Тестируем
process_login(1) # Должен вывести "Login denied for user 1"
process_login(2) # Должен вывести "User 2 successfully logged in"

class AdvancedHookManager:
def __init__(self):
self.hooks = {}

def register(self, hook_name, callback, priority=10):
"""Регистрирует функцию-обработчик с указанным приоритетом"""
if hook_name not in self.hooks:
self.hooks[hook_name] = []

self.hooks[hook_name].append((priority, callback))
# Сортируем обработчики по приоритету (от высшего к низшему)
self.hooks[hook_name].sort(key=lambda x: x[0], reverse=True)

def unregister(self, hook_name, callback):
"""Удаляет обработчик из списка"""
if hook_name not in self.hooks:
return

self.hooks[hook_name] = [(p, c) for p, c in self.hooks[hook_name] if c != callback]

def trigger(self, hook_name, *args, **kwargs):
"""Вызывает обработчики в порядке их приоритета"""
if hook_name not in self.hooks:
return []

results = []
for _, callback in self.hooks[hook_name]:
results.append(callback(*args, **kwargs))
return results

class EnhancedHookManager:
def __init__(self):
self.hooks = {}

def register(self, hook_name, callback, priority=10):
if hook_name not in self.hooks:
self.hooks[hook_name] = []

self.hooks[hook_name].append((priority, callback))
self.hooks[hook_name].sort(key=lambda x: x[0], reverse=True)

def trigger(self, hook_name, *args, **kwargs):
if hook_name not in self.hooks:
return None

result = None
for _, callback in self.hooks[hook_name]:
result = callback(*args, **kwargs)
# Если обработчик вернул False, прерываем цепочку
if result is False:
break
return result

class Component:
def __init__(self):
self._before_init_hooks = []
self._after_init_hooks = []
self._before_destroy_hooks = []

# Вызываем хуки перед инициализацией
self._trigger_hooks(self._before_init_hooks)
self._initialize()
# Вызываем хуки после инициализации
self._trigger_hooks(self._after_init_hooks)

def _initialize(self):
# Реальная инициализация компонента
pass

def add_before_init_hook(self, hook):
self._before_init_hooks.append(hook)

def add_after_init_hook(self, hook):
self._after_init_hooks.append(hook)

def add_before_destroy_hook(self, hook):
self._before_destroy_hooks.append(hook)

def destroy(self):
# Вызываем хуки перед уничтожением
self._trigger_hooks(self._before_destroy_hooks)
# Очищаем ресурсы

def _trigger_hooks(self, hooks):
for hook in hooks:
hook(self)

class EventEmitter:
def __init__(self):
self._event_handlers = {}

def on(self, event_name, handler):
"""Подписка на событие"""
if event_name not in self._event_handlers:
self._event_handlers[event_name] = []
self._event_handlers[event_name].append(handler)
return self # Для цепочки вызовов

def off(self, event_name, handler=None):
"""Отписка от события"""
if event_name not in self._event_handlers:
return self

if handler is None:
del self._event_handlers[event_name]
else:
self._event_handlers[event_name] = [
h for h in self._event_handlers[event_name] if h != handler
]
return self

def emit(self, event_name, *args, **kwargs):
"""Генерация события"""
if event_name not in self._event_handlers:
return False

for handler in self._event_handlers[event_name]:
handler(*args, **kwargs)
return True

# Пример использования
class UserManager(EventEmitter):
def __init__(self):
super().__init__()
self.users = {}

def create_user(self, user_id, name):
self.emit('before_create_user', user_id, name)

if user_id in self.users:
raise ValueError(f"User {user_id} already exists")

self.users[user_id] = {'name': name}

self.emit('after_create_user', user_id, self.users[user_id])
return self.users[user_id]

# Использование
user_manager = UserManager()
user_manager.on('before_create_user', lambda user_id, name: print(f"About to create user {name}"))
user_manager.on('after_create_user', lambda user_id, user: print(f"Created user {user['name']}"))

user_manager.create_user(1, "John Doe")

def hook_decorator(pre_hooks=None, post_hooks=None):
pre_hooks = pre_hooks or []
post_hooks = post_hooks or []

def decorator(func):
def wrapper(*args, **kwargs):
# Выполняем pre-хуки
for hook in pre_hooks:
hook_result = hook(*args, **kwargs)
# Можно модифицировать args/kwargs на основе результата хука
if isinstance(hook_result, tuple) and len(hook_result) == 2:
args, kwargs = hook_result

# Выполняем оригинальную функцию
result = func(*args, **kwargs)

# Выполняем post-хуки
for hook in post_hooks:
hook_result = hook(result, *args, **kwargs)
# Возможность модифицировать результат
if hook_result is not None:
result = hook_result

return result

return wrapper

return decorator

# Пример использования
def log_args(func_name):
def hook(*args, **kwargs):
print(f"Calling {func_name} with args: {args}, kwargs: {kwargs}")
return args, kwargs # Возвращаем неизмененные аргументы
return hook

def log_result(func_name):
def hook(result, *args, **kwargs):
print(f"{func_name} returned: {result}")
return result # Возвращаем неизмененный результат
return hook

@hook_decorator(
pre_hooks=[log_args("add")],
post_hooks=[log_result("add")]
)
def add(a, b):
return a + b

result = add(2, 3) # Выведет "Calling add with args..." и "add returned: 5"

class MiddlewareManager:
def __init__(self):
self.middlewares = []

def add_middleware(self, middleware):
self.middlewares.append(middleware)

def process_request(self, request):
for middleware in self.middlewares:
if hasattr(middleware, 'process_request'):
request = middleware.process_request(request)
return request

def process_response(self, response):
# Обрабатываем в обратном порядке
for middleware in reversed(self.middlewares):
if hasattr(middleware, 'process_response'):
response = middleware.process_response(response)
return response

# Пример промежуточного ПО
class LoggingMiddleware:
def process_request(self, request):
print(f"Processing request: {request}")
return request

def process_response(self, response):
print(f"Processing response: {response}")
return response

class AuthenticationMiddleware:
def process_request(self, request):
if 'auth_token' not in request:
request['authenticated'] = False
else:
# Проверка токена
request['authenticated'] = True
return request

# Использование
manager = MiddlewareManager()
manager.add_middleware(LoggingMiddleware())
manager.add_middleware(AuthenticationMiddleware())

request = {'url': '/api/users', 'auth_token': '12345'}
processed_request = manager.process_request(request)
# Обработка запроса...
response = {'status': 200, 'data': ['user1', 'user2']}
processed_response = manager.process_response(response)

import asyncio

class AsyncHookManager:
def __init__(self):
self.hooks = {}

def register(self, hook_name, callback):
if hook_name not in self.hooks:
self.hooks[hook_name] = []
self.hooks[hook_name].append(callback)

async def trigger(self, hook_name, *args, **kwargs):
"""Асинхронно запускает все обработчики"""
if hook_name not in self.hooks:
return []

# Запускаем все обработчики параллельно
tasks = [
asyncio.create_task(callback(*args, **kwargs)) 
for callback in self.hooks[hook_name]
]

# Ждем завершения всех задач
results = await asyncio.gather(*tasks, return_exceptions=True)

# Фильтруем исключения
processed_results = []
for result in results:
if isinstance(result, Exception):
print(f"Hook exception: {result}")
else:
processed_results.append(result)

return processed_results

# Пример использования
async def main():
hook_manager = AsyncHookManager()

# Регистрация асинхронных обработчиков
async def log_operation(operation_id):
await asyncio.sleep(0.1) # Имитация I/O операции
print(f"Операция {operation_id} зарегистрирована")
return True

async def validate_operation(operation_id):
await asyncio.sleep(0.2)
print(f"Операция {operation_id} проверена")
return operation_id % 2 == 0 # Четные – валидные

hook_manager.register("before_operation", log_operation)
hook_manager.register("before_operation", validate_operation)

# Вызов хуков
results = await hook_manager.trigger("before_operation", 42)

if all(results):
print("Все проверки пройдены")
else:
print("Операция отклонена")

# asyncio.run(main())

class FilterHookManager:
def __init__(self):
self.filter_hooks = {}

def add_filter(self, filter_name, callback, priority=10):
if filter_name not in self.filter_hooks:
self.filter_hooks[filter_name] = []

self.filter_hooks[filter_name].append((priority, callback))
self.filter_hooks[filter_name].sort(key=lambda x: x[0])

def apply_filters(self, filter_name, value, *args, **kwargs):
"""Последовательно применяет все фильтры к значению"""
if filter_name not in self.filter_hooks:
return value

# Каждый фильтр получает результат предыдущего
current_value = value
for _, callback in self.filter_hooks[filter_name]:
current_value = callback(current_value, *args, **kwargs)

return current_value

# Пример использования
filter_manager = FilterHookManager()

# Добавляем фильтры для обработки текста
def sanitize_html(content):
# Удаляем потенциально опасные теги
return content.replace("<script>", "&lt;script&gt;")

def add_copyright(content):
return content + "\n© 2023 Company Name"

def format_paragraphs(content):
# Оборачиваем текст в параграфы, если он еще не в них
if not content.strip().startswith("<p>"):
paragraphs = content.split("\n\n")
return "".join([f"<p>{p}</p>" for p in paragraphs if p.strip()])
return content

# Регистрируем фильтры с разными приоритетами
filter_manager.add_filter("content_filter", sanitize_html, 5) # Сначала безопасность
filter_manager.add_filter("content_filter", format_paragraphs, 10) # Потом форматирование
filter_manager.add_filter("content_filter", add_copyright, 15) # В конце копирайт

# Применяем фильтры
raw_content = "Привет, мир!\n\n<script>alert('XSS');</script>\n\nЕще один абзац."
processed_content = filter_manager.apply_filters("content_filter", raw_content)
print(processed_content)

class ContextualHookManager:
def __init__(self):
self.hooks = {}

def register(self, hook_name, callback):
if hook_name not in self.hooks:
self.hooks[hook_name] = []
self.hooks[hook_name].append(callback)

def trigger(self, hook_name, context, *args, **kwargs):
"""Вызывает хуки с передачей контекста выполнения"""
if hook_name not in self.hooks:
return []

results = []
for callback in self.hooks[hook_name]:
# Передаем контекст и аргументы
results.append(callback(context, *args, **kwargs))
return results

# Пример контекста и обработчика
class RequestContext:
def __init__(self, user=None, request_data=None):
self.user = user
self.request_data = request_data or {}
self.response_data = {}
self.errors = []

def add_error(self, error):
self.errors.append(error)

def is_valid(self):
return len(self.errors) == 0

# Пример использования
hook_manager = ContextualHookManager()

def validate_permissions(context, resource_id, action):
if not context.user:
context.add_error("Требуется авторизация")
return False

# Проверка прав доступа
if action == "delete" and context.user["role"] != "admin":
context.add_error("Недостаточно прав для удаления")
return False

return True

def log_action(context, resource_id, action):
print(f"Пользователь {context.user['id'] if context.user else 'anonymous'} "
f"пытается выполнить действие {action} для ресурса {resource_id}")
return True

# Регистрируем обработчики
hook_manager.register("pre_action", validate_permissions)
hook_manager.register("pre_action", log_action)

# Использование
def perform_action(user, resource_id, action):
context = RequestContext(user=user)

# Проверяем, можно ли выполнить действие
results = hook_manager.trigger("pre_action", context, resource_id, action)

if all(results) and context.is_valid():
print(f"Выполняем действие {action} для ресурса {resource_id}")
return True
else:
print(f"Действие отклонено: {context.errors}")
return False

# Тестирование
admin_user = {"id": 1, "role": "admin"}
regular_user = {"id": 2, "role": "user"}

perform_action(admin_user, 42, "delete") # Разрешено
perform_action(regular_user, 42, "delete") # Отклонено

class CompositeHookManager:
def __init__(self):
self.hooks = {}
self.hook_groups = {}

def register(self, hook_name, callback, group=None):
if hook_name not in self.hooks:
self.hooks[hook_name] = []

self.hooks[hook_name].append(callback)

if group:
if group not in self.hook_groups:
self.hook_groups[group] = set()
self.hook_groups[group].add(hook_name)

def trigger_group(self, group_name, *args, **kwargs):
"""Вызывает все хуки в группе"""
if group_name not in self.hook_groups:
return {}

results = {}
for hook_name in self.hook_groups[group_name]:
hook_results = self.trigger(hook_name, *args, **kwargs)
results[hook_name] = hook_results

return results

def trigger(self, hook_name, *args, **kwargs):
"""Вызывает конкретный хук"""
if hook_name not in self.hooks:
return []

results = []
for callback in self.hooks[hook_name]:
results.append(callback(*args, **kwargs))

return results

# Пример динамического определения хуков на основе конфигурации
class ConfigurableHookSystem:
def __init__(self, hook_manager):
self.hook_manager = hook_manager
self.config = {}

def load_config(self, config):
"""Загружает конфигурацию хуков из словаря"""
self.config = config

# Регистрируем хуки на основе конфигурации
for hook_name, handlers in config.get('hooks', {}).items():
for handler_info in handlers:
module_name = handler_info['module']
function_name = handler_info['function']
group = handler_info.get('group')

# Динамически импортируем обработчик
module = __import__(module_name, fromlist=[function_name])
handler = getattr(module, function_name)

# Регистрируем обработчик
self.hook_manager.register(hook_name, handler, group)

def trigger_hook(self, hook_name, *args, **kwargs):
"""Вызывает хук, если он включен в конфигурации"""
if not self.is_hook_enabled(hook_name):
return []

return self.hook_manager.trigger(hook_name, *args, **kwargs)

def is_hook_enabled(self, hook_name):
"""Проверяет, включен ли хук в конфигурации"""
hook_config = self.config.get('hooks', {}).get(hook_name, [])
return bool(hook_config) and self.config.get('enabled', True)

import os
import importlib.util

class CLIPluginManager:
def __init__(self, plugins_dir="plugins"):
self.plugins_dir = plugins_dir
self.commands = {}
self.pre_execute_hooks = []
self.post_execute_hooks = []

# Загружаем плагины при инициализации
self.load_plugins()

def load_plugins(self):
"""Загружает все плагины из директории"""
if not os.path.exists(self.plugins_dir):
os.makedirs(self.plugins_dir)
return

for filename in os.listdir(self.plugins_dir):
if filename.endswith('.py') and not filename.startswith('_'):
self.load_plugin(filename)

def load_plugin(self, filename):
"""Загружает отдельный плагин"""
plugin_path = os.path.join(self.plugins_dir, filename)
plugin_name = os.path.splitext(filename)[0]

# Загружаем модуль плагина
spec = importlib.util.spec_from_file_location(plugin_name, plugin_path)
plugin_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(plugin_module)

# Регистрируем команды и хуки
if hasattr(plugin_module, 'register'):
plugin_module.register(self)

def register_command(self, command_name, handler):
"""Регистрирует новую команду"""
self.commands[command_name] = handler

def register_pre_execute_hook(self, hook):
"""Регистрирует хук, выполняемый перед командой"""
self.pre_execute_hooks.append(hook)

def register_post_execute_hook(self, hook):
"""Регистрирует хук, выполняемый после команды"""
self.post_execute_hooks.append(hook)

def execute(self, command_name, *args, **kwargs):
"""Выполняет команду с учетом хуков"""
if command_name not in self.commands:
print(f"Ошибка: команда '{command_name}' не найдена")
return False

# Контекст выполнения
context = {
'command': command_name,
'args': args,
'kwargs': kwargs,
'abort': False
}

# Вызываем pre-execute хуки
for hook in self.pre_execute_hooks:
hook(context)
if context['abort']:
print(f"Выполнение команды '{command_name}' отменено хуком")
return False

# Выполняем команду
result = self.commands[command_name](*args, **kwargs)

# Обновляем контекст результатом
context['result'] = result

# Вызываем post-execute хуки
for hook in self.post_execute_hooks:
hook(context)

return result

import time

def log_command(context):
"""Логирует информацию о вызываемой команде"""
command = context['command']
args = context['args']
kwargs = context['kwargs']
print(f"[LOG] Выполняется команда '{command}' с аргументами {args} и {kwargs}")

def log_result(context):
"""Логирует результат выполнения команды"""
command = context['command']
result = context.get('result')
print(f"[LOG] Команда '{command}' вернула результат: {result}")

def register(plugin_manager):
"""Регистрирует хуки в менеджере плагинов"""
plugin_manager.register_pre_execute_hook(log_command)
plugin_manager.register_post_execute_hook(log_result)

# Также можно зарегистрировать команды
plugin_manager.register_command('version', lambda: "Logger Plugin v1.0")

# Создаем менеджер плагинов
manager = CLIPluginManager()

# Определяем базовые команды
def add(a, b):
return a + b

def subtract(a, b):
return a – b

# Регистрируем команды
manager.register_command('add', add)
manager.register_command('subtract', subtract)

# Выполнение команд (с автоматическим вызовом хуков)
result = manager.execute('add', 5, 3)
print(f"Результат: {result}")

result = manager.execute('version')
print(f"Результат: {result}")

class Model:
_hooks = {
'before_save': [],
'after_save': [],
'before_delete': [],
'after_delete': []
}

@classmethod
def register_hook(cls, hook_name, callback):
"""Регистрирует хук для всех моделей этого класса"""
if hook_name in cls._hooks:
cls._hooks[hook_name].append(callback)

def __init__(self, **kwargs):
self.id = None
self.data = kwargs

def save(self):
"""Сохраняет модель в хранилище"""
# Вызываем хуки before_save
for hook in self.__class__._hooks['before_save']:
hook(self)

# В реальном ORM здесь был бы код сохранения в БД
print(f"Сохраняем модель {self.__class__.__name__} с данными: {self.data}")
if not self.id:
self.id = 1 # В реальности был бы автоинкремент или UUID

# Вызываем хуки after_save
for hook in self.__class__._hooks['after_save']:
hook(self)

return self

def delete(self):
"""Удаляет модель из хранилища"""
# Вызываем хуки before_delete
for hook in self.__class__._hooks['before_delete']:
hook(self)

# В реальном ORM здесь был бы код удаления из БД
print(f"Удаляем модель {self.__class__.__name__} с ID: {self.id}")

# Вызываем хуки after_delete
for hook in self.__class__._hooks['after_delete']:
hook(self)

# Создаем конкретную модель
class User(Model):
def __init__(self, username, email, **kwargs):
super().__init__(**kwargs)
self.data['username'] = username
self.data['email'] = email

# Определяем хуки
def validate_email(model):
"""Проверяет формат email перед сохранением"""
if 'email' in model.data and '@' not in model.data['email']:
raise ValueError(f"Некорректный email: {model.data['email']}")

def timestamp_creation(model):
"""Добавляет временную метку создания"""
import time
if not model.id: # Новая запись
model.data['created_at'] = time.time()

def log_deletion(model):
"""Логирует удаление записи"""
print(f"[AUDIT] Удалена запись {model.__class__.__name__} с ID {model.id}")

# Регистрируем хуки для модели User
User.register_hook('before_save', validate_email)
User.register_hook('before_save', timestamp_creation)
User.register_hook('after_delete', log_deletion)

# Пример использования
try:
# Создание с валидным email
user1 = User(username="john_doe", email="john@example.com")
user1.save()

# Создание с невалидным email вызовет исключение
user2 = User(username="bad_user", email="invalid-email")
user2.save()
except ValueError as e:
print(f"Ошибка: {e}")

# Удаление (вызовет хук log_deletion)
user1.delete()

class Request:
def __init__(self, path, method, headers=None, body=None):
self.path = path
self.method = method
self.headers = headers or {}
self.body = body
self.params = {}
self.user = None

class Response:
def __init__(self, body="", status=200, headers=None):
self.body = body
self.status = status
self.headers = headers or {}

class WebFramework:
def __init__(self):
self.routes = {}
self.middleware = []
self.hooks = {
'before_request': [],
'after_request': [],
'before_response': [],
'on_error': []
}

def register_hook(self, hook_name, callback):
"""Регистрирует хук для обработки запросов"""
if hook_name in self.hooks:
self.hooks[hook_name].append(callback)

def add_middleware(self, middleware):
"""Добавляет middleware для обработки запросов/ответов"""
self.middleware.append(middleware)

def route(self, path, method='GET'):
"""Декоратор для регистрации обработчиков маршрутов"""
def decorator(handler):
self.routes[(path, method)] = handler
return handler
retur