Request/Response объекты, статус-коды, типы ответов, обработка ошибок, жизненный цикл запроса в aiohttp.
Разберёмся, как aiohttp превращает TCP-соединение в объект Request и как из вашего кода получается HTTP-ответ.
Прежде чем писать код, важно понять путь запроса от клиента до вашего обработчика и обратно:
Клиент (браузер/curl)
│
▼
TCP-соединение (принято сервером)
│
▼
HTTP-парсер aiohttp (разбирает байты на метод, путь, заголовки, тело)
│
▼
Router (находит handler по методу + пути)
│
▼
Middleware chain (оборачивают запрос)
│
▼
Ваш handler(request) → web.Response
│
▼
HTTP-сериализация (статус, заголовки, тело → байты)
│
▼
TCP-ответ клиенту
Каждый шаг прозрачен для вас как разработчика, но понимание этого пути помогает отлаживать ошибки. Например, если вы получаете 404 Not Found — роутер не нашёл matching роут. Если 405 Method Not Allowed — роут есть, но для другого HTTP-метода.
Минимальный aiohttp-сервер состоит из приложения и одного обработчика:
from aiohttp import web
async def hello(request: web.Request) -> web.Response:
return web.Response(text="Hello, World!")
app = web.Application()
app.router.add_get('/', hello)
if __name__ == '__main__':
web.run_app(app, host='localhost', port=8080)Что делает web.run_app():
on_startup хуки перед приёмом запросовВажно:
web.run_app()предназначен только для разработки. В продакшене используйте Gunicorn. Подробнее — в теме про деплой.
web.run_app(
app,
host='0.0.0.0', # Слушать все сетевые интерфейсы
port=8080,
backlog=128, # Максимум ожидающих TCP-подключений в очереди
reuse_address=True, # Разрешить повторное использование адреса
handle_signals=True # Обрабатывать SIGINT/SIGTERM
)backlog — размер очереди ожидающих подключений. Если сервер обрабатывает запросы медленно и очередь заполняется, новые клиенты получат Connection refused. Значение по умолчанию (128) достаточно для большинства случаев.
reuse_address — позволяет перезапустить сервер на том же порту без ожидания TIME_WAIT. Полезно при частых перезапусках во время разработки.
Когда клиент отправляет HTTP-запрос, aiohttp создаёт объект web.Request и передаёт его обработчику. Request — это «конверт», содержащий всё, что отправил клиент.
async def handler(request: web.Request) -> web.Response:
# HTTP-метод
method = request.method # 'GET', 'POST', 'PUT', 'DELETE'
# Путь и URL
path = request.path # '/api/users'
query_string = request.query_string # 'page=1&sort=name'
# Query-параметры (после ?)
page = request.query.get('page', '1') # Значение по умолчанию — '1'
sort = request.query.get('sort') # None если нет параметра
# Заголовки
content_type = request.content_type # 'application/json'
auth = request.headers.get('Authorization') # 'Bearer token...'
# Тело запроса (для POST/PUT/PATCH)
data = await request.json() # Парсинг JSON из тела
text = await request.text() # Сырой текст тела
form = await request.post() # FormData (multipart/form-data)
# IP-адрес клиента
client_ip = request.remote # '127.0.0.1'
# Доступ к приложению
db_pool = request.app['db']
return web.Response(text="OK")Это два разных источника данных, которые новички часто путают:
request.query — параметры из URL после ?: ?page=1&sort=name. Всегда строкиrequest.match_info — параметры из шаблона роута: /users/{id}. Тоже всегда строки# Роут: /users/{id}
# Запрос: GET /users/42?page=1
async def handler(request):
request.match_info['id'] # '42' — из пути
request.query.get('page') # '1' — из query stringОба значения — строки. Если нужен int, преобразуйте явно: int(request.match_info['id']).
async def create_user(request: web.Request) -> web.Response:
# Читаем JSON из тела
data = await request.json()
username = data.get('username')
email = data.get('email')
if not username or not email:
raise web.HTTPBadRequest(
text='Username и email обязательны'
)
# Сохранение в БД (псевдокод)
# user = await db.users.create(username, email)
return web.json_response(
{'id': 1, 'username': username, 'email': email},
status=201
)Типичная ошибка: вызвать
request.json()дважды. Тело запроса читается один раз — поток уже прочитан. Второй вызов вернёт пустой результат или ошибку. Если нужно проверить тело несколько раз, сохраните результат в переменную.
aiohttp предоставляет несколько способов вернуть ответ. Выбор зависит от типа данных.
return web.Response(
text='Hello, World!',
status=200,
headers={'X-Custom-Header': 'my-value'}
)Используется для простых текстовых ответов. text автоматически кодируется в UTF-8.
return web.json_response(
{'status': 'ok', 'data': {'id': 1}},
status=200,
headers={'X-Total-Count': '42'}
)json_response автоматически:
json.dumps()Content-Type: application/json; charset=utf-8Под капотом:
json_responseиспользует стандартныйjsonмодуль Python. Для больших ответов это может быть медленно — в теме про оптимизацию рассмотримujson.
async def download_binary(request: web.Request) -> web.Response:
data = b'\x89PNG\r\n\x1a\n...' # Бинарные данные
return web.Response(
body=data,
content_type='application/octet-stream',
headers={
'Content-Disposition': 'attachment; filename="data.bin"'
}
)Используйте body (не text) для бинарных данных. text ожидает строку и будет кодировать её, что испортит бинарные данные.
async def download_file(request: web.Request) -> web.Response:
return web.FileResponse(
'/path/to/report.pdf',
headers={
'Content-Disposition': 'attachment; filename="report.pdf"'
}
)FileResponse эффективно отдаёт файлы — не загружает весь файл в память, а читает чанками. Полезно для больших файлов.
async def delete_item(request: web.Request) -> web.Response:
# Успешное удаление — без тела ответа
return web.Response(status=204)Статус 204 (No Content) означает «успех, но тела ответа нет». Используется для DELETE-операций.
Код состояния — это трёхзначное число, которое сервер возвращает клиенту. Оно говорит: «что случилось с запросом».
| Код | Значение | Когда использовать |
|---|---|---|
| 200 | OK | Успешный GET/PUT/PATCH |
| 201 | Created | Ресурс создан после POST |
| 204 | No Content | Успешное удаление (DELETE) |
| 301 | Moved Permanently | Ресурс перемещён навсегда |
| 302 | Found | Временный редирект |
| 400 | Bad Request | Ошибка в данных запроса (невалидный JSON, отсутствие поля) |
| 401 | Unauthorized | Не предоставлена аутентификация |
| 403 | Forbidden | Аутентифицирован, но нет прав |
| 404 | Not Found | Ресурс не существует |
| 405 | Method Not Allowed | Роут есть, но для другого метода |
| 409 | Conflict | Конфликт (дубликат, одновременное изменение) |
| 422 | Unprocessable Entity | Ошибка валидации (Pydantic) |
| 500 | Internal Server Error | Неожиданная ошибка сервера |
Вместо создания Response вручную, можно выбросить исключение. aiohttp перехватит его и сформирует правильный ответ:
async def get_task(request: web.Request) -> web.Response:
task_id = int(request.match_info['id'])
task = await find_task(task_id)
if not task:
raise web.HTTPNotFound(text='Task not found')
return web.json_response(task)Это удобнее, чем return web.Response(status=404, text='...'), потому что:
returnraise web.HTTPFound('/new-location')HTTPFound (302) говорит клиенту: «ресурс теперь тут». Браузер автоматически перейдёт по новому URL.
Соберём простые обработчики для нашего приложения. Обратите внимание: каждый handler — async def, принимает request и возвращает Response.
from aiohttp import web
from typing import List
# Временное хранилище (в реальности — база данных)
tasks: List[dict] = []
next_id: int = 1
async def list_tasks(request: web.Request) -> web.Response:
"""GET /tasks — список всех задач"""
return web.json_response({'tasks': tasks})
async def get_task(request: web.Request) -> web.Response:
"""GET /tasks/{id} — получить одну задачу"""
task_id = int(request.match_info['id'])
for task in tasks:
if task['id'] == task_id:
return web.json_response(task)
raise web.HTTPNotFound(text=f'Task {task_id} not found')
async def create_task(request: web.Request) -> web.Response:
"""POST /tasks — создать задачу"""
global next_id
data = await request.json()
if not data.get('title'):
raise web.HTTPBadRequest(text='Title is required')
task = {
'id': next_id,
'title': data['title'],
'description': data.get('description', ''),
'completed': False
}
tasks.append(task)
next_id += 1
return web.json_response(task, status=201)
async def delete_task(request: web.Request) -> web.Response:
"""DELETE /tasks/{id} — удалить задачу"""
task_id = int(request.match_info['id'])
for i, task in enumerate(tasks):
if task['id'] == task_id:
tasks.pop(i)
return web.Response(status=204)
raise web.HTTPNotFound(text=f'Task {task_id} not found')
def create_app() -> web.Application:
app = web.Application()
app.router.add_get('/tasks', list_tasks)
app.router.add_get('/tasks/{id}', get_task)
app.router.add_post('/tasks', create_task)
app.router.add_delete('/tasks/{id}', delete_task)
return app
if __name__ == '__main__':
web.run_app(create_app(), port=8080)# Создать задачу
curl -X POST http://localhost:8080/tasks \
-H "Content-Type: application/json" \
-d '{"title": "Learn aiohttp"}'
# Получить список
curl http://localhost:8080/tasks
# Получить одну задачу
curl http://localhost:8080/tasks/1
# Удалить задачу
curl -X DELETE http://localhost:8080/tasks/1text вместо body для бинарных данных# Неправильно — text кодирует строку в UTF-8, бинарные данные испортятся
return web.Response(text=b'\x89PNG') # TypeError
# Правильно — body принимает байты напрямую
return web.Response(body=b'\x89PNG', content_type='image/png')await для request.json()# Неправильно — request.json() возвращает корутину
data = request.json() # <coroutine object>
title = data['title'] # TypeError!
# Правильно
data = await request.json()
title = data['title']# Неправильно — 200 означает «всё ок», но ресурс уже существовал
return web.json_response(task, status=200)
# Правильно — 201 означает «ресурс создан»
return web.json_response(task, status=201)Убедитесь, что можете ответить:
web.Response(status=204) вместо web.json_response()?request.query.get('page', '1'), если клиент передал ?page=abc?request.match_info отличается от request.query?Далее: Роутинг и обработчики запросов