Исключения в запросах Python requests: как правильно их обрабатывать

Расширенные примеры обработки исключений в requests

1. Комбинированный перехват нескольких типов исключений

Как различать таймаут, ошибку соединения и HTTP-ошибку в одном запросе?

import requests
from requests.exceptions import Timeout, ConnectionError, HTTPError

try:
    r = requests.get('https://httpbin.org/delay/1', timeout=3)
    r.raise_for_status()
except Timeout:
    print('Таймаут: сервер не ответил вовремя.')
except ConnectionError:
    print('Ошибка соединения: недоступен хост.')
except HTTPError as e:
    print(f'HTTP-ошибка: {e.response.status_code}')
except requests.exceptions.RequestException as e:
    print(f'Другая ошибка: {e}')

Таймаут: сервер не ответил вовремя.

Порядок блоков except важен: сначала идут более конкретные исключения, затем базовое.

2. Повторные попытки с задержкой и обработкой исключений

Как реализовать автоматический повтор запроса при временных сбоях?

import requests
from requests.exceptions import ConnectionError, Timeout
import time

def request_with_retry(url, max_retries=3, delay=2):
    for attempt in range(1, max_retries + 1):
        try:
            r = requests.get(url, timeout=5)
            r.raise_for_status()
            return r
        except (ConnectionError, Timeout) as e:
            print(f'Попытка {attempt} не удалась: {e}')
            if attempt < max_retries:
                time.sleep(delay * attempt)  # увеличиваем задержку
            else:
                raise
    return None

try:
    response = request_with_retry('https://httpbin.org/status/503', max_retries=2)
    print('Успешно!', response.status_code)
except Exception as e:
    print(f'Запрос не удался после всех попыток: {e}')

Попытка 1 не удалась: 503 Server Error: SERVICE_UNAVAILABLE
Попытка 2 не удалась: 503 Server Error: SERVICE_UNAVAILABLE
Запрос не удался после всех попыток: 503 Server Error: SERVICE_UNAVAILABLE

Обратите внимание: raise_for_status() выбрасывает HTTPError, который не входит в перехватываемые исключения, поэтому при 503 выполняется except (ConnectionError, Timeout)? Нет, HTTPError не является подклассом ConnectionError или Timeout, поэтому он не будет перехвачен. В примере выше он упадёт. Правильная логика: если нужно повторять на 503, то нужно перехватывать HTTPError и проверять статус. Исправим:

import requests
from requests.exceptions import ConnectionError, Timeout, HTTPError

def request_with_retry_fixed(url, max_retries=3, delay=2):
    for attempt in range(1, max_retries + 1):
        try:
            r = requests.get(url, timeout=5)
            r.raise_for_status()
            return r
        except (ConnectionError, Timeout) as e:
            print(f'Сетевой сбой (попытка {attempt}): {e}')
        except HTTPError as e:
            if e.response.status_code in (503, 502, 504):  # временные ошибки сервера
                print(f'Временная ошибка сервера (попытка {attempt}): {e.response.status_code}')
            else:
                raise
        if attempt < max_retries:
            time.sleep(delay * attempt)
    raise Exception('Запрос не выполнен после всех попыток')

3. Логирование исключений без прерывания программы

Как записать ошибку в лог и продолжить выполнение?

import logging
import requests
from requests.exceptions import RequestException

logging.basicConfig(level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')

urls = ['https://httpbin.org/status/200', 'https://nonexistent.domain.test', 'https://httpbin.org/get']

for url in urls:
    try:
        r = requests.get(url, timeout=5)
        r.raise_for_status()
        print(f'{url}: успех (код {r.status_code})')
    except RequestException as e:
        logging.error(f'Ошибка при запросе {url}: {e}')

2025-04-08 12:00:00,000 - ERROR - Ошибка при запросе https://nonexistent.domain.test: ...
https://httpbin.org/status/200: успех (код 200)
https://httpbin.org/get: успех (код 200)

4. Использование контекстного менеджера для автоматического закрытия сессии

Как гарантировать закрытие сессии даже при исключении?

import requests
from requests.exceptions import RequestException

url = 'https://httpbin.org/delay/2'

try:
    with requests.Session() as session:
        resp = session.get(url, timeout=1)
except RequestException as e:
    print(f'Сессия автоматически закрыта после ошибки: {e}')

Выход из блока with гарантирует вызов session.close(), даже если возникло исключение.

5. Создание собственного класса исключений для обёртки ошибок requests

Как унифицировать ошибки в своём приложении?

import requests
from requests.exceptions import RequestException

class AppRequestError(Exception):
    def __init__(self, original_exception, url):
        self.original = original_exception
        self.url = url
        super().__init__(f'Ошибка при запросе к {url}: {original_exception}')

try:
    r = requests.get('https://httpbin.org/status/500')
    r.raise_for_status()
except RequestException as e:
    raise AppRequestError(e, 'https://httpbin.org/status/500') from e

При этом сохраняется цепочка исключений (с помощью from e), что помогает при отладке.

6. Обработка исключений при потоковой загрузке большого файла

Как безопасно загружать файл порциями и реагировать на обрыв соединения?

import requests
from requests.exceptions import ChunkedEncodingError, ConnectionError

url = 'https://httpbin.org/stream-bytes/2048'
local_filename = '/tmp/test.bin'

try:
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        with open(local_filename, 'wb') as f:
            for chunk in r.iter_content(chunk_size=256):
                if chunk:  # фильтровать keep-alive новые блоки
                    f.write(chunk)
except (ChunkedEncodingError, ConnectionError) as e:
    print(f'Поток прерван: {e}')
    # можно удалить частично скачанный файл
    import os
    if os.path.exists(local_filename):
        os.remove(local_filename)

7. Извлечение деталей исключения: ответ, запрос, статус

Как получить объект ответа или запроса из исключения?

import requests
from requests.exceptions import HTTPError

try:
    r = requests.get('https://httpbin.org/status/401')
    r.raise_for_status()
except HTTPError as e:
    # доступны атрибуты response и request
    print(f'Статус: {e.response.status_code}')
    print(f'Заголовки ответа: {e.response.headers}')
    print(f'Метод запроса: {e.request.method}')
    print(f'URL запроса: {e.request.url}')

Статус: 401
Заголовки ответа: ...
Метод запроса: GET
URL запроса: https://httpbin.org/status/401

8. Обработка исключений при использовании прокси и аутентификации

Как отличить ошибку аутентификации от сетевой проблемы с прокси?

import requests
from requests.exceptions import ProxyError, SSLError, ConnectionError

proxies = {
    'http': 'http://user:pass@proxy.example:8080',
    'https': 'http://user:pass@proxy.example:8080'
}

try:
    r = requests.get('https://api.github.com', proxies=proxies, timeout=10)
    r.raise_for_status()
except ProxyError:
    print('Ошибка прокси: неверный адрес или учётные данные.')
except SSLError:
    print('Ошибка SSL: проблема с сертификатом.')
except ConnectionError:
    print('Соединение не установлено (возможно, прокси недоступен).')

9. Поведение при отсутствии таймаута и его перехват

Что произойдёт, если не указать timeout и сервер завис?

import requests
from requests.exceptions import Timeout

# Опасно: нет таймаута
try:
    r = requests.get('https://httpbin.org/delay/100')  # зависнет на 100 секунд
    print(r.status_code)
except Timeout:
    print('Таймаут (никогда не сработает, если timeout не задан).')
except requests.exceptions.ConnectionError:
    print('Соединение разорвано (если сервер закрыл соединение).')

Без параметра timeout запрос может ожидать ответа неограниченно долго. Исключение Timeout не возникнет. Рекомендуется всегда задавать timeout.

10. Обработка исключений в асинхронных запросах (aiohttp как альтернатива)

Как работать с ошибками в асинхронном коде (если заменить requests на aiohttp)?

Хотя это выходит за рамки requests, для полноты: в aiohttp исключения аналогичны, но обрабатываются в асинхронных функциях:

import aiohttp
import asyncio

async def fetch(session, url):
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as resp:
            return await resp.text()
    except aiohttp.ClientConnectorError as e:
        print(f'Ошибка соединения: {e}')
    except asyncio.TimeoutError:
        print('Таймаут')

async def main():
    async with aiohttp.ClientSession() as session:
        await fetch(session, 'https://httpbin.org/get')

asyncio.run(main())