Маршрутизация, динамические параметры, class-based views, RouteTableDef, валидация данных через Pydantic.
Научитесь гибко настраивать маршруты: от простых 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. Это полезно при редиректах и в шаблонах:
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.
Для ресурса с несколькими методами (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__ для внедрения зависимостейНедостатки:
self.request вместо прямого request — чуть больше набора текстаКогда использовать: для CRUD-ресурсов с 3+ методами — class-based. Для простых эндпоинтов с 1-2 методами — отдельные функции.
Вместо пошагового добавления роутов можно собрать их в таблицу:
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() собирает их вместе.
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, а не NoneField(ge=1, le=5) отвергнет priority: 10"3" станет int 3 для поля priorityТипичная ошибка: не обрабатывать исключение Pydantic. При невалидных данных
TaskCreate(**raw_data)выброситValidationError. Без try/except это превратится в 500 Internal Server Error, хотя проблема на стороне клиента — нужен 400 Bad Request.
Для 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 и действуйте соответственно.
# Неправильный порядок — второй роут никогда не сработает
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+} (любые буквы+цифры), первый роут перехватил бы запрос.
await для request.json() в class-based viewclass TaskView(web.View):
async def post(self):
# Неправильно — забыл await
data = self.request.json() # coroutine object!
# Правильно
data = await self.request.json()# Неправильно — выбросит 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')add_getУбедитесь, что можете ответить:
add_get() отличается от add_route('GET', ...)?{id} только цифрами?ValidationError от Pydantic?Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.