Роутинг и обработчики запросов

Научитесь гибко настраивать маршруты: от простых URL до динамических параметров, class-based views и валидации через Pydantic.

#Основы маршрутизации

Роутинг — это связь между HTTP-запросом (метод + путь) и вашей функцией-обработчиком. В aiohttp за это отвечает app.router.

#Методы добавления роутов

from aiohttp import web

async def list_users(request: web.Request) -> web.Response:
    return web.json_response({'users': []})

async def create_user(request: web.Request) -> web.Response:
    data = await request.json()
    return web.json_response(data, status=201)

app = web.Application()

# Краткие методы для популярных HTTP-методов
app.router.add_get('/users', list_users)
app.router.add_post('/users', create_user)

# Универсальный метод — подходит для любого HTTP-метода
app.router.add_route('PATCH', '/users/{id}', update_user)

Краткие методы (add_get, add_post, add_put, add_delete) — это обёртки над add_route. Они удобнее для чтения, но add_route нужен для нестандартных методов (PATCH, HEAD, OPTIONS).

Под капотом: add_get('/users', handler) — это просто сокращение для add_route('GET', '/users', handler). Никакой разницы в производительности нет.

#Динамические параметры

Статические роуты (/users) полезны только для списков. Для работы с конкретным ресурсом нужны параметры в URL.

#Именованные параметры

# Роут с параметром {id}
app.router.add_get('/users/{id}', get_user)

async def get_user(request: web.Request) -> web.Response:
    # Извлекаем параметр из match_info
    user_id = request.match_info['id']  # Всегда строка!

    # Конвертируем в int, если нужно число
    user_id_int = int(user_id)

    # ... логика поиска пользователя
    return web.json_response({'id': user_id_int})

Запросы:

GET /users/42 → match_info = {'id': '42'}
GET /users/abc → match_info = {'id': 'abc'} (не ошибка роутинга!)

Типичная ошибка: забыть конвертировать match_info в нужный тип. request.match_info['id'] всегда строка. Если вы ожидаете число и используете его в сравнении if user_id > 100, сравнение будет строковым, а не числовым: '9' > '100' вернёт True.

#Несколько параметров

app.router.add_get('/users/{user_id}/tasks/{task_id}', get_user_task)

async def get_user_task(request: web.Request) -> web.Response:
    user_id = request.match_info['user_id']
    task_id = request.match_info['task_id']
    # ...

#Ограничение через регулярные выражения

Можно ограничить, какие значения принимает параметр:

# Только цифры для ID
app.router.add_get(
    r'/users/{id:\d+}',  # \d+ = одна или более цифр
    get_user
)

# UUID-формат
app.router.add_get(
    r'/items/{item_id:[0-9a-f-]+}',
    get_item
)

С префиксом r'' строка становится «сырой» — не нужно экранировать обратный слэш. Без r'' пришлось бы писать '\\d+'.

Если параметр не соответствует паттерну, роут не совпадёт и aiohttp вернёт 404:

GET /users/42 → совпало
GET /users/abc → 404 (не подходит под \d+)

#«Жадный» параметр — захват всего пути

Иногда нужно захватить весь остаток URL, включая слеши:

app.router.add_get(r'/files/{filepath:.*}', serve_file)

async def serve_file(request: web.Request) -> web.Response:
    filepath = request.match_info['filepath']
    # GET /files/docs/report.pdf → filepath = 'docs/report.pdf'
    # GET /files/a/b/c.txt  → filepath = 'a/b/c.txt'

Паттерн .* означает «ноль или более любых символов», включая /.

#Именованные роуты и генерация URL

Роутам можно дать имя, а затем генерировать по ним URL. Это полезно при редиректах и в шаблонах:

app.router.add_get(
    '/users/{id}',
    get_user,
    name='user-detail'  # Имя роута
)

# Генерация URL в обработчике
async def create_and_redirect(request: web.Request) -> web.Response:
    user_id = 42
    # Генерируем /users/42
    url = request.app.router['user-detail'].url_for(
        id=str(user_id)
    )
    raise web.HTTPFound(str(url))

Под капотом: url_for() принимает строки для всех параметров. Даже если роут ожидает число ({id:\d+}), передавать нужно str(42), а не 42.

#Class-based views

Для ресурса с несколькими методами (GET, POST, PUT, DELETE) удобно объединить обработчики в класс:

class UserView(web.View):
    async def get(self) -> web.Response:
        user_id = self.request.match_info['id']
        # ... логика GET
        return web.json_response({'id': user_id})

    async def post(self) -> web.Response:
        data = await self.request.json()
        # ... логика POST
        return web.json_response(data, status=201)

    async def delete(self) -> web.Response:
        user_id = self.request.match_info['id']
        # ... логика DELETE
        return web.Response(status=204)

# Регистрация: все методы на одном пути
app.router.add_route('*', '/users/{id}', UserView)

Преимущества class-based views:

Все обработчики ресурса в одном месте
Общие методы (например, _get_user()) доступны через self
Можно переопределить __init__ для внедрения зависимостей

Недостатки:

Более verbose, чем отдельные функции
self.request вместо прямого request — чуть больше набора текста

Когда использовать: для CRUD-ресурсов с 3+ методами — class-based. Для простых эндпоинтов с 1-2 методами — отдельные функции.

#RouteTableDef — декларативный стиль

Вместо пошагового добавления роутов можно собрать их в таблицу:

from aiohttp import web

routes = web.RouteTableDef()

@routes.get('/tasks')
async def list_tasks(request: web.Request) -> web.Response:
    return web.json_response({'tasks': []})

@routes.post('/tasks')
async def create_task(request: web.Request) -> web.Response:
    data = await request.json()
    return web.json_response(data, status=201)

@routes.get('/tasks/{id}')
async def get_task(request: web.Request) -> web.Response:
    task_id = request.match_info['id']
    # ...
    return web.json_response({'id': task_id})

# Регистрация всех роутов одной строкой
app = web.Application()
app.router.add_routes(routes)

Это удобно для группировки: каждый модуль определяет свои роуты через декораторы, а create_app() собирает их вместе.

#Валидация данных запроса через Pydantic

aiohttp не имеет встроенной валидации — вы получаете сырой JSON через request.json(). Для валидации используем Pydantic:

from pydantic import BaseModel, Field

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: str = Field(default='', max_length=1000)
    priority: int = Field(default=1, ge=1, le=5)

async def create_task(request: web.Request) -> web.Response:
    # Читаем сырой JSON
    raw_data = await request.json()

    # Валидируем через Pydantic
    try:
        task_data = TaskCreate(**raw_data)
    except Exception as e:
        raise web.HTTPBadRequest(text=str(e))

    # task_data.title — гарантированно строка 1-200 символов
    # task_data.priority — гарантированно int от 1 до 5

    # ... сохранение в БД
    return web.json_response(
        task_data.model_dump(),
        status=201
    )

Что даёт Pydantic:

Типобезопасность: task_data.title — точно str, а не None
Валидация границ: Field(ge=1, le=5) отвергнет priority: 10
Автоконвертация: строка "3" станет int 3 для поля priority

Типичная ошибка: не обрабатывать исключение Pydantic. При невалидных данных TaskCreate(**raw_data) выбросит ValidationError. Без try/except это превратится в 500 Internal Server Error, хотя проблема на стороне клиента — нужен 400 Bad Request.

#Обработка любого HTTP-метода

Для webhook-ов и прокси может понадобиться обработать любой метод на одном пути:

async def webhook_handler(request: web.Request) -> web.Response:
    method = request.method  # 'POST', 'GET', 'PUT' ...

    if method == 'POST':
        data = await request.json()
        # обработка POST
    elif method == 'GET':
        # проверка статуса
        return web.json_response({'status': 'ok'})
    else:
        raise web.HTTPMethodNotAllowed(
            text=f'Method {method} not supported'
        )

app.router.add_route('*', '/webhook', webhook_handler)

'*' означает «любой HTTP-метод». Внутри обработчика проверяйте request.method и действуйте соответственно.

#Типичные ошибки

#1. Порядок роутов имеет значение

# Неправильный порядок — второй роут никогда не сработает
app.router.add_get(r'/users/{id:\d+}', get_user)
app.router.add_get('/users/admin', get_admin_panel)  # Никогда не вызовется!

# Правильно — специфичные роуты раньше общих
app.router.add_get('/users/admin', get_admin_panel)
app.router.add_get(r'/users/{id:\d+}', get_user)

Роутер проверяет роуты в порядке добавления. /users/admin совпадает с паттерном {id:\d+}? Нет, потому что admin не цифры. Но если бы паттерн был {id:\w+} (любые буквы+цифры), первый роут перехватил бы запрос.

#2. Забытый `await` для `request.json()` в class-based view

class TaskView(web.View):
    async def post(self):
        # Неправильно — забыл await
        data = self.request.json()  # coroutine object!

        # Правильно
        data = await self.request.json()

#3. Конвертация match_info в int без обработки ошибок

# Неправильно — выбросит 500 при нечисловом id
async def handler(request):
    task_id = int(request.match_info['id'])  # ValueError для 'abc'

# Правильно
async def handler(request):
    try:
        task_id = int(request.match_info['id'])
    except ValueError:
        raise web.HTTPBadRequest(text='Invalid task ID')

#Связь с другими темами

Middleware (следующая тема) — добавите промежуточную обработку: логирование каждого запроса, CORS, auth
REST API (тема 8) — примените роутинг для полноценного Task Manager API с репозиториями
WebSocket — роутинг для WebSocket-подключений через add_get

#Проверка понимания

Убедитесь, что можете ответить:

Чем add_get() отличается от add_route('GET', ...)?
Как ограничить параметр {id} только цифрами?
Когда лучше использовать class-based view вместо отдельных функций?
Что произойдёт, если не обработать ValidationError от Pydantic?