Что такое сессия в Requests?

Сессия в Requests

Session в Requests — это объект, который управляет состоянием HTTP-сессии, позволяя вам сохранять параметры между несколькими запросами (cookies, заголовки, аутентификация). Сессия повторно использует базовое TCP-соединение, что повышает производительность при множественных запросах к одному сервису.

Основное применение

Без сессии каждый запрос создаёт новое соединение:

import requests

# Без сессии — 3 отдельных соединения
requests.get('https://api.example.com/users')
requests.post('https://api.example.com/login', json={"user": "admin"})
requests.get('https://api.example.com/profile')

С сессией — одно переиспользуемое соединение:

import requests

session = requests.Session()

# Все запросы используют одно соединение
session.get('https://api.example.com/users')
session.post('https://api.example.com/login', json={"user": "admin"})
session.get('https://api.example.com/profile')

session.close()  # Закрываем соединение

Управление Cookies

Сессия автоматически сохраняет и отправляет cookies:

import requests

session = requests.Session()

# Первый запрос (логин)
response = session.post(
    'https://api.example.com/login',
    json={"username": "admin", "password": "secret"}
)
# Сервер возвращает Set-Cookie в заголовке
# Session автоматически сохраняет этот cookie

print(session.cookies)  # CookieJar со всеми cookies
print(session.cookies.get_dict())  # Словарь cookies

# Второй запрос — cookie отправляется автоматически
profile = session.get('https://api.example.com/profile')
print(profile.json())  # Сервер распознаёт, что мы авторизованы

Установка заголовков по умолчанию

import requests

session = requests.Session()

# Устанавливаем заголовки для всех запросов в сессии
session.headers.update({
    'User-Agent': 'MyApp/1.0',
    'Authorization': 'Bearer token_123abc',
    'Accept': 'application/json'
})

# Все запросы будут содержать эти заголовки
response1 = session.get('https://api.example.com/users')
response2 = session.get('https://api.example.com/posts')
# Оба запроса включают Authorization заголовок

Базовая аутентификация

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth

session = requests.Session()

# Способ 1: HTTPBasicAuth
session.auth = HTTPBasicAuth('username', 'password')
response = session.get('https://api.example.com/protected')

# Способ 2: Сокращённый синтаксис
session.auth = ('username', 'password')

# Все запросы в сессии используют эту аутентификацию
for url in ['https://api.example.com/users', 'https://api.example.com/posts']:
    response = session.get(url)

Практический пример: API клиент с сессией

import requests
from typing import Any, Dict

class APIClient:
    """Клиент для работы с API"""
    
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.session = requests.Session()
        
        # Устанавливаем общие заголовки
        self.session.headers.update({
            'Authorization': f'Bearer {api_key}',
            'Content-Type': 'application/json',
            'User-Agent': 'APIClient/1.0'
        })
    
    def get(self, endpoint: str, params: Dict[str, Any] = None) -> Dict:
        """GET запрос"""
        url = f"{self.base_url}/{endpoint}"
        response = self.session.get(url, params=params)
        response.raise_for_status()
        return response.json()
    
    def post(self, endpoint: str, data: Dict[str, Any] = None) -> Dict:
        """POST запрос"""
        url = f"{self.base_url}/{endpoint}"
        response = self.session.post(url, json=data)
        response.raise_for_status()
        return response.json()
    
    def put(self, endpoint: str, data: Dict[str, Any] = None) -> Dict:
        """PUT запрос"""
        url = f"{self.base_url}/{endpoint}"
        response = self.session.put(url, json=data)
        response.raise_for_status()
        return response.json()
    
    def delete(self, endpoint: str) -> bool:
        """DELETE запрос"""
        url = f"{self.base_url}/{endpoint}"
        response = self.session.delete(url)
        response.raise_for_status()
        return True
    
    def close(self):
        """Закрыть сессию"""
        self.session.close()
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.close()

# Использование
with APIClient('https://api.example.com', 'my_api_key') as client:
    users = client.get('users')
    new_user = client.post('users', {'name': 'Alice', 'email': 'alice@example.com'})
    client.put('users/1', {'name': 'Bob'})
    client.delete('users/2')

Контекстный менеджер (Context Manager)

import requests

# Правильно: сессия закроется автоматически
with requests.Session() as session:
    response = session.get('https://api.example.com/users')
    print(response.json())
# Соединение автоматически закрыто после выхода из блока with

Обработка ошибок в сессии

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

def create_session_with_retries():
    """Создать сессию с автоматическими переповторами"""
    session = requests.Session()
    
    # Конфигурируем переповторы
    retry_strategy = Retry(
        total=3,  # 3 попытки всего
        backoff_factor=1,  # Экспоненциальная задержка: 1s, 2s, 4s
        status_forcelist=[429, 500, 502, 503, 504],  # Статус-коды для переповтора
    )
    
    # Применяем стратегию к обоим HTTP и HTTPS
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    
    return session

# Использование
session = create_session_with_retries()
try:
    response = session.get('https://api.example.com/data', timeout=10)
    response.raise_for_status()
except requests.RequestException as e:
    print(f"Ошибка запроса: {e}")
finally:
    session.close()

Пулинг соединений

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.connection import create_connection

def patched_create_connection(address, *args, **kwargs):
    """Кастомная функция для создания соединения"""
    host, port = address
    print(f"Подключаемся к {host}:{port}")
    return create_connection(address, *args, **kwargs)

session = requests.Session()

# Настраиваем пулинг соединений
adapter = HTTPAdapter(
    pool_connections=10,  # Максимум соединений в пуле
    pool_maxsize=20,  # Максимум переиспользуемых соединений
    max_retries=3
)
session.mount('https://', adapter)

# Делаем несколько запросов (переиспользуя соединения)
for i in range(5):
    response = session.get(f'https://api.example.com/item/{i}')
    print(f"Статус: {response.status_code}")

Сравнение: Session vs requests.get()

import time
import requests

urls = ['https://httpbin.org/uuid'] * 10

# БЕЗ сессии — 10 новых соединений
start = time.time()
for url in urls:
    requests.get(url)
time_without_session = time.time() - start
print(f"Без сессии: {time_without_session:.2f}s")  # медленнее

# С СЕССИЕЙ — 1 переиспользуемое соединение
start = time.time()
with requests.Session() as session:
    for url in urls:
        session.get(url)
time_with_session = time.time() - start
print(f"С сессией: {time_with_session:.2f}s")  # быстрее (в 2-3 раза)

Лучшие практики

1. Используйте контекстный менеджер — гарантирует закрытие соединения
2. Переиспользуйте одну сессию — для множественных запросов к одному сервису
3. Устанавливайте timeout — избегайте зависаний
4. Обрабатывайте исключения — используйте raise_for_status()
5. Конфигурируйте переповторы — для устойчивости к сетевым ошибкам

Session — это критически важный инструмент для эффективной работы с HTTP-запросами в Python, обеспечивая производительность и управление состоянием.