Обработка ошибок в Python requests: защита от сетевых сбоев

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

try:
response = requests.get('https://api.example.com/data', timeout=3)
response.raise_for_status() # Вызовет HTTPError при 4XX/5XX статусах
data = response.json()
except HTTPError as http_err:
print(f"HTTP error occurred: {http_err}")
except ConnectionError as conn_err:
print(f"Connection error occurred: {conn_err}")
except Timeout as timeout_err:
print(f"Timeout error occurred: {timeout_err}")
except RequestException as req_err:
print(f"Unexpected error occurred: {req_err}")

import requests
from requests.exceptions import RequestException

try:
# Блок кода, где может произойти ошибка
response = requests.get('https://api.example.com/data')
response.raise_for_status() # Генерирует исключение при ошибках HTTP
data = response.json() # Может вызвать ValueError, если ответ не является JSON
except RequestException as e:
# Обработка ошибок, связанных с запросами
print(f"Произошла ошибка при запросе: {e}")
except ValueError as e:
# Обработка ошибок парсинга JSON
print(f"Ошибка при парсинге ответа: {e}")
except Exception as e:
# Обработка всех остальных исключений
print(f"Неожиданная ошибка: {e}")
else:
# Код, который выполняется, если исключений не было
print("Запрос успешно выполнен")
finally:
# Код, который выполняется всегда, независимо от наличия исключений
print("Завершение операции")

import requests
from requests.exceptions import HTTPError, ConnectionError, Timeout, TooManyRedirects, RequestException
import logging

logger = logging.getLogger(__name__)

def fetch_data_from_api(url, timeout=5):
try:
response = requests.get(url, timeout=timeout)
response.raise_for_status()
return response.json()
except HTTPError as http_err:
# Обработка ошибок HTTP-статусов
status_code = http_err.response.status_code
if status_code == 404:
logger.warning(f"Ресурс не найден: {url}")
return None
elif 400 <= status_code < 500:
logger.error(f"Клиентская ошибка: {http_err}")
raise
elif 500 <= status_code < 600:
logger.error(f"Серверная ошибка: {http_err}")
raise
except ConnectionError as conn_err:
logger.error(f"Ошибка соединения при запросе к {url}: {conn_err}")
raise
except Timeout as timeout_err:
logger.error(f"Тайм-аут при запросе к {url}: {timeout_err}")
raise
except TooManyRedirects as redirect_err:
logger.error(f"Слишком много перенаправлений при запросе к {url}: {redirect_err}")
raise
except RequestException as req_err:
logger.critical(f"Непредвиденная ошибка requests при запросе к {url}: {req_err}")
raise
except ValueError as val_err:
logger.error(f"Невалидный JSON в ответе от {url}: {val_err}")
raise

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

def request_with_retry(url, max_retries=3, backoff_factor=0.3):
"""Выполнить запрос с автоматическими повторными попытками при ошибках соединения."""
retries = 0
while retries <= max_retries:
try:
response = requests.get(url, timeout=10)
response.raise_for_status()
return response
except (ConnectionError, Timeout) as e:
retries += 1
if retries > max_retries:
raise Exception(f"Не удалось выполнить запрос после {max_retries} попыток: {e}")

# Экспоненциальная задержка между попытками
wait_time = backoff_factor * (2 ** (retries – 1))
print(f"Попытка {retries}/{max_retries} не удалась. Повторяем через {wait_time:.1f} сек...")
time.sleep(wait_time)
except RequestException as e:
# Другие ошибки requests не пытаемся автоматически исправить
raise Exception(f"Ошибка запроса: {e}")

import requests
from requests.exceptions import RequestException

class SafeSession:
def __init__(self, base_url=None, timeout=10):
self.session = requests.Session()
self.base_url = base_url
self.timeout = timeout

def __enter__(self):
return self

def __exit__(self, exc_type, exc_val, exc_tb):
self.session.close()

def get(self, url_path, **kwargs):
if self.base_url and not url_path.startswith(('http://', 'https://')):
url = f"{self.base_url.rstrip('/')}/{url_path.lstrip('/')}"
else:
url = url_path

kwargs.setdefault('timeout', self.timeout)

try:
response = self.session.get(url, **kwargs)
response.raise_for_status()
return response
except RequestException as e:
print(f"Ошибка при запросе {url}: {e}")
raise

# Использование
with SafeSession(base_url='https://api.example.com') as session:
try:
response = session.get('/users')
data = response.json()
print(data)
except Exception as e:
print(f"Произошла ошибка: {e}")

import requests
from requests.exceptions import HTTPError

def handle_response(url):
try:
response = requests.get(url)
response.raise_for_status()
return response.json()
except HTTPError as http_err:
status_code = http_err.response.status_code

# Клиентские ошибки
if status_code == 400:
print("Некорректный запрос")
elif status_code == 401:
print("Требуется аутентификация")
elif status_code == 403:
print("Доступ запрещен")
elif status_code == 404:
print(f"Ресурс не найден: {url}")
return None
elif status_code == 429:
print("Слишком много запросов. Попробуйте позже")

# Серверные ошибки
elif 500 <= status_code < 600:
print(f"Серверная ошибка: {status_code}")

# Логируем полное исключение
print(f"HTTPError: {http_err}")
raise

import functools
import time
import requests
from requests.exceptions import RequestException

def safe_request(max_retries=3, backoff_factor=0.5, timeout=10):
"""Декоратор для безопасного выполнения HTTP-запросов с повторными попытками."""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
retries = 0
while True:
try:
# Добавляем таймаут по умолчанию, если не указан явно
kwargs.setdefault('timeout', timeout)
return func(*args, **kwargs)
except (requests.ConnectionError, requests.Timeout) as e:
retries += 1
if retries > max_retries:
print(f"Достигнут лимит повторных попыток ({max_retries})")
raise

# Экспоненциальная задержка
wait = backoff_factor * (2 ** (retries – 1))
print(f"Попытка {retries}/{max_retries} не удалась из-за {e.__class__.__name__}. "
f"Повторная попытка через {wait:.1f} сек...")
time.sleep(wait)
except RequestException as e:
print(f"Ошибка запроса {e.__class__.__name__}: {e}")
raise
return wrapper
return decorator

# Пример использования
@safe_request(max_retries=3)
def fetch_user_data(user_id):
return requests.get(f'https://api.example.com/users/{user_id}')

import time
import requests
from requests.exceptions import RequestException

class CircuitBreaker:
def __init__(self, failure_threshold=5, recovery_timeout=30, name="default"):
self.failure_threshold = failure_threshold # Порог неудач
self.recovery_timeout = recovery_timeout # Время до восстановления (сек)
self.name = name
self.failures = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED, OPEN, HALF-OPEN

def __call__(self, func):
def wrapper(*args, **kwargs):
if self.state == "OPEN":
# Проверяем, прошло ли достаточно времени для восстановления
if time.time() – self.last_failure_time > self.recovery_timeout:
print(f"Circuit {self.name} переходит в состояние HALF-OPEN")
self.state = "HALF-OPEN"
else:
raise Exception(f"Circuit {self.name} разомкнут. Запрос не выполняется")

try:
result = func(*args, **kwargs)

# Если успешно и предохранитель был наполовину открыт, закрываем его
if self.state == "HALF-OPEN":
self.state = "CLOSED"
self.failures = 0
print(f"Circuit {self.name} успешно восстановлен")

return result

except RequestException as e:
self.failures += 1
self.last_failure_time = time.time()

# Если превысили порог или предохранитель уже наполовину открыт
if self.failures >= self.failure_threshold or self.state == "HALF-OPEN":
self.state = "OPEN"
print(f"Circuit {self.name} разомкнут после {self.failures} неудач")

raise

return wrapper

# Пример использования
weather_cb = CircuitBreaker(name="weather_api", failure_threshold=3, recovery_timeout=60)

@weather_cb
def get_weather(city):
response = requests.get(f"https://api.weather.example.com/{city}", timeout=5)
response.raise_for_status()
return response.json()

import requests
from requests.exceptions import RequestException

def get_data_with_fallback(primary_url, fallback_url, timeout=5):
"""Получение данных с использованием резервного URL при недоступности основного."""
try:
# Пробуем основной URL
response = requests.get(primary_url, timeout=timeout)
response.raise_for_status()
return response.json()
except RequestException as primary_err:
print(f"Первичный API недоступен: {primary_err}")
try:
# При ошибке используем резервный URL
print(f"Используем резервный API: {fallback_url}")
response = requests.get(fallback_url, timeout=timeout)
response.raise_for_status()
return response.json()
except RequestException as fallback_err:
print(f"Резервный API также недоступен: {fallback_err}")
# Оба API недоступны, возвращаем кешированные данные
return get_cached_data()

def get_cached_data():
"""Получение последних кешированных данных."""
# В реальном приложении здесь был бы код для извлечения данных из кеша
print("Использование кешированных данных")
return {"status": "cached", "data": "Last known good data"}

import time
import requests
from requests.exceptions import RequestException
import logging

# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class APIMonitor:
def __init__(self):
self.error_counts = {} # Счетчики ошибок по типам
self.response_times = [] # Времена ответов
self.status_counts = {} # Счетчики HTTP-статусов

def request(self, url, method='get', **kwargs):
"""Выполняет HTTP-запрос с мониторингом производительности и ошибок."""
start_time = time.time()
error_occurred = False

try:
response = getattr(requests, method.lower())(url, **kwargs)

# Учитываем статус-код
status = response.status_code
self.status_counts[status] = self.status_counts.get(status, 0) + 1

# Проверка на ошибочные статусы
response.raise_for_status()

return response

except RequestException as e:
error_occurred = True

# Учитываем тип ошибки
error_type = e.__class__.__name__
self.error_counts[error_type] = self.error_counts.get(error_type, 0) + 1

# Логируем информацию об ошибке
logger.error(f"API Error {error_type} при запросе к {url}: {str(e)}")

raise

finally:
# Измеряем время ответа
response_time = time.time() – start_time
self.response_times.append(response_time)

# Логируем метрики
logger.info(f"API запрос к {url}: время={response_time:.3f}с, "
f"успех={not error_occurred}")

def get_stats(self):
"""Возвращает текущую статистику запросов."""
avg_response_time = sum(self.response_times) / len(self.response_times) if self.response_times else 0

return {
'error_counts': self.error_counts,
'status_counts': self.status_counts,
'total_requests': len(self.response_times),
'avg_response_time': avg_response_time,
'success_rate': 1 – sum(self.error_counts.values()) / len(self.response_times) if self.response_times else 0
}

# Пример использования
api_monitor = APIMonitor()

try:
response = api_monitor.request('https://api.example.com/data', timeout=5)
data = response.json()
# Обработка данных...
except RequestException:
# Обработка ошибки...
pass

# Вывод статистики
print(api_monitor.get_stats())

import requests
from requests.exceptions import RequestException
from jsonschema import validate, ValidationError

# Схема для валидации JSON
user_schema = {
"type": "object",
"required": ["id", "name", "email"],
"properties": {
"id": {"type": "integer"},
"name": {"type": "string"},
"email": {"type": "string", "format": "email"},
"is_active": {"type": "boolean"}
}
}

def fetch_and_validate_user(user_id):
"""Получение данных пользователя с валидацией схемы JSON."""
try:
response = requests.get(f"https://api.example.com/users/{user_id}", timeout=5)
response.raise_for_status()

user_data = response.json()

# Валидация данных по схеме
try:
validate(instance=user_data, schema=user_schema)
return user_data
except ValidationError as e:
print(f"Полученные данные не соответствуют схеме: {e}")
# Возможно логирование или другая обработка ошибки валидации
raise ValueError(f"Невалидные данные пользователя: {e}")

except RequestException as e:
print(f"Ошибка HTTP-запроса: {e}")
raise