HTTP, REST и клиенты requests/httpx
Запрос и ответ
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-клиента
- Всегда
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!).
Что происходит, когда вы открываете сайт
Классический вопрос на собеседовании. Краткий ответ по шагам:
- Браузер разбирает URL; ищет IP в кешах, затем спрашивает DNS.
- Устанавливает TCP-соединение (тройное рукопожатие) с IP:443.
- TLS-рукопожатие: проверка сертификата, договорённость о ключах шифрования.
- Отправляет HTTP-запрос
GET /с заголовками и cookies. - Запрос проходит балансировщик/обратный прокси (nginx) → приложение (gunicorn/uvicorn + Django/FastAPI) → база данных, кеш.
- Приходит ответ с 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
Разбор 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": "Аня"}
from urllib.parse import urlsplit, parse_qsl
def parse_request(raw):
head, _, body = raw.partition("\r\n\r\n")
lines = head.split("\r\n")
method, target, version = lines[0].split(" ")
url = urlsplit(target)
headers = {}
for line in lines[1:]:
name, _, value = line.partition(":")
headers[name.strip().lower()] = value.strip()
return {
"method": method,
"path": url.path,
"version": version,
"query": dict(parse_qsl(url.query)),
"headers": headers,
"body": body,
}
Повторы с экспоненциальной задержкой. Напишите 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 нужно пробросить")
import time
def fetch_with_retry(fetch, url, retries=3, base_delay=0.5, sleep=time.sleep):
for attempt in range(retries + 1):
last = attempt == retries
try:
response = fetch(url)
except ConnectionError:
if last:
raise
else:
if response.status_code < 500 and response.status_code != 429:
return response
if last:
return response
sleep(base_delay * 2 ** attempt)
Какой код ответа вернуть?
| 201 | 204 | 400 | 401 | 403 | 404 | 409 | |
|---|---|---|---|---|---|---|---|
| POST /users — пользователь создан | |||||||
| DELETE /users/5 — удалён, тела ответа нет | |||||||
| В JSON нет обязательного поля email | |||||||
| Запрос без токена авторизации | |||||||
| Токен верный, но это чужой заказ, а прав администратора нет | |||||||
| GET /users/999 — такого пользователя нет | |||||||
| Регистрация с email, который уже занят |
Какие HTTP-методы идемпотентны (повторный одинаковый запрос не меняет результат по сравнению с первым)?
Войдите, чтобы отмечать пройденные уроки.