Session, auth, adapters: persistent connections и расширяемость requests

В прошлом уроке мы дёргали API через requests.get(). Это удобно для одного запроса. Но если у тебя ETL pipeline, который шлёт 10 000 запросов к одному API — каждый requests.get() открывает новое TCP-соединение, делает TLS handshake, и закрывается. Это медленно. На 10 000 запросов overhead handshake-ов может составить 80 процентов времени работы.

В этом уроке разберёмся с requests.Session() — объектом, который держит persistent TCP/TLS соединения, переиспользует их, держит cookies и общие headers. Посмотрим механизмы аутентификации (Basic, Digest, кастомные), HTTPAdapter для тонкой настройки соединений, и hooks для логирования и трансформации ответов. Это инструменты, которые отделяют “просто работает” от “работает быстро и надёжно”.

Что не так с requests.get()

import requests
import time

start = time.time()
for _ in range(50):
    requests.get("https://jsonplaceholder.typicode.com/users/1")
print(f"Without session: {time.time() - start:.2f}s")
# Without session: 4.83s

start = time.time()
session = requests.Session()
for _ in range(50):
    session.get("https://jsonplaceholder.typicode.com/users/1")
print(f"With session: {time.time() - start:.2f}s")
# With session: 0.91s

Разница ~5x. Где она берётся:

Что делает Session

Объект requests.Session():

1. Connection pool — держит открытые TCP/TLS соединения на каждый хост, переиспользует их (HTTP keep-alive).
2. Cookie persistence — cookies из ответов автоматически сохраняются и шлются с следующими запросами на тот же домен.
3. Default config — общие headers, auth, hooks, proxies, verify настраиваются один раз и применяются ко всем запросам.

session = requests.Session()

# Общие headers -- будут на каждом запросе
session.headers.update({
    "Authorization": "Bearer eyJhbGc...",
    "User-Agent": "MyETL/1.0",
    "Accept": "application/json",
})

# Cookies сохраняются автоматически
session.get("https://example.com/login")  # сервер прислал session_id
session.get("https://example.com/profile")  # session_id отправляется с запросом

# API идентичный requests.get/post/put/...
r = session.get("https://api.example.com/users/1")

with requests.Session() as session:
    r = session.get("https://api.example.com/users/1")
    # session.close() автоматически в конце блока

Auth: встроенные и кастомные

requests поддерживает Basic, Digest и custom auth.

HTTPBasicAuth

Самая простая схема — login/password в Authorization header (base64). Используется в legacy системах, internal APIs.

import requests
from requests.auth import HTTPBasicAuth

# Длинная форма
r = requests.get(
    "https://api.example.com/secret",
    auth=HTTPBasicAuth("alice", "password123")
)

# Короткая (tuple -- equivalent)
r = requests.get(
    "https://api.example.com/secret",
    auth=("alice", "password123")
)

Что ляжет в headers:

Authorization: Basic YWxpY2U6cGFzc3dvcmQxMjM=

Это base64("alice:password123"). Не шифрование — кодирование. Если канал не TLS — credentials читабельны для любого, кто перехватил трафик. Используй только с HTTPS.

HTTPDigestAuth

Более старая схема с challenge-response. Сервер шлёт nonce, клиент его hash-ит с паролем. Безопаснее Basic, но тоже legacy. Используется в некоторых router’ах, корпоративных системах.

from requests.auth import HTTPDigestAuth

r = requests.get(
    "https://api.example.com/secret",
    auth=HTTPDigestAuth("alice", "password123")
)

requests делает 2 запроса: первый получает 401 + WWW-Authenticate с nonce, второй с правильным response.

Кастомный auth callable

Собственная схема — любой callable, принимающий Request и возвращающий его (модифицированный):

import requests
from requests.auth import AuthBase
import hmac, hashlib, time

class HMACAuth(AuthBase):
    """HMAC signature auth -- типично для крипто-бирж и internal API."""

    def __init__(self, api_key, secret):
        self.api_key = api_key
        self.secret = secret

    def __call__(self, request):
        timestamp = str(int(time.time()))
        message = f"{timestamp}{request.method}{request.path_url}{request.body or ''}"
        signature = hmac.new(
            self.secret.encode(),
            message.encode(),
            hashlib.sha256
        ).hexdigest()

        request.headers["X-API-Key"] = self.api_key
        request.headers["X-Timestamp"] = timestamp
        request.headers["X-Signature"] = signature
        return request

# Использование
session = requests.Session()
session.auth = HMACAuth("my_key", "my_secret")
r = session.get("https://api.exchange.com/balance")

Кастомный auth = подкласс AuthBase с __call__(request). Подходит для любой схемы: HMAC, AWS Signature V4, JWT с rotation.

HTTPAdapter: тюнинг соединений

HTTPAdapter — это объект, через который Session делает запросы. По умолчанию у Session есть один adapter для http:// и один для https://. Можно создавать кастомные:

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

adapter = HTTPAdapter(
    pool_connections=20,    # количество connection pools (один pool на хост)
    pool_maxsize=20,        # макс соединений в одном pool
    max_retries=Retry(      # auto-retry на сетевые ошибки
        total=3,
        backoff_factor=0.5,
        status_forcelist=[502, 503, 504]
    )
)

session.mount("https://", adapter)
session.mount("http://", adapter)

pool_maxsize=20 важен для конкурентности. Если у тебя ThreadPoolExecutor с 50 потоками, а pool_maxsize=10 — 40 потоков будут стоять в очереди за соединениями.

Mounting per-prefix

Можно использовать разные adapter-ы для разных URL:

session = requests.Session()

# Для production-API -- короткие таймауты, мало retry
prod_adapter = HTTPAdapter(max_retries=2)
session.mount("https://api.prod.example.com/", prod_adapter)

# Для test-API -- больше retry, агрессивнее
test_adapter = HTTPAdapter(max_retries=5)
session.mount("https://api.test.example.com/", test_adapter)

# Для всех остальных HTTPS -- дефолт (3 retry)
default_adapter = HTTPAdapter(max_retries=Retry(total=3))
session.mount("https://", default_adapter)

Match идёт по самому длинному префиксу (как в роутерах).

Hooks: response callback

Hooks позволяют вклиниться в pipeline обработки. Самый полезный — response:

import requests
import logging

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)

def log_response(response, *args, **kwargs):
    log.info(f"{response.request.method} {response.url} -> {response.status_code} ({response.elapsed.total_seconds():.2f}s)")
    if response.status_code >= 400:
        log.warning(f"Error body: {response.text[:200]}")

session = requests.Session()
session.hooks["response"] = log_response

session.get("https://jsonplaceholder.typicode.com/users/1")
# INFO: GET https://jsonplaceholder.typicode.com/users/1 -> 200 (0.43s)

session.get("https://jsonplaceholder.typicode.com/users/9999")
# INFO: GET https://jsonplaceholder.typicode.com/users/9999 -> 404 (0.21s)
# WARNING: Error body: {}

Hook вызывается на каждый response. Полезно для:

• Централизованного логирования.
• Сбора metrics (latency per endpoint).
• Auto-decode/transform response.
• Auto-raise on errors (response.raise_for_status() в hook-е).

Несколько hooks — session.hooks["response"] = [hook1, hook2].

Полный production pattern

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def make_api_session(base_url, token, timeout=30):
    """Production-ready Session для API клиента."""
    session = requests.Session()

    # Общие headers
    session.headers.update({
        "Authorization": f"Bearer {token}",
        "User-Agent": "DataPipeline/1.0",
        "Accept": "application/json",
    })

    # Adapter с retry и pooling
    retry = Retry(
        total=3,
        backoff_factor=0.5,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "PUT", "DELETE", "HEAD", "OPTIONS"],  # idempotent only
    )
    adapter = HTTPAdapter(
        pool_connections=10,
        pool_maxsize=20,
        max_retries=retry,
    )
    session.mount("https://", adapter)
    session.mount("http://", adapter)

    # Hook для логирования
    def log_hook(r, *args, **kwargs):
        elapsed = r.elapsed.total_seconds()
        if r.status_code >= 400:
            print(f"ERROR {r.status_code}: {r.request.method} {r.url} ({elapsed:.2f}s)")

    session.hooks["response"] = log_hook

    # Дефолтный таймаут не задаётся в Session, нужно явно. Workaround:
    original_request = session.request
    def request_with_timeout(method, url, **kwargs):
        kwargs.setdefault("timeout", timeout)
        return original_request(method, url, **kwargs)
    session.request = request_with_timeout

    return session


# Использование
with make_api_session("https://api.example.com", "my_token") as s:
    r = s.get("/users/1")
    r.raise_for_status()
    print(r.json())

Попробуй сам

import requests
import time

# Сравнение: с Session vs без
URL = "https://jsonplaceholder.typicode.com/users/1"
N = 30

# Без Session
start = time.time()
for _ in range(N):
    requests.get(URL)
no_session = time.time() - start

# С Session
start = time.time()
with requests.Session() as s:
    for _ in range(N):
        s.get(URL)
with_session = time.time() - start

print(f"Without Session: {no_session:.2f}s")
print(f"With Session:    {with_session:.2f}s")
print(f"Speedup:         {no_session / with_session:.1f}x")

Сделай Session с Basic Auth для https://httpbin.org/basic-auth/user/passwd:

with requests.Session() as s:
    s.auth = ("user", "passwd")
    r = s.get("https://httpbin.org/basic-auth/user/passwd")
    print(r.status_code)  # 200
    print(r.json())       # {"authenticated": true, "user": "user"}

Добавь hook, который логирует все запросы в файл api.log. Запусти 5 разных запросов — посмотри лог.

DE-контекст

1. Airflow tasks с одним API: оборачивай в Session внутри task. 100 запросов к одному endpoint станут в 5 раз быстрее.
2. Pagination loops: запрос за запросом следующих страниц — Session обязателен.
3. OAuth2 token refresh: hook на response ловит 401, обновляет token, ретраит запрос.
4. Multi-API ETL: одна Session на API. Можно держать несколько в dict-е {"github": gh_session, "stripe": stripe_session, ...}.
5. Test environments: mount adapter с базой localhost:8080 для test, api.prod.x.com для prod — переключение через переменную окружения.

Killer takeaway

requests.Session() — это connection pool + cookie jar + общие настройки. На 10+ запросов к одному API даёт 3-5x ускорение за счёт HTTP keep-alive (один TCP/TLS handshake на много запросов). Используй with requests.Session() as s: для авто-закрытия. Auth: HTTPBasicAuth (legacy, login:password в base64), HTTPDigestAuth (challenge-response), custom callable через подкласс AuthBase для HMAC/AWS-signature/JWT. HTTPAdapter тюнит pool_maxsize и max_retries; mount per-prefix позволяет разные настройки для разных URL. Hooks (session.hooks["response"]) — централизованное логирование, metrics, auto-handling. Production pattern: Session + Adapter с Retry + hook для логов + base headers + дефолтный timeout — основа любого API-клиента в твоём ETL.