Промежуточное ПО: логирование, аутентификация, CORS, обработка ошибок. Сигналы жизненного цикла приложения.
Middleware — это конвейер, через который проходит каждый запрос. Разберёмся, как строить этот конвейер и когда использовать сигналы.
Middleware (промежуточное ПО) — функция, которая оборачивает обработчик запроса. Она выполняется до вызова handler'а и после получения ответа.
Представьте матрёшку: каждый middleware — это слой, внутри которого находится следующий слой или сам handler.
Запрос → [Middleware 1] → [Middleware 2] → [Handler]
Ответ ← [Middleware 1] ← [Middleware 2] ← [Handler]
Типичные задачи middleware:
@web.middlewarefrom aiohttp import web
@web.middleware
async def log_middleware(
request: web.Request,
handler
) -> web.Response:
"""Логирование каждого запроса"""
print(f"[{request.method}] {request.path}")
# Вызываем следующий middleware или handler
response = await handler(request)
print(f"[{response.status}] {request.path}")
return response
# Регистрация в приложении
app = web.Application(middlewares=[log_middleware])Middleware принимает два аргумента:
request — объект запросаhandler — функция следующего middleware или handler'аMiddleware обязан вернуть web.Response. Он может:
handler (обычный случай)raise web.HTTPUnauthorized)Middleware выполняются как стек (LIFO для обратного прохода):
@web.middleware
async def m1(request, handler):
print("M1: ДО")
response = await handler(request)
print("M1: ПОСЛЕ")
return response
@web.middleware
async def m2(request, handler):
print("M2: ДО")
response = await handler(request)
print("M2: ПОСЛЕ")
return response
app = web.Application(middlewares=[m1, m2])
app.router.add_get('/', hello)При запросе вывод будет:
M1: ДО
M2: ДО
Handler выполняется
M2: ПОСЛЕ
M1: ПОСЛЕ
Ключевое правило: middleware выполняются в порядке объявления на пути к handler'у и в обратном порядке на пути обратно. Это позволяет «оборачивать» запрос и ответ: первый middleware видит запрос первым и ответ последним.
Практичный пример: логирование с замером времени выполнения.
import logging
import time
from aiohttp import web
logger = logging.getLogger(__name__)
@web.middleware
async def logging_middleware(
request: web.Request,
handler
) -> web.Response:
start_time = time.perf_counter()
# Логирование запроса
logger.info(
f"{request.method} {request.path} "
f"from {request.remote}"
)
try:
response = await handler(request)
except Exception:
# Логирование ошибки (и пробрасываем дальше)
logger.exception(f"Error handling {request.path}")
raise
# Логирование ответа
duration_ms = (time.perf_counter() - start_time) * 1000
logger.info(
f"{response.status} {request.path} "
f"completed in {duration_ms:.2f}ms"
)
return responseОбратите внимание: try/except логирует ошибку, но не перехватывает её — raise пробрасывает исключение дальше. Обработкой ошибок займётся другой middleware.
from aiohttp import web
# Пути, не требующие аутентификации
PUBLIC_PATHS = frozenset({
'/api/v1/auth/login',
'/api/v1/auth/register',
'/health',
})
@web.middleware
async def auth_middleware(
request: web.Request,
handler
) -> web.Response:
# Публичные роуты — без проверки
if request.path in PUBLIC_PATHS:
return await handler(request)
# Проверяем заголовок Authorization
auth_header = request.headers.get('Authorization')
if not auth_header:
raise web.HTTPUnauthorized(text='Authorization header required')
parts = auth_header.split()
if len(parts) != 2 or parts[0] != 'Bearer':
raise web.HTTPUnauthorized(text='Invalid authorization header')
token = parts[1]
# Валидация токена (псевдокод — будет в теме про auth)
# payload = decode_jwt(token)
# if not payload:
# raise web.HTTPUnauthorized(text='Invalid token')
# Добавляем пользователя в request — handler получит его
request['user'] = {'id': 1, 'email': 'user@example.com'}
return await handler(request)Под капотом:
request['user']работает, потому чтоweb.Requestподдерживает доступ по ключу как dict. Это стандартный способ передавать данные между middleware и handler'ом. Middleware кладёт данные, handler их читает.
Браузеры отправляют предварительный OPTIONS-запрос (preflight) перед кросс-доменными запросами с нестандартными заголовками. Middleware должен обработать оба случая:
from aiohttp import web
@web.middleware
async def cors_middleware(
request: web.Request,
handler
) -> web.Response:
# Preflight-запрос — отвечаем сразу, не вызывая handler
if request.method == 'OPTIONS':
return web.Response(
headers={
'Access-Control-Allow-Origin': '*',
'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
'Access-Control-Allow-Headers': 'Content-Type, Authorization',
'Access-Control-Max-Age': '86400',
}
)
# Обычный запрос — вызываем handler и добавляем заголовок к ответу
response = await handler(request)
response.headers['Access-Control-Allow-Origin'] = '*'
return responseЦентрализованная обработка ошибок — один из самых полезных паттернов middleware:
from aiohttp import web
import logging
logger = logging.getLogger(__name__)
@web.middleware
async def error_handler_middleware(
request: web.Request,
handler
) -> web.Response:
try:
return await handler(request)
except web.HTTPException:
# HTTP-исключения уже содержат правильный статус — пробрасываем
raise
except ValueError as e:
logger.warning(f"ValueError on {request.path}: {e}")
raise web.HTTPBadRequest(text=str(e))
except Exception as e:
logger.exception(f"Unexpected error on {request.path}: {e}")
raise web.HTTPInternalServerError(text='Internal server error')Порядок middleware здесь критичен: error_handler должен быть последним в списке (ближе всего к handler'у), чтобы перехватить ошибки от всех предыдущих middleware и handler'а.
app = web.Application(middlewares=[
logging_middleware, # 1-й: видит запрос первым, ответ последним
cors_middleware, # 2-й: добавляет CORS-заголовки
auth_middleware, # 3-й: проверяет токен
error_handler_middleware, # 4-й: перехватывает ошибки от handler'а
])Иногда middleware нужно настроить — например, задать список публичных роутов или лимит запросов. Для этого используется фабрика: функция, возвращающая middleware.
from aiohttp import web
from typing import AbstractSet
def auth_middleware_factory(
public_paths: AbstractSet[str] = frozenset()
):
"""Фабрика: создаёт middleware с настройками"""
@web.middleware
async def middleware(
request: web.Request,
handler
) -> web.Response:
if request.path in public_paths:
return await handler(request)
# ... проверка токена ...
return await handler(request)
return middleware
# Использование
app = web.Application(middlewares=[
auth_middleware_factory(
public_paths=frozenset({'/login', '/register', '/health'})
)
])Фабрика — это обычная функция. Она создаёт замыкание, которое запоминает public_paths и использует их при каждом запросе.
Middleware часто нуждается в общих ресурсах: пул БД, Redis, конфигурация. Они хранятся в request.app:
@web.middleware
async def db_session_middleware(
request: web.Request,
handler
) -> web.Response:
db_pool = request.app['db']
async with db_pool.acquire() as conn:
request['db'] = conn # Передаём соединение handler'у
return await handler(request)Сигналы — это не middleware. Они не обрабатывают запросы, а вызываются один раз при определённых событиях жизни приложения.
Вызывается один раз при старте, до начала приёма запросов:
async def init_db(app: web.Application) -> None:
app['db'] = await asyncpg.create_pool(
'postgresql://user:pass@localhost/task_manager',
min_size=5,
max_size=20
)
print("Database connected")
app.on_startup.append(init_db)Вызывается при получении сигнала остановки (SIGINT/SIGTERM), до on_cleanup:
async def notify_clients(app: web.Application) -> None:
"""Уведомить WebSocket-клиентов о перезапуске"""
ws_manager = app.get('ws_manager')
if ws_manager:
await ws_manager.broadcast({'type': 'server_shutdown'})
app.on_shutdown.append(notify_clients)Вызывается после on_shutdown, для освобождения ресурсов:
async def close_db(app: web.Application) -> None:
await app['db'].close()
print("Database closed")
app.on_cleanup.append(close_db)SIGINT/SIGTERM
↓
Сервер перестаёт принимать новые запросы
↓
on_shutdown (уведомить клиентов, завершить транзакции)
↓
on_cleanup (закрыть соединения, освободить ресурсы)
↓
Event loop закрывается
def create_app() -> web.Application:
app = web.Application()
# Инициализация ресурсов
app.on_startup.append(init_db)
app.on_startup.append(init_redis)
# Очистка ресурсов (обратный порядок — хорошая практика)
app.on_cleanup.append(close_redis)
app.on_cleanup.append(close_db)
# Middleware (порядок важен!)
app.middlewares.append(logging_middleware)
app.middlewares.append(cors_middleware)
app.middlewares.append(auth_middleware)
app.middlewares.append(error_handler_middleware)
# Роуты
app.router.add_get('/health', health_check)
app.router.add_get('/tasks', list_tasks)
return appreturn await handler(request)@web.middleware
async def broken_middleware(request, handler):
print("Запрос обработан")
# Забыл вернуть response — handler не вызовется, клиент повиснет
# return await handler(request) <-- забыл!
# Правильно
@web.middleware
async def working_middleware(request, handler):
print("Запрос обработан")
return await handler(request)# Неправильно — ошибка перехвачена и потеряна
@web.middleware
async def bad_middleware(request, handler):
try:
return await handler(request)
except Exception:
return web.Response(status=500) # Без логирования!
# Правильно — логируем и пробрасываем
@web.middleware
async def good_middleware(request, handler):
try:
return await handler(request)
except Exception:
logger.exception("Error in handler")
raise # Пробрасываем дальше# Неправильно — error_handler стоит первым, но не перехватит ошибки auth
app = web.Application(middlewares=[
error_handler_middleware, # Стоит ближе к запросу — не перехватит
auth_middleware, # Ошибки auth не будут обработаны
])
# Правильно — error_handler ближе к handler'у
app = web.Application(middlewares=[
auth_middleware, # Обрабатывает запрос первым
error_handler_middleware, # Перехватывает ошибки от auth и handler'а
])Убедитесь, что можете ответить:
error_handler_middleware должен быть последним в списке?Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.