Управление сетевыми ошибками в Python Requests: от теории к практике
Основные ошибки библиотеки requests и их обработка
Библиотека requests в Python предоставляет удобный интерфейс для HTTP-запросов, но в реальных сетевых приложениях часто возникают исключения. Понимание типов ошибок и умение их обрабатывать - ключ к стабильной работе. Рассмотрим наиболее распространенные исключения и способы их отлова.
Как универсально обработать все исключения requests?
Самый надежный способ - перехватывать базовое исключение RequestException и анализировать его тип. Это позволяет предусмотреть любую сетевую или HTTP-ошибку.
import requests
from requests.exceptions import RequestException
try:
response = requests.get('https://httpbin.org/status/404')
response.raise_for_status() # генерирует HTTPError для плохих кодов
except RequestException as e:
print(f'Произошла ошибка: {e}')
if isinstance(e, requests.exceptions.ConnectionError):
print('Ошибка подключения')
elif isinstance(e, requests.exceptions.Timeout):
print('Превышено время ожидания')
elif isinstance(e, requests.exceptions.HTTPError):
print(f'HTTP ошибка: {e.response.status_code}')
else:
print('Другая ошибка')Python 2 requests (библиотека requests в python 2)
Пояснение: raise_for_status() активирует исключение только для кодов 4xx/5xx. Без него HTTPError не возникнет. Базовое исключение RequestException покрывает все типы, включая внутренние ошибки библиотеки.
Типичные проблемы:
- Забывают импортировать
RequestExceptionизrequests.exceptions. - Не вызывают
raise_for_status(), поэтому HTTPError не отлавливается. - Путают порядок проверки типов (сначала более специфичные, потом общие).
Как обработать ошибку подключения (ConnectionError) и выполнить повтор?
При потере сети или недоступности хоста requests выбрасывает ConnectionError. Часто требуется автоматически повторить запрос через некоторое время.
import requests
import time
from requests.exceptions import ConnectionError
def fetch_with_retry(url, retries=3, delay=2):
for attempt in range(1, retries+1):
try:
response = requests.get(url, timeout=5)
response.raise_for_status()
return response
except ConnectionError as e:
print(f'Попытка {attempt}: ошибка подключения - {e}')
if attempt < retries:
time.sleep(delay)
else:
raise
return None
result = fetch_with_retry('https://httpbin.org/delay/10', retries=2, delay=1)Python 3 requests (библиотека requests в python 3)
Пояснение: Функция пробует выполнить запрос до retries раз. Между попытками пауза delay секунд. Если все попытки неудачны, исключение пробрасывается выше. Таймаут timeout=5 защищает от зависания.
Возможные проблемы:
- Слишком частые повторы без задержки могут усугубить нагрузку на сервер.
- Не учитывается ошибка Timeout – она тоже может требовать повтора.
Как отловить и корректно обработать Timeout?
Исключение Timeout возникает, если сервер не отвечает в течение заданного времени. requests поддерживает раздельные таймауты на соединение и чтение.
import requests
from requests.exceptions import Timeout
try:
# timeout = (connect_timeout, read_timeout)
response = requests.get('https://httpbin.org/delay/10', timeout=(3, 5))
response.raise_for_status()
except Timeout as e:
print(f'Таймаут: {e}')
# Можно повторить запрос или сообщить пользователюPython requests url (выполнение запроса по url с помощью requests в python)
Пояснение: Первое значение - максимальное время на установление соединения, второе - на получение данных. Если не указать, таймаут не действует, и запрос может зависнуть навсегда.
Частые ошибки:
- Указание только одного числа - тогда это таймаут и на соединение, и на чтение одновременно.
- Слишком строгие таймауты могут приводить к ложным срабатываниям на медленных сетях.
Что делать при HTTP-ошибках (4xx, 5xx)?
HTTP-статусы ошибок обрабатываются исключением HTTPError. Его обычно генерирует метод raise_for_status().
import requests
try:
response = requests.get('https://httpbin.org/status/403')
if response.status_code == 403:
print('Доступ запрещен, предпринимаем альтернативные действия')
response.raise_for_status()
except requests.exceptions.HTTPError as e:
print(f'HTTP ошибка: {e.response.status_code} - {e.response.reason}')
# Анализ тела ответа для деталей
if e.response.status_code == 404:
print('Ресурс не найден')
elif 500 <= e.response.status_code < 600:
print('Серверная ошибка, повтор через 10 секунд')Python requests headers (заголовки запросов в python requests)
Пояснение: Можно проверить код ответа до вызова raise_for_status(), чтобы выполнить специфичную логику. Само исключение доступно в e.response.
Проблема: Если не вызывать raise_for_status(), HTTPError не возникнет, и программа продолжит работать с некорректными данными.
Как избежать ошибки TooManyRedirects?
По умолчанию requests следует перенаправлениям (до 30 раз). Если их слишком много, выбрасывается TooManyRedirects.
import requests
from requests.exceptions import TooManyRedirects
try:
response = requests.get('https://httpbin.org/redirect/10', allow_redirects=True)
response.raise_for_status()
except TooManyRedirects as e:
print(f'Слишком много перенаправлений: {e}')
# Можно отключить перенаправления и обработать вручную
response = requests.get('https://httpbin.org/redirect/10', allow_redirects=False)
print(f'Статус первого ответа: {response.status_code}')
print(f'Заголовок Location: {response.headers.get("Location")}')Python requests exceptions (исключения в python requests)
Пояснение: Параметр allow_redirects=False возвращает первый ответ (обычно 302) без автоматического следования. Затем приложение может самостоятельно анализировать цепочку.
Типичная ошибка: Игнорирование лимита перенаправлений может привести к рекурсии в нестабильных сценариях.
Как настроить повторные попытки на уровне Session?
Объект Session позволяет автоматически повторять запросы при определенных исключениях с помощью HTTPAdapter и Retry.
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[500, 502, 503, 504],
allowed_methods=['GET', 'POST']
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount('https://', adapter)
session.mount('http://', adapter)
try:
response = session.get('https://httpbin.org/status/503', timeout=5)
response.raise_for_status()
except requests.exceptions.RetryError as e:
print(f'Исчерпаны попытки: {e}')
except requests.exceptions.RequestException as e:
print(f'Ошибка: {e}')Python requests codes (коды ответов http в python requests)
Пояснение: Retry настраивает количество попыток, фактор задержки (backoff_factor) и коды статусов, при которых нужно повторить. HTTPAdapter монтируется на протоколы.
Возможная проблема: Неправильная настройка allowed_methods может привести к повторению идемпотентных запросов (POST по умолчанию не повторяется, если не указать явно).
Как комбинировать обработку с логированием?
Логирование позволяет отследить все исключения и действия по восстановлению.
import requests
import logging
from requests.exceptions import RequestException, Timeout
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def safe_get(url, **kwargs):
try:
response = requests.get(url, timeout=5, **kwargs)
response.raise_for_status()
return response
except Timeout as e:
logger.warning(f'Таймаут для {url}: {e}')
raise
except RequestException as e:
logger.error(f'Ошибка запроса {url}: {e}')
raise
try:
data = safe_get('https://httpbin.org/delay/15')
except RequestException:
passПояснение: Каждый тип ошибки логируется с соответствующим уровнем. Это упрощает отладку в production-среде.
Проблема: Избыточное логирование может захламлять файлы, если не настроить фильтры.
Расширенные примеры обработки ошибок requests
Использование tenacity для повторных попыток с экспоненциальной задержкой
Библиотека tenacity предоставляет декоратор @retry с гибкими настройками: количество попыток, задержка, исключения для повтора.
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests
from requests.exceptions import ConnectionError, Timeout
@retry(
stop=stop_after_attempt(4),
wait=wait_exponential(multiplier=1, min=2, max=30),
retry=retry_if_exception_type((ConnectionError, Timeout))
)
def fetch_with_tenacity(url):
response = requests.get(url, timeout=5)
response.raise_for_status()
return response
try:
result = fetch_with_tenacity('https://httpbin.org/status/503')
print('Успешно:', result.status_code)
except requests.exceptions.RequestException as e:
print('Ошибка после всех попыток:', e)2025-03-25 12:00:01: Попытка 1, задержка 2.0 сек 2025-03-25 12:00:03: Попытка 2, задержка 4.0 сек 2025-03-25 12:00:07: Попытка 3, задержка 8.0 сек 2025-03-25 12:00:15: Попытка 4, задержка 16.0 сек Ошибка после всех попыток: 503 Server Error
Пояснение: tenacity автоматически вычисляет задержку по формуле: min( multiplier * (2^(attempt-1)), max ). Декоратор применяется только к указанным исключениям. При превышении числа попыток исключение пробрасывается.
Кастомное исключение с детальной информацией об ответе
Иногда полезно создать собственное исключение, которое включает полный ответ сервера для отладки.
import requests
from requests.exceptions import HTTPError
class DetailedHTTPError(HTTPError):
def __init__(self, request, response):
super().__init__(f'HTTP {response.status_code}: {response.reason}')
self.request = request
self.response = response
def check_response(response):
if response.status_code >= 400:
raise DetailedHTTPError(response.request, response)
try:
resp = requests.get('https://httpbin.org/status/418')
check_response(resp)
except DetailedHTTPError as e:
print('Детали ошибки:')
print('URL:', e.request.url)
print('Статус:', e.response.status_code)
print('Тело:', e.response.text[:200])Детали ошибки: URL: https://httpbin.org/status/418 Статус: 418 Тело: I'm a teapot
Пояснение: Класс наследуется от HTTPError, поэтому его можно перехватывать в общем обработчике RequestException. Метод check_response может использоваться как замена raise_for_status().
Обработка ошибок прокси (ProxyError)
При использовании прокси requests может выбрасывать requests.exceptions.ProxyError (подкласс ConnectionError).
import requests
try:
proxies = {'http': 'http://invalid:8080', 'https': 'http://invalid:8080'}
response = requests.get('https://httpbin.org/ip', proxies=proxies, timeout=5)
response.raise_for_status()
except requests.exceptions.ProxyError as e:
print('Ошибка подключения к прокси:', e)
# Проверка аутентификации или доступности прокси
except requests.exceptions.ConnectionError as e:
print('Общая ошибка соединения (возможно прокси):', e)Ошибка подключения к прокси: HTTPSConnectionPool(host='httpbin.org', port=443): Max retries exceeded with url: /ip (Caused by ProxyError('Cannot connect to proxy.', OSError(...)))Пояснение: ProxyError наследуется от ConnectionError, поэтому порядок проверки важен: сначала нужно отлавливать более специфичное исключение. Рекомендуется проверять настройки прокси и аутентификацию.
Обработка ошибок SSL (SSLError)
Проблемы с сертификатами вызывают requests.exceptions.SSLError (подкласс ConnectionError).
import requests
from requests.exceptions import SSLError
try:
response = requests.get('https://expired.badssl.com/', timeout=5)
response.raise_for_status()
except SSLError as e:
print('Ошибка SSL:', e)
# Можно отключить проверку (не рекомендуется)
# response = requests.get(url, verify=False)Ошибка SSL: HTTPSConnectionPool(host='expired.badssl.com', port=443): Max retries exceeded with url: / (Caused by SSLError(SSLCertVerificationError(...)))
Пояснение: Для диагностики можно использовать параметр verify с путем к сертификату CA или False (только для тестирования). В production следует обновлять сертификаты.
Обработка ошибок при потоковой загрузке файлов (stream=True)
При использовании stream=True возможны обрывы соединения во время чтения данных, что вызывает requests.exceptions.ChunkedEncodingError или ConnectionError.
import requests
from requests.exceptions import ChunkedEncodingError, ConnectionError
def download_large_file(url, filepath):
try:
with requests.get(url, stream=True, timeout=10) as r:
r.raise_for_status()
with open(filepath, 'wb') as f:
for chunk in r.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
except (ChunkedEncodingError, ConnectionError) as e:
print(f'Потоковая ошибка: {e}')
# Можно сохранить частично загруженный файл и повторить
return False
return True
# Пример с нестабильным сервером
success = download_large_file('https://httpbin.org/bytes/1024', 'test.bin')(если возникла ошибка): Потоковая ошибка: ('Connection broken: ...', ...)Пояснение: Обработка обрыва внутри итерационного чтения позволяет корректно завершить загрузку или сохранить частичные данные. Рекомендуется также использовать try-finally для закрытия файла.