Python · API · Продвинутый уровень

Обработка ошибок

5 задач

Надёжно обрабатывайте HTTP-ошибки, сетевые сбои и таймауты в API-скриптах.

Краткое введение в тему и пояснения перед упражнениями (упражнения ниже):

Обработка ошибок в requests

#
Реальные скрипты для работы с API дают сбои. Сервер может вернуть 404, сеть может оборваться, запрос может превысить время ожидания. Явная обработка этих случаев делает скрипты надёжными. Есть два типа сбоев: 1. **HTTP-ошибки** -- сервер отвечает, но с кодом ошибки (4xx или 5xx). По умолчанию `requests` не бросает исключение -- он возвращает ответ, и вызов `.json()` может вернуть payload ошибки вместо данных. 2. **Сетевые ошибки** -- запрос вообще не доходит до сервера или сервер не отвечает. Эти ошибки бросают исключения: ConnectionError, Timeout или базовый RequestException. `raise_for_status()` устраняет пробел для HTTP-ошибок -- бросает HTTPError при статусе 4xx или 5xx: ```python import requests response = requests.get('https://apilearn.tukas.dev/api/products/99999/') response.raise_for_status() # бросает HTTPError при 404 data = response.json() # выполняется только если статус 2xx ``` Для сетевых ошибок используйте try/except: ```python try: response = requests.get(url, timeout=5) response.raise_for_status() data = response.json() except requests.exceptions.Timeout: print('Запрос превысил время ожидания') except requests.exceptions.ConnectionError: print('Сетевая ошибка -- не удалось связаться с сервером') except requests.exceptions.HTTPError as e: print(f'HTTP ошибка: {e.response.status_code}') ``` Параметр `timeout=` устанавливает лимит времени в секундах. Без него зависший сервер может остановить скрипт навсегда: ```python # Бросает Timeout если нет ответа в течение 5 секунд response = requests.get(url, timeout=5) # Отдельные таймауты для подключения и чтения response = requests.get(url, timeout=(3, 10)) # (connect, read) ``` Иерархия исключений requests (от конкретных к общим): ``` RequestException Базовый класс для всех исключений requests ConnectionError Не удалось связаться с сервером ProxyError Ошибка подключения через прокси Timeout Запрос превысил время ожидания ConnectTimeout Таймаут подключения ReadTimeout Таймаут чтения HTTPError Статус 4xx или 5xx (raise_for_status) TooManyRedirects Цикл перенаправлений ``` Перехват `RequestException` обрабатывает все типы сразу -- полезно для простых скриптов, где различие между типами ошибок не важно.

Практические паттерны обработки ошибок

#
Практические приёмы обработки ошибок API: Базовая обёртка, которая превращает ошибки в понятные сообщения: ```python import requests def safe_get(url, **kwargs): try: response = requests.get(url, timeout=5, **kwargs) response.raise_for_status() return response.json() except requests.exceptions.Timeout: print(f'Ошибка: запрос к {url} превысил время ожидания') except requests.exceptions.ConnectionError: print(f'Ошибка: не удалось подключиться к {url}') except requests.exceptions.HTTPError as e: print(f'HTTP ошибка: {e.response.status_code}') return None ``` Отдельный перехват 401 для сообщения о проблеме с авторизацией: ```python try: response = requests.get(url, headers=headers) response.raise_for_status() except requests.exceptions.HTTPError as e: if e.response.status_code == 401: print('Ошибка авторизации — проверьте токен') else: print(f'HTTP ошибка: {e.response.status_code}') ``` Retry с exponential backoff для временных сбоев (429, 503): ```python import time def retry_get(url, retries=3, **kwargs): for attempt in range(retries): try: response = requests.get(url, timeout=5, **kwargs) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: if e.response.status_code in (429, 503) and attempt < retries - 1: wait = 2 ** attempt # 1s, 2s, 4s print(f'Попытка {attempt + 1} неудачна, ждём {wait}s...') time.sleep(wait) else: raise return None ``` Exponential backoff означает: ждать 1 секунду после первой неудачи, 2 секунды после второй, 4 секунды после третьей — так мы не мешаем серверу, который перегружен.

Справочник обработки ошибок

#
Справочник по обработке ошибок: **raise_for_status()** ```python response.raise_for_status() # Бросает HTTPError при статусе 4xx или 5xx # Ничего не делает при 2xx или 3xx # e.response.status_code — код статуса в блоке except ``` **timeout=** ```python requests.get(url, timeout=5) # 5 секунд для подключения + ответа requests.get(url, timeout=(3, 10)) # (подключение, чтение) отдельно # Бросает requests.exceptions.Timeout ``` **Иерархия исключений** ``` RequestException — перехватывает любую ошибку requests ├── ConnectionError — сетевая ошибка, DNS ├── Timeout — превышено время ожидания └── HTTPError — 4xx/5xx от raise_for_status() ``` **Канонический шаблон try/except** ```python try: response = requests.get(url, timeout=5) response.raise_for_status() data = response.json() except requests.exceptions.Timeout: print('Превышено время ожидания') except requests.exceptions.ConnectionError: print('Нет соединения') except requests.exceptions.HTTPError as e: print(f'HTTP {e.response.status_code}') ``` Порядок `except` важен: более специфичные классы ставьте выше — `Timeout` и `ConnectionError` перед `RequestException`.

Упражнения:

01

Использование raise_for_status()

#

Сделайте GET к /api/products/99999/ (несуществующий продукт). Без raise_for_status() выведите код статуса. Затем добавьте raise_for_status() и обработайте исключение HTTPError, выведя сообщение об ошибке.

import requests

BASE_URL = 'https://apilearn.tukas.dev'

# Часть 1: без raise_for_status — проверьте код статуса
response = requests.get(f'{BASE_URL}/api/products/99999/')
print(response.status_code)

# Часть 2: с raise_for_status и обработкой HTTPError
Решение 
import requests

BASE_URL = 'https://apilearn.tukas.dev'

# Часть 1: без raise_for_status
response = requests.get(f'{BASE_URL}/api/products/99999/')
print(response.status_code)  # 404

# Часть 2: с raise_for_status
try:
    response = requests.get(f'{BASE_URL}/api/products/99999/')
    response.raise_for_status()
except requests.exceptions.HTTPError as e:
    print(f'HTTP ошибка: {e.response.status_code}')
02

Обработка ошибки авторизации

#

Попробуйте получить /api/users/profile/ БЕЗ заголовка авторизации. Обработайте HTTPError и выведите специальное сообщение, если статус 401.

import requests

BASE_URL = 'https://apilearn.tukas.dev'

# GET профиля без авторизации — обработайте 401 отдельно
try:
    response = requests.get(f'{BASE_URL}/api/users/profile/')
    response.raise_for_status()
except requests.exceptions.HTTPError as e:
    # Выведите разные сообщения для 401 и других кодов
    pass
Решение 
import requests

BASE_URL = 'https://apilearn.tukas.dev'

try:
    response = requests.get(f'{BASE_URL}/api/users/profile/')
    response.raise_for_status()
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 401:
        print('Ошибка авторизации — предоставьте действительный токен')
    else:
        print(f'HTTP ошибка: {e.response.status_code}')
03

Добавить timeout к запросу

#

Сделайте GET к /api/products/ с timeout=5. Обработайте Timeout и ConnectionError отдельно. Выведите список продуктов при успехе или сообщение об ошибке.

import requests

BASE_URL = 'https://apilearn.tukas.dev'

# GET с timeout и обработкой сетевых ошибок
Решение 
import requests

BASE_URL = 'https://apilearn.tukas.dev'

try:
    response = requests.get(f'{BASE_URL}/api/products/', timeout=5)
    response.raise_for_status()
    products = response.json()['results']
    print(f'Получено {len(products)} продуктов')
except requests.exceptions.Timeout:
    print('Ошибка: запрос превысил время ожидания')
except requests.exceptions.ConnectionError:
    print('Ошибка: не удалось подключиться к API')
except requests.exceptions.HTTPError as e:
    print(f'HTTP ошибка: {e.response.status_code}')
04

Функция safe_get()

#

Напишите функцию safe_get(url, **kwargs), которая делает GET с timeout=5, вызывает raise_for_status() и возвращает JSON при успехе или None при любой ошибке (выводя сообщение). Протестируйте с валидным и несуществующим URL.

import requests

BASE_URL = 'https://apilearn.tukas.dev'

def safe_get(url, **kwargs):
    # Реализуйте с timeout, raise_for_status и обработкой всех ошибок
    pass

# Тест 1: валидный URL
result = safe_get(f'{BASE_URL}/api/products/')
if result:
    print(f'Продуктов: {len(result["results"])}')

# Тест 2: несуществующий эндпоинт
result = safe_get(f'{BASE_URL}/api/products/99999/')
print(result)
Решение 
import requests

BASE_URL = 'https://apilearn.tukas.dev'

def safe_get(url, **kwargs):
    try:
        response = requests.get(url, timeout=5, **kwargs)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.Timeout:
        print(f'Ошибка: timeout при запросе к {url}')
    except requests.exceptions.ConnectionError:
        print(f'Ошибка: нет соединения с {url}')
    except requests.exceptions.HTTPError as e:
        print(f'HTTP {e.response.status_code} для {url}')
    return None

result = safe_get(f'{BASE_URL}/api/products/')
if result:
    print(f'Продуктов: {len(result["results"])}')

result = safe_get(f'{BASE_URL}/api/products/99999/')
print(result)
05

Retry с backoff

#

Напишите функцию retry_get(url, retries=3), которая повторяет запрос при кодах 429 или 503, ожидая 2**attempt секунд между попытками. При других ошибках — сразу бросать исключение. Выведите сообщение перед каждым ожиданием.

import requests
import time

BASE_URL = 'https://apilearn.tukas.dev'

def retry_get(url, retries=3, **kwargs):
    # Повторяйте при 429/503, exponential backoff, бросайте при других ошибках
    pass

# Тест с рабочим эндпоинтом (не активирует retry)
result = retry_get(f'{BASE_URL}/api/products/')
print(result['count'])
Решение 
import requests
import time

BASE_URL = 'https://apilearn.tukas.dev'

def retry_get(url, retries=3, **kwargs):
    for attempt in range(retries):
        try:
            response = requests.get(url, timeout=5, **kwargs)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code in (429, 503) and attempt < retries - 1:
                wait = 2 ** attempt
                print(f'Попытка {attempt + 1} неудачна ({e.response.status_code}), ждём {wait}s...')
                time.sleep(wait)
            else:
                raise
    return None

result = retry_get(f'{BASE_URL}/api/products/')
print(result['count'])