Requests в Python: GET, POST, JSON и тайм-ауты

Библиотека Requests отправляет HTTP-запросы из Python. requests.get() читает ресурс, requests.post(..., json=...) отправляет JSON, а response.json() разбирает ответ. В рабочем вызове сразу задавай timeout и проверяй HTTP-статус: корректный JSON ещё не означает, что операция прошла успешно.

Клиентские примеры обращаются к локальному учебному API. Внешний сервис и API-ключ не нужны. Сетевые блоки запускаются в терминале; отдельная самопроверка структуры ответа в конце доступна прямо на странице.

Установка Requests в окружение проекта

В терминале проекта создай окружение и установи версию, на которой проверен пример:

python -m venv .venv

Активируй его в Linux или macOS:

source .venv/bin/activate

В Windows PowerShell используется другая команда:

.venv\Scripts\Activate.ps1

После активации команды одинаковы:

python -m pip install requests==2.33.1
python -c "import requests; print(requests.__version__)"

Ожидаемая версия — 2.33.1. Если Python установлен под именем python3, используй его в команде создания окружения. Привязка python -m pip устанавливает пакет для того же интерпретатора; остальные вопросы активации и выбора Python разобраны в статье про venv.

Локальный API для повторения примеров

Сохрани следующий код в файл demo_api.py. Он хранит две задачи в памяти, читает фильтр done, принимает новую задачу и предоставляет два специальных адреса для проверки ошибок: /slow и /empty.

import json
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlsplit

tasks = [
    {"id": 1, "title": "Прочитать условие", "done": True},
    {"id": 2, "title": "Проверить решение", "done": False},
]

class Handler(BaseHTTPRequestHandler):
    def reply(self, status, payload):
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        try:
            self.wfile.write(body)
        except BrokenPipeError:
            print("Клиент закрыл соединение до ответа")

    def do_GET(self):
        url = urlsplit(self.path)
        if url.path == "/slow":
            time.sleep(0.2)
            self.reply(200, {"ready": True})
            return
        if url.path == "/empty":
            self.send_response(204)
            self.end_headers()
            return
        if url.path != "/tasks":
            self.reply(404, {"error": "Маршрут не найден"})
            return
        done = parse_qs(url.query).get("done", [None])[0]
        if done not in (None, "true", "false"):
            self.reply(400, {"error": "done должен быть true или false"})
            return
        result = tasks if done is None else [
            task for task in tasks if task["done"] == (done == "true")
        ]
        self.reply(200, result)

    def do_POST(self):
        if urlsplit(self.path).path != "/tasks":
            self.reply(404, {"error": "Маршрут не найден"})
            return
        try:
            length = int(self.headers.get("Content-Length", "0"))
            if not 0 < length <= 4096:
                raise ValueError("недопустимый размер тела")
            if self.headers.get_content_type() != "application/json":
                raise ValueError("нужен Content-Type application/json")
            data = json.loads(self.rfile.read(length))
            title = data.get("title") if isinstance(data, dict) else None
            if not isinstance(title, str) or not 1 <= len(title.strip()) <= 200:
                raise ValueError("title должен содержать от 1 до 200 символов")
        except ValueError as error:
            self.reply(400, {"error": str(error)})
            return
        task = {"id": len(tasks) + 1, "title": title.strip(), "done": False}
        tasks.append(task)
        self.reply(201, task)

if __name__ == "__main__":
    with HTTPServer(("127.0.0.1", 48463), Handler) as server:
        print("Учебный API: http://127.0.0.1:48463")
        try:
            server.serve_forever()
        except KeyboardInterrupt:
            print("API остановлен")

Запусти сервер в первом терминале:

python demo_api.py

Оставь его работающим. Во втором терминале активируй то же окружение и запускай клиентские блоки ниже: каждый можно сохранить в отдельный .py-файл и вызвать через python имя_файла.py.

Сервер принимает подключения только с твоего компьютера, обрабатывает запросы последовательно и теряет добавленные задачи после остановки. Это учебная модель; модуль http.server не предназначен для промышленного сервера. Если порт 48463 занят, выбери свободный и замени его в сервере и клиентских URL. Чтобы завершить работу и освободить порт, нажми Ctrl+C в терминале сервера.

GET-запрос: параметры, статус и JSON ответа

Получим только незавершённые задачи. Фильтр передаётся через params; Requests добавит его в строку запроса:

import requests

response = requests.get(
    "http://127.0.0.1:48463/tasks",
    params={"done": "false"},
    headers={"Accept": "application/json"},
    timeout=(3, 10),
)
response.raise_for_status()
tasks = response.json()

print(response.url)
print(response.status_code)
print(tasks)
assert tasks == [{"id": 2, "title": "Проверить решение", "done": False}]

http://127.0.0.1:48463/tasks?done=false
200
[{'id': 2, 'title': 'Проверить решение', 'done': False}]

Такой вывод получается сразу после запуска сервера, до создания новой задачи. В params мы передали строку "false", потому что именно такое значение принимает учебный API. Автоматическая подстановка Python-значения False дала бы другую строку.

Accept сообщает, какой формат ответа ожидает клиент. Адрес, параметры и заголовки — разные части HTTP-запроса к API. Параметры params и json описаны в Quickstart Requests.

raise_for_status() не подтверждает конкретный код 200: например, 201 и 204 тоже проходят проверку. response.json() разбирает синтаксис, но не проверяет нужные поля. Пустое тело ответа 204 не является JSON. Поведение методов закреплено в API Requests; формат данных отдельно разобран в статье о JSON.

Python / 01koddo

Статус и JSON проверяются отдельно

POST-запрос с JSON-телом

Создадим новую задачу. В json= передаётся обычный словарь, а не заранее сериализованная строка:

import requests

response = requests.post(
    "http://127.0.0.1:48463/tasks",
    json={"title": "Разобрать ошибку"},
    timeout=(3, 10),
)
response.raise_for_status()
created = response.json()

print(response.status_code)
print(created)
assert response.status_code == 201
assert created["title"] == "Разобрать ошибку"
assert created["done"] is False

201
{'id': 3, 'title': 'Разобрать ошибку', 'done': False}

При повторном запуске будет создана ещё одна задача с новым id: у этого API каждый POST создаёт запись. Перезапуск сервера возвращает исходное состояние.

json=payload сериализует данные и задаёт Content-Type: application/json. data={...} отправляет форму, а params={...} дописывает параметры к URL даже у POST. Для нашего обработчика форма не подходит: он вернёт 400. Различие показано в руководстве Requests по телу запроса.

Проверить отказ можно, заменив тело на json={"title": " "}: сервер не добавит задачу, вернёт 400, и raise_for_status() поднимет исключение. Подробности значений кодов есть в справочнике HTTP-статусов.

Timeout: сколько Requests ждёт ответ

Requests не задаёт тайм-аут по умолчанию. В timeout=(3, 10) первое число ограничивает ожидание соединения, второе — ожидание данных при чтении. Это не общий дедлайн в 13 секунд: сервер может присылать тело частями, а подключение к разным IP-адресам может требовать отдельных попыток. Точные границы описаны в документации Requests о тайм-аутах.

Для проверки учебный /slow ждёт 0,2 секунды перед ответом. Дадим клиенту только 0,05 секунды на чтение:

import requests

try:
    requests.get("http://127.0.0.1:48463/slow", timeout=(3, 0.05))
except requests.exceptions.ReadTimeout:
    print("Сервер не ответил за интервал чтения")
else:
    raise AssertionError("Ожидался ReadTimeout")

Сервер не ответил за интервал чтения

Сервер может затем написать, что клиент закрыл соединение: он закончил паузу и попытался отправить ответ уже ушедшему клиенту. Это ожидаемая ветка примера.

Числа 3 и 10 — настройки учебного клиента. Для своего API выбирай их по допустимому времени ожидания. После тайм-аута POST не повторяй запрос вслепую: сервер мог создать задачу, а клиент не получил ответ. Нужны проверка результата или поддерживаемый сервером механизм защиты от повторного создания.

Как различать сетевую ошибку, HTTPError и плохой JSON

Один клиент обработает успешный ответ, отсутствующий маршрут и ответ без тела:

import requests

for path in ("/tasks", "/missing", "/empty"):
    try:
        response = requests.get(
            f"http://127.0.0.1:48463{path}", timeout=(3, 10)
        )
        response.raise_for_status()
        payload = response.json()
    except requests.exceptions.Timeout:
        print(path, "тайм-аут соединения или чтения")
    except requests.exceptions.ConnectionError:
        print(path, "не удалось подключиться")
    except requests.exceptions.HTTPError as error:
        print(path, "HTTP", error.response.status_code)
    except requests.exceptions.JSONDecodeError:
        print(path, "тело ответа не содержит JSON")
    except requests.exceptions.RequestException as error:
        print(path, type(error).__name__)
    else:
        print(path, "JSON получен:", type(payload).__name__)

/tasks JSON получен: list
/missing HTTP 404
/empty тело ответа не содержит JSON

После остановки сервера тот же клиент попадёт в ветку ConnectionError. /empty — успешный ответ 204 без тела, поэтому проверка HTTP пройдёт, а разбор JSON завершится ошибкой. Если конкретный API обещает 204, обрабатывай этот статус как завершённую операцию и не вызывай .json().

Специальные исключения стоят перед общим RequestException, иначе общая ветка перехватит их раньше. Иерархия исключений приведена в API Requests; порядок обработчиков разобран в статье про try/except.

JSON разобран: проверяем полезные данные

Даже синтаксически правильный JSON может содержать {"error": "..."} вместо списка или строку "false" вместо логического значения. Функция completed_titles проверяет список задач и типы полей title и done, затем возвращает названия завершённых задач:

def completed_titles(payload):
    if not isinstance(payload, list):
        raise ValueError("ожидался список задач")
    titles = []
    for task in payload:
        if not isinstance(task, dict):
            raise ValueError("задача должна быть объектом")
        if not isinstance(task.get("title"), str):
            raise ValueError("title должен быть строкой")
        if type(task.get("done")) is not bool:
            raise ValueError("done должен быть true или false")
        if task["done"]:
            titles.append(task["title"])
    return titles

payload = [
    {"id": 1, "title": "Прочитать условие", "done": True},
    {"id": 2, "title": "Проверить решение", "done": False},
]
assert completed_titles(payload) == ["Прочитать условие"]
assert completed_titles([]) == []

for invalid in ({"error": "нет доступа"}, [{"title": "Тест", "done": "false"}]):
    try:
        completed_titles(invalid)
    except ValueError:
        continue
    raise AssertionError("Ожидалась ошибка структуры ответа")

print(completed_titles(payload))

['Прочитать условие']

В клиентском файле после определения этой функции ей можно передать response.json(). Для закрепления добавь проверку отсутствующего title, затем запроси /tasks?done=true у локального сервера и сравни результат с ожидаемым названием.

Для серии запросов к одному API пригодится requests.Session: она сохраняет cookies и настройки и использует пул соединений. Requests выполняет сетевые вызовы синхронно; если код уже работает в asyncio, прямой вызов заблокирует цикл событий.