Механизмы сохранения состояния в HTTP-клиентах: объект Session библиотеки requests

Расширенные примеры использования Session в реальных сценариях

Пример 1. Автоматическое обновление токена аутентификации

Некоторые API используют токены с ограниченным сроком жизни. После истечения токена нужно отправить запрос на обновление. Организуем это внутри сессии через перехват ошибок.

import requests

class RefreshingSession:
    def __init__(self, auth_url, credentials):
        self.session = requests.Session()
        self.auth_url = auth_url
        self.credentials = credentials
        self._get_token()
    
    def _get_token(self):
        resp = self.session.post(self.auth_url, json=self.credentials)
        resp.raise_for_status()
        token = resp.json()['access_token']
        self.session.headers.update({'Authorization': f'Bearer {token}'})
    
    def request(self, method, url, **kwargs):
        r = self.session.request(method, url, **kwargs)
        if r.status_code == 401:
            # Токен истёк, обновляем
            self._get_token()
            r = self.session.request(method, url, **kwargs)
        return r

# Использование:
client = RefreshingSession('https://api.example.com/auth', {'client_id': '...', 'client_secret': '...'})
response = client.request('GET', 'https://api.example.com/data')
print(response.text)

(пример вывода: данные в JSON, без ошибок 401)

Этот паттерн особенно полезен для OAuth2, где токены живут, например, 15 минут.

Пример 2. Работа с несколькими сайтами в одной сессии

Если нужно последовательно обращаться к разным хостам, куки и заголовки не должны смешиваться. Создаём отдельные сессии для каждого домена.

import requests

# Сессия для site1
s1 = requests.Session()
s1.headers.update({'Host': 'site1.com'})
s1.get('https://site1.com/login')

# Сессия для site2
s2 = requests.Session()
s2.cookies.set('auth_cookie', 'xxxx', domain='site2.com')
s2.get('https://site2.com/profile')

# Запросы не пересекаются
print(s1.cookies)  # только куки site1
print(s2.cookies)  # только куки site2

Можно также использовать requests.Session() как менеджер контекста для временных сессий.

Пример 3. Настройка таймаутов и повторных попыток через адаптер

Сессия позволяет монтировать собственные адаптеры HTTP и HTTPS, чтобы задать таймауты, пул соединений и стратегию retry.

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

session = requests.Session()

retry_strategy = Retry(
    total=3,
    backoff_factor=1,
    status_forcelist=[500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy, pool_connections=10, pool_maxsize=10)

# Монтируем адаптер ко всем URL
session.mount('https://', adapter)
session.mount('http://', adapter)

# Выполняем запрос с таймаутом в 5 секунд
try:
    response = session.get('https://httpbin.org/delay/1', timeout=5)
    print(response.status_code)
except requests.exceptions.Timeout:
    print('Таймаут истёк')
except requests.exceptions.ConnectionError:
    print('Ошибка соединения')
except requests.exceptions.RetryError:
    print('Исчерпаны попытки')

200 (или сообщение об ошибке, если сервер недоступен)

Параметр timeout можно задать и в общих настройках сессии через session.request, но адаптер даёт более тонкий контроль.

Пример 4. Использование прокси для всех запросов сессии

Сессия упрощает работу с прокси: достаточно задать словарь один раз.

import requests

session = requests.Session()
session.proxies = {
    'http': 'http://user:pass@proxy.example.com:8080',
    'https': 'https://user:pass@proxy.example.com:8080',
}
# Все запросы пойдут через прокси
response = session.get('https://httpbin.org/ip')
print(response.json()['origin'])  # IP прокси

Если нужно изменить прокси для одного запроса, можно передать параметр proxies в метод.

Пример 5. Потоковое скачивание файла с сохранением сессии

Скачивание большого файла с авторизацией - сессия поддерживает потоковую загрузку через stream=True.

import requests

session = requests.Session()
session.auth = ('user', 'pass')

url = 'https://httpbin.org/stream-bytes/1024'
response = session.get(url, stream=True)
response.raise_for_status()

with open('downloaded.bin', 'wb') as f:
    for chunk in response.iter_content(chunk_size=128):
        if chunk:
            f.write(chunk)
print('Файл загружен')

Важно: при потоковой загрузке соединение держится открытым, поэтому следует явно закрывать ответ (или использовать контекстный менеджер).

Пример 6. Проверка SSL-сертификата и настройка verify

Иногда требуется отключить проверку SSL или указать свой CA-файл. Это можно сделать как для всей сессии, так и для отдельного запроса.

import requests

session = requests.Session()
# Отключить проверку (не рекомендуется для продакшена)
session.verify = False
# или указать путь к сертификату
# session.verify = '/path/to/custom_ca.pem'

# Подавление предупреждения об SSL
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

response = session.get('https://self-signed.badssl.com/')
print(response.status_code)  # 200, но с предупреждением

Более правильный подход - использовать certifi или явно передавать verify в запрос:

response = session.get('https://example.com', verify=True)

Пример 7. Использование сессии с httpbin.org для отладки

Сервис httpbin.org позволяет тестировать куки, заголовки и аутентификацию. Продемонстрируем полный цикл.

import requests

session = requests.Session()
session.headers.update({'User-Agent': 'MyApp'})

# 1. Установить куки через сервер
session.get('https://httpbin.org/cookies/set?test=123')
# 2. Проверить, что куки сохранились
resp = session.get('https://httpbin.org/cookies')
print('Куки после установки:', resp.json()['cookies'])  # {'test': '123'}

# 3. Отправить запрос с базовой аутентификацией
session.auth = ('user', 'pass')
resp = session.get('https://httpbin.org/basic-auth/user/pass')
print('Аутентификация:', resp.status_code)  # 200

# 4. Проверить итоговый заголовок User-Agent
resp = session.get('https://httpbin.org/headers')
print('User-Agent:', resp.json()['headers']['User-Agent'])

Куки после установки: {'test': '123'}
Аутентификация: 200
User-Agent: MyApp

Такой подход помогает отлаживать взаимодействие с реальными API.

Пример 8. Обработка больших ответов и сжатие

Сессия поддерживает сжатие gzip/deflate автоматически. Можно явно указать заголовки Accept-Encoding.

import requests

session = requests.Session()
# Явно запросить gzip
session.headers['Accept-Encoding'] = 'gzip'

response = session.get('https://httpbin.org/gzip')
print('Содержимое:', response.text[:100])  # декомпрессия выполнена автоматически
print('Заголовки:', dict(response.headers))  # Content-Encoding: gzip

Размер ответа может быть большим - используйте iter_content для потоковой обработки.

Все примеры демонстрируют, как объект Session упрощает управление состоянием HTTP-клиента. Комбинируя описанные техники, можно строить устойчивые и производительные сетевые приложения.