HTTP, REST и клиенты requests/httpx

⏱ 45 мин · 🧩 заданий: 4

Запрос и ответ

HTTP — текстовый протокол «запрос — ответ» поверх TCP (в HTTP/2 и HTTP/3 формат бинарный, но смысл тот же). Клиент отправляет запрос, сервер отвечает. Сервер не помнит предыдущих запросов (stateless): всё нужное — в самом запросе (токен, cookie).

POST /api/v1/orders?notify=true HTTP/1.1
Host: shop.example.com
Content-Type: application/json
Authorization: Bearer eyJhbGciOi...
Content-Length: 39

{"product_id": 42, "quantity": 2}
HTTP/1.1 201 Created
Content-Type: application/json
Location: /api/v1/orders/1001

{"id": 1001, "status": "new", "total": 1998}

Стартовая строка (метод, путь, версия / версия, код, фраза) → заголовки Имя: значение → пустая строка → тело. Его можно разобрать руками:

raw = (
    "POST /api/v1/orders?notify=true HTTP/1.1\r\n"
    "Host: shop.example.com\r\n"
    "Content-Type: application/json\r\n"
    "\r\n"
    '{"product_id": 42, "quantity": 2}'
)
head, _, body = raw.partition("\r\n\r\n")
start, *header_lines = head.split("\r\n")
method, target, version = start.split()
headers = {k.lower(): v.strip() for k, v in (line.split(":", 1) for line in header_lines)}
print(method, target, version)
print(headers)
print(body)

Методы

Метод Смысл Безопасный Идемпотентный Тело
GET получить ✔ ✔ нет
POST создать / действие да
PUT заменить целиком ✔ да
PATCH изменить частично да
DELETE удалить ✔ обычно нет
HEAD, OPTIONS заголовки / возможности (CORS) ✔ ✔ нет

Коды состояния

from http import HTTPStatus

for code in (200, 201, 204, 301, 304, 400, 401, 403, 404, 409, 422, 429, 500, 502, 503, 504):
    s = HTTPStatus(code)
    print(code, s.phrase)
  • 2xx — успех: 200 OK, 201 Created (+ заголовок Location), 204 No Content.
  • 3xx — перенаправление: 301/308 навсегда, 302/307 временно, 304 Not Modified (кеш актуален).
  • 4xx — ошибка клиента: 400 неверный запрос, 401 не аутентифицирован («кто вы?»), 403 нет прав («знаю, кто вы, но нельзя»), 404 нет ресурса, 409 конфликт, 422 не прошла валидация, 429 слишком много запросов.
  • 5xx — ошибка сервера: 500 баг, 502 прокси не достучался до приложения, 503 перегрузка или обслуживание, 504 таймаут на прокси.

Заголовки, которые нужно знать

Заголовок Зачем
Content-Type формат тела: application/json, text/html; charset=utf-8, multipart/form-data
Accept какой формат хочет клиент
Authorization Bearer <токен> или Basic <base64>
Cookie / Set-Cookie сессии в браузере; флаги HttpOnly, Secure, SameSite
Cache-Control, ETag, If-None-Match кеширование и 304
Location адрес созданного ресурса или редиректа
User-Agent кто клиент
Access-Control-Allow-Origin CORS: каким сайтам браузер разрешит читать ответ
Retry-After когда повторить после 429/503

URL и query-параметры

from urllib.parse import urlsplit, parse_qs, urlencode, quote, urljoin

url = "https://user@shop.example.com:8443/catalog/search?q=%D0%BA%D1%80%D1%83%D0%B6%D0%BA%D0%B0&tag=new&tag=sale#reviews"
parts = urlsplit(url)
print(parts.scheme, parts.hostname, parts.port, parts.path, parts.fragment)
print(parse_qs(parts.query))                              # значения — списки: ключ может повторяться

params = {"q": "кружка с котом", "page": 2, "sort": "-price"}
print("https://shop.example.com/search?" + urlencode(params))   # кодирование безопасно для URL
print(quote("отчёт 2026/01.pdf"), quote("отчёт 2026/01.pdf", safe=""))
print(urljoin("https://api.example.com/v1/users/", "42/orders"))

Никогда не склеивайте query-строку вручную через f-строку: пробел, & или # в значении сломают URL. Клиенты requests/httpx делают это сами через params=.

REST

REST — стиль проектирования API вокруг ресурсов (существительные в URL) и стандартных методов HTTP.

GET    /api/v1/users              список (с пагинацией и фильтрами)
POST   /api/v1/users              создать          → 201 + Location
GET    /api/v1/users/42           один             → 200 или 404
PUT    /api/v1/users/42           заменить целиком
PATCH  /api/v1/users/42           изменить поля
DELETE /api/v1/users/42           удалить          → 204
GET    /api/v1/users/42/orders    вложенный ресурс
POST   /api/v1/orders/7/cancel    действие, которое не ложится в CRUD

Практики хорошего API:

  • Пагинация: ?limit=20&offset=40 (просто, но медленно на больших смещениях) или курсорная ?after=<id> (стабильна при вставках).
  • Фильтры и сортировка: ?status=paid&sort=-created_at.
  • Версионирование: /api/v1/… или заголовок. Ломающие изменения — только в новой версии.
  • Единый формат ошибок: {"error": {"code": "email_taken", "message": "…", "field": "email"}} с правильным кодом.
  • Идемпотентность: для платежей — заголовок Idempotency-Key, чтобы повтор POST не списал деньги дважды.

Альтернативы REST: GraphQL (клиент сам описывает, какие поля нужны, один эндпоинт), gRPC (бинарный протокол на HTTP/2 со схемой Protobuf — для общения сервисов между собой), WebSocket (двусторонний канал для чатов и уведомлений).

Клиент requests

import requests

resp = requests.get(
    "https://api.github.com/search/repositories",
    params={"q": "language:python", "sort": "stars", "per_page": 5},
    headers={"Accept": "application/vnd.github+json"},
    timeout=10,                       # без таймаута запрос может висеть вечно!
)
resp.raise_for_status()               # 4xx/5xx → requests.HTTPError
data = resp.json()
for repo in data["items"]:
    print(repo["full_name"], repo["stargazers_count"])

resp = requests.post("https://httpbin.org/post", json={"name": "Аня"}, timeout=10)   # json= сам ставит Content-Type
print(resp.status_code, resp.headers["Content-Type"], resp.elapsed)

Для серии запросов к одному сервису используйте сессию: она переиспользует TCP-соединения (keep-alive) и хранит общие заголовки и cookies.

session = requests.Session()
session.headers["Authorization"] = f"Bearer {token}"
for page in range(1, 4):
    r = session.get("https://api.example.com/items", params={"page": page}, timeout=5)

Три обязательных правила HTTP-клиента

  1. Всегда timeout=. 2. Всегда проверяйте статус (raise_for_status()). 3. Повторяйте только то, что безопасно повторять (идемпотентные запросы, 5xx/429/сетевые ошибки), с растущей задержкой и ограничением числа попыток.

Повторы с экспоненциальной задержкой

import random

def backoff_delays(retries, base=0.5, cap=8.0, jitter=True):
    for attempt in range(retries):
        delay = min(cap, base * 2 ** attempt)
        yield round(random.uniform(0, delay), 2) if jitter else delay   # «full jitter»

print(list(backoff_delays(6, jitter=False)))
random.seed(1)
print(list(backoff_delays(6)))

Случайная добавка (jitter) нужна, чтобы тысячи клиентов после сбоя не повторяли запросы одновременно и не положили сервер снова. В продакшене берут готовые решения: urllib3.Retry для requests, библиотеку tenacity, транспорт с повторами в httpx.

httpx и асинхронные запросы

httpx — современная альтернатива с почти тем же API, поддержкой HTTP/2 и асинхронным клиентом. Он нужен в async-коде (FastAPI, боты): requests блокирует цикл событий.

import asyncio
import httpx

async def fetch_all(urls):
    async with httpx.AsyncClient(timeout=10) as client:
        responses = await asyncio.gather(*(client.get(u) for u in urls), return_exceptions=True)
    return [r.status_code if isinstance(r, httpx.Response) else repr(r) for r in responses]

print(asyncio.run(fetch_all(["https://example.com", "https://httpbin.org/status/503"])))

В тестах внешние HTTP-вызовы подменяют: responses/requests-mock для requests, respx или httpx.MockTransport для httpx.

Аутентификация

import base64, json, hmac, hashlib

# Basic: логин:пароль в base64 — это НЕ шифрование, только поверх HTTPS
print("Basic " + base64.b64encode("admin:s3cret".encode()).decode())

# JWT: header.payload.signature — base64url, подпись HMAC-SHA256
def b64url(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()

secret = b"server-secret"
header = b64url(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
payload = b64url(json.dumps({"sub": "42", "role": "user", "exp": 1767225600}).encode())
signature = b64url(hmac.new(secret, f"{header}.{payload}".encode(), hashlib.sha256).digest())
token = f"{header}.{payload}.{signature}"
print(token)

# любой может прочитать payload — секретов туда не кладут
padded = payload + "=" * (-len(payload) % 4)
print(json.loads(base64.urlsafe_b64decode(padded)))
  • Сессии и cookies — классика серверных сайтов (Django): сервер хранит сессию, браузер — cookie с её id.
  • Токены (Bearer) — для API и мобильных клиентов. JWT проверяется по подписи без похода в базу, но его трудно отозвать досрочно, поэтому живёт он недолго (минуты), а обновляется refresh-токеном.
  • OAuth 2.0 — вход «через Google/GitHub» и выдача доступа сторонним приложениям.
  • Пароли хранят только как медленный хеш с солью: bcrypt, argon2, scrypt (не sha256!).

Что происходит, когда вы открываете сайт

Классический вопрос на собеседовании. Краткий ответ по шагам:

  1. Браузер разбирает URL; ищет IP в кешах, затем спрашивает DNS.
  2. Устанавливает TCP-соединение (тройное рукопожатие) с IP:443.
  3. TLS-рукопожатие: проверка сертификата, договорённость о ключах шифрования.
  4. Отправляет HTTP-запрос GET / с заголовками и cookies.
  5. Запрос проходит балансировщик/обратный прокси (nginx) → приложение (gunicorn/uvicorn + Django/FastAPI) → база данных, кеш.
  6. Приходит ответ с HTML; браузер парсит его, догружает CSS/JS/картинки (часто с CDN), строит DOM и отрисовывает страницу.

Итог

  • HTTP: стартовая строка, заголовки, пустая строка, тело; сервер stateless.
  • Методы: GET/PUT/DELETE идемпотентны, POST — нет. Коды: 2xx успех, 4xx клиент, 5xx сервер; различайте 401 и 403.
  • REST — ресурсы в URL, смысл в методе, правильные коды, пагинация, версия, единый формат ошибок.
  • Клиент: timeout, raise_for_status, сессия, повторы с backoff и jitter; httpx.AsyncClient для async.

Практика решено 0 из 4

Вы не вошли — ответы проверяются, но прогресс не сохранится. Войти или зарегистрироваться.
Задание 1 Программа на Python

Разбор HTTP-запроса. Напишите parse_request(raw), которая разбирает текст HTTP/1.1-запроса (строки разделены \r\n, заголовки отделены от тела пустой строкой) и возвращает словарь:

  • method, path (без query-строки), version;
  • query — словарь параметров из query-строки (для повторяющихся ключей — последнее значение; значения декодированы);
  • headers — словарь с именами заголовков в нижнем регистре, значения без пробелов по краям;
  • body — строка после пустой строки (может быть пустой).

Подсказка: urllib.parse.urlsplit и parse_qsl.

Какие тесты проверяют код (2)
raw = "GET /search?q=python+%D0%BA%D1%83%D1%80%D1%81&page=2 HTTP/1.1\r\nHost: example.com\r\nAccept:  application/json \r\n\r\n"
r = parse_request(raw)
assert (r["method"], r["path"], r["version"]) == ("GET", "/search", "HTTP/1.1")
assert r["query"] == {"q": "python курс", "page": "2"}
assert r["headers"] == {"host": "example.com", "accept": "application/json"} and r["body"] == ""

raw = 'POST /api/users HTTP/1.1\r\nContent-Type: application/json\r\nX-Request-ID: a:b:c\r\n\r\n{"name": "Аня"}'
r = parse_request(raw)
assert r["path"] == "/api/users" and r["query"] == {}
assert r["headers"]["content-type"] == "application/json" and r["headers"]["x-request-id"] == "a:b:c"
import json; assert json.loads(r["body"]) == {"name": "Аня"}

Задание 2 Программа на Python

Повторы с экспоненциальной задержкой. Напишите fetch_with_retry(fetch, url, retries=3, base_delay=0.5, sleep=time.sleep):

  • вызывает fetch(url), который возвращает объект с атрибутом status_code (или бросает ConnectionError);
  • если статус 2xx–4xx (кроме 429) — сразу возвращает ответ: повторять ошибки клиента бессмысленно;
  • если статус 5xx или 429, либо ConnectionError — ждёт base_delay * 2**попытка (0.5, 1, 2…) и повторяет;
  • всего не больше 1 + retries вызовов; если и последняя попытка неудачна — вернуть последний ответ или пробросить последнее ConnectionError.

sleep передаётся параметром, чтобы в тестах не ждать по-настоящему.

Какие тесты проверяют код (3)
from types import SimpleNamespace as R
def make_fetch(results):
    calls = []
    def fetch(url):
        calls.append(url)
        r = results[len(calls) - 1]
        if isinstance(r, Exception):
            raise r
        return R(status_code=r)
    return fetch, calls
sleeps = []
fetch, calls = make_fetch([503, ConnectionError("сеть"), 200])
assert fetch_with_retry(fetch, "/x", sleep=sleeps.append).status_code == 200
assert len(calls) == 3 and sleeps == [0.5, 1.0]

sleeps = []
fetch, calls = make_fetch([404])
assert fetch_with_retry(fetch, "/x", sleep=sleeps.append).status_code == 404 and len(calls) == 1 and sleeps == []
fetch, calls = make_fetch([429, 500, 502, 503])
assert fetch_with_retry(fetch, "/x", retries=3, sleep=sleeps.append).status_code == 503
assert len(calls) == 4 and sleeps == [0.5, 1.0, 2.0]

sleeps = []
fetch, calls = make_fetch([ConnectionError("1"), ConnectionError("2"), ConnectionError("3")])
try:
    fetch_with_retry(fetch, "/x", retries=2, base_delay=1, sleep=sleeps.append)
except ConnectionError as e:
    assert str(e) == "3" and sleeps == [1, 2] and len(calls) == 3
else:
    raise AssertionError("После всех попыток ConnectionError нужно пробросить")

Задание 3 Заполнить таблицу

Какой код ответа вернуть?

201204400401403404409
POST /users — пользователь создан
DELETE /users/5 — удалён, тела ответа нет
В JSON нет обязательного поля email
Запрос без токена авторизации
Токен верный, но это чужой заказ, а прав администратора нет
GET /users/999 — такого пользователя нет
Регистрация с email, который уже занят
Отметьте ячейку — «+», пустая — «−».
Задание 4 Несколько вариантов

Какие HTTP-методы идемпотентны (повторный одинаковый запрос не меняет результат по сравнению с первым)?

Войдите, чтобы отмечать пройденные уроки.