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

try:
risky_operation()
except Exception as e:
print(f"Произошла ошибка: {e}")

try:
number = int(input("Введите число: "))
result = 100 / number
print(result)
except ValueError:
print("Введено не число")
except ZeroDivisionError:
print("Деление на ноль недопустимо")
except Exception as e:
print(f"Непредвиденная ошибка: {e}")
else:
print("Операция выполнена успешно")
finally:
print("Блок завершен")

import traceback

def function_with_error():
x = 1 / 0

try:
function_with_error()
except Exception as e:
error_traceback = traceback.format_exc()
print("Произошла ошибка:")
print(error_traceback)
# Здесь можно сохранить трассировку в лог или БД

import traceback

try:
1 / 0
except Exception:
# Получаем текущий тип исключения, значение и объект трассировки
exc_type, exc_value, exc_tb = traceback.sys.exc_info()

# Извлекаем фреймы стека
tb_frames = traceback.extract_tb(exc_tb)

# Выводим информацию по каждому фрейму
for frame in tb_frames:
filename, line, func, text = frame
print(f"Файл: {filename}, строка {line}, в функции {func}")
print(f" Код: {text}")

import traceback
import sys

try:
1 / 0
except Exception:
exc_type, exc_value, exc_tb = sys.exc_info()
exception_details = traceback.format_exception(exc_type, exc_value, exc_tb)
formatted_exception = ''.join(exception_details)
print(formatted_exception)

import traceback
import threading

try:
risky_operation()
except Exception as e:
thread_info = f"Thread: {threading.current_thread().name}"
error_info = f"{thread_info}\n{traceback.format_exc()}"
print(error_info)

def with_context(func, context_info):
"""Декоратор для добавления контекстной информации к исключениям."""
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
error_msg = f"Контекст: {context_info}\n{traceback.format_exc()}"
logger.error(error_msg)
raise # Пробрасываем исключение дальше
return wrapper

# Использование
process_user_data = with_context(process_user_data, "Обработка пользовательских данных")

import logging

# Настройка базового логгера
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s',
handlers=[
logging.FileHandler("app.log"),
logging.StreamHandler()
]
)

# Получение логгера для текущего модуля
logger = logging.getLogger(__name__)

try:
x = 1 / 0
except Exception as e:
# exc_info=True добавляет трассировку исключения
logger.error(f"Произошла ошибка: {e}", exc_info=True)

import logging

# Настройка корневого логгера
root_logger = logging.getLogger()
root_logger.setLevel(logging.WARNING)

# Файловый обработчик для всех сообщений
file_handler = logging.FileHandler("app.log")
file_handler.setFormatter(logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s'))
root_logger.addHandler(file_handler)

# Обработчик для критических ошибок
error_handler = logging.FileHandler("errors.log")
error_handler.setLevel(logging.ERROR)
error_handler.setFormatter(logging.Formatter('%(asctime)s – %(name)s – %(levelname)s – %(message)s'))
root_logger.addHandler(error_handler)

# Логгер для конкретного модуля
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG) # Более детальный уровень для этого модуля

import logging.config

# Конфигурация логирования
logging_config = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'standard': {
'format': '%(asctime)s – %(name)s – %(levelname)s – %(message)s'
},
'detailed': {
'format': '%(asctime)s – %(name)s – %(levelname)s – %(pathname)s:%(lineno)d – %(message)s'
},
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'INFO',
'formatter': 'standard',
},
'file': {
'class': 'logging.FileHandler',
'level': 'DEBUG',
'formatter': 'standard',
'filename': 'app.log',
},
'error_file': {
'class': 'logging.FileHandler',
'level': 'ERROR',
'formatter': 'detailed',
'filename': 'error.log',
},
},
'loggers': {
'': { # корневой логгер
'handlers': ['console', 'file', 'error_file'],
'level': 'DEBUG',
'propagate': True
},
'api': { # логгер для API-модуля
'handlers': ['file', 'error_file'],
'level': 'INFO',
'propagate': False
},
}
}

# Применяем конфигурацию
logging.config.dictConfig(logging_config)

# Получаем логгер
logger = logging.getLogger(__name__)

class RequestContextFilter(logging.Filter):
"""Добавляет информацию о запросе к логам."""

def __init__(self, request=None):
super().__init__()
self.request = request

def filter(self, record):
if self.request:
record.user_id = getattr(self.request, 'user_id', 'anonymous')
record.request_id = getattr(self.request, 'id', 'unknown')
return True

# Использование
logger = logging.getLogger(__name__)
logger.addFilter(RequestContextFilter(current_request))

import functools
import logging
import traceback

logger = logging.getLogger(__name__)

def exception_handler(fallback_return=None, reraise=False):
"""
Декоратор для перехвата и логирования исключений.

Args:
fallback_return: Значение, возвращаемое при ошибке
reraise: Пробрасывать исключение дальше после логирования
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
func_args = ", ".join([
repr(arg) for arg in args
] + [
f"{k}={repr(v)}" for k, v in kwargs.items()
])

error_msg = f"Ошибка в {func.__name__}({func_args}): {str(e)}"
logger.error(error_msg, exc_info=True)

if reraise:
raise
return fallback_return
return wrapper
return decorator

# Использование
@exception_handler(fallback_return=[], reraise=False)
def fetch_user_data(user_id):
# Потенциально опасный код
return db.query(f"SELECT * FROM users WHERE id = {user_id}")

import time
import random
import logging

logger = logging.getLogger(__name__)

def retry(max_attempts=3, backoff_factor=0.5, allowed_exceptions=(Exception,)):
"""
Декоратор для повторных попыток выполнения функции при ошибке.

Args:
max_attempts: Максимальное количество попыток
backoff_factor: Фактор для экспоненциального увеличения задержки
allowed_exceptions: Типы исключений, для которых делать повторы
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
attempt = 0
while attempt < max_attempts:
try:
return func(*args, **kwargs)
except allowed_exceptions as e:
attempt += 1
if attempt == max_attempts:
logger.error(f"Все попытки исчерпаны для {func.__name__}: {e}")
raise

# Экспоненциальная задержка с случайным фактором
delay = backoff_factor * (2 ** (attempt – 1)) * (0.5 + random.random())
logger.warning(
f"Попытка {attempt} для {func.__name__} не удалась: {e}. "
f"Повтор через {delay:.2f} сек."
)
time.sleep(delay)
return wrapper
return decorator

# Использование
@retry(max_attempts=5, backoff_factor=1.0, allowed_exceptions=(ConnectionError, TimeoutError))
def fetch_remote_data(url):
# Потенциально нестабильная операция
return requests.get(url, timeout=5).json()

import time
import logging
from enum import Enum

class CircuitState(Enum):
CLOSED = "CLOSED" # Нормальное состояние
OPEN = "OPEN" # Блокировка операций
HALF_OPEN = "HALF_OPEN" # Проверка на восстановление

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.state = CircuitState.CLOSED
self.failure_count = 0
self.last_failure_time = 0
self.logger = logging.getLogger(f"circuit_breaker.{name}")

def __call__(self, func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if self.state == CircuitState.OPEN:
# Проверка времени восстановления
if time.time() – self.last_failure_time >= self.recovery_timeout:
self.logger.info(f"Circuit {self.name} переходит в HALF_OPEN")
self.state = CircuitState.HALF_OPEN
else:
raise CircuitBreakerError(f"Circuit {self.name} открыт")

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

# Успешный вызов в HALF_OPEN возвращает в CLOSED
if self.state == CircuitState.HALF_OPEN:
self.logger.info(f"Circuit {self.name} восстановлен")
self.reset()

return result
except Exception as e:
self.record_failure()
raise

return wrapper

def record_failure(self):
self.failure_count += 1
self.last_failure_time = time.time()

if self.state == CircuitState.CLOSED and self.failure_count >= self.failure_threshold:
self.logger.warning(f"Circuit {self.name} открыт после {self.failure_count} ошибок")
self.state = CircuitState.OPEN

def reset(self):
self.failure_count = 0
self.state = CircuitState.CLOSED

class CircuitBreakerError(Exception):
"""Исключение, возникающее при открытом Circuit Breaker."""
pass

# Использование
payment_circuit = CircuitBreaker(
failure_threshold=3,
recovery_timeout=60,
name="payment_service"
)

@payment_circuit
def process_payment(payment_data):
# Вызов внешнего платежного сервиса
return payment_api.charge(payment_data)

def safe_operation(operation, args=None, kwargs=None, fallback=None, log_error=True):
"""
Безопасно выполняет операцию, возвращая fallback при ошибке.

Args:
operation: Функция для выполнения
args: Позиционные аргументы для функции
kwargs: Именованные аргументы для функции
fallback: Значение, возвращаемое при ошибке
log_error: Логировать ли ошибку
"""
args = args or ()
kwargs = kwargs or {}

try:
return operation(*args, **kwargs)
except Exception as e:
if log_error:
logger.error(f"Ошибка при выполнении {operation.__name__}: {e}", exc_info=True)

# Если fallback – функция, вызываем ее
if callable(fallback):
return fallback(*args, **kwargs)
return fallback

# Использование
def get_cached_data():
# Запасная функция, которая возвращает кешированные данные
return cache.get('user_recommendations', [])

user_recommendations = safe_operation(
fetch_recommendations,
args=(user_id,),
fallback=get_cached_data,
log_error=True
)

class ContextualException(Exception):
"""Исключение с расширенным контекстом для отладки."""

def __init__(self, message, context=None):
super().__init__(message)
self.message = message
self.context = context or {}
self.timestamp = datetime.now()

def __str__(self):
context_str = ", ".join(f"{k}={v}" for k, v in self.context.items())
return f"{self.message} [Context: {context_str}]"

def to_dict(self):
"""Представление исключения в виде словаря для логирования."""
return {
"error_type": self.__class__.__name__,
"message": self.message,
"timestamp": self.timestamp.isoformat(),
"context": self.context
}

# Специализированные типы исключений
class ValidationError(ContextualException):
"""Ошибка валидации данных."""
pass

class ServiceUnavailableError(ContextualException):
"""Ошибка доступа к внешнему сервису."""
pass

# Использование
def process_user_input(user_input):
if not validate_input(user_input):
raise ValidationError(
"Некорректные входные данные",
context={
"input": user_input,
"validation_rules": get_validation_rules(),
"user_id": get_current_user_id()
}
)

class ExceptionMiddleware:
"""Middleware для централизованной обработки исключений в веб-приложении."""

def __init__(self, app):
self.app = app
self.logger = logging.getLogger("exception.middleware")

async def __call__(self, scope, receive, send):
try:
await self.app(scope, receive, send)
except Exception as e:
# Логирование исключения
request_id = scope.get("request_id", "unknown")
self.logger.error(
f"Необработанное исключение в запросе {request_id}",
exc_info=True,
extra={"scope": self.extract_scope_info(scope)}
)

# Отправка уведомления при критических ошибках
if self.is_critical_error(e):
self.send_alert(e, request_id, scope)

# Формирование ответа об ошибке
response = self.format_error_response(e)
await self.send_response(send, response)

def extract_scope_info(self, scope):
"""Извлекает безопасную для логирования информацию из scope."""
# Реализация зависит от фреймворка
pass

def is_critical_error(self, exception):
"""Определяет, является ли исключение критическим."""
return isinstance(exception, (DatabaseError, CriticalServiceError))

def send_alert(self, exception, request_id, scope):
"""Отправляет уведомление о критической ошибке."""
# Реализация зависит от системы мониторинга
pass

def format_error_response(self, exception):
"""Форматирует HTTP-ответ на основе типа исключения."""
if isinstance(exception, ValidationError):
return {"status": 400, "body": {"error": str(exception)}}
if isinstance(exception, AuthenticationError):
return {"status": 401, "body": {"error": "Unauthorized"}}
# Для всех остальных исключений
return {"status": 500, "body": {"error": "Internal Server Error"}}

async def send_response(self, send, response):
"""Отправляет сформированный ответ клиенту."""
# Реализация зависит от фреймворка
pass

# Использование (для FastAPI)
app = FastAPI()
app.add_middleware(ExceptionMiddleware)

import json
import logging
import traceback
import sys

class StructuredExceptionFormatter(logging.Formatter):
"""Форматирует исключения в JSON для машинной обработки."""

def format(self, record):
log_data = {
"timestamp": self.formatTime(record, self.datefmt),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}

# Добавляем дополнительные поля
for key, value in record.__dict__.items():
if key not in log_data and not key.startswith("_"):
log_data[key] = value

# Обрабатываем исключение, если оно есть
if record.exc_info:
exc_type, exc_value, exc_traceback = record.exc_info
log_data["exception"] = {
"type": exc_type.__name__,
"message": str(exc_value),
"traceback": self.format_traceback(exc_traceback),
}

return json.dumps(log_data)

def format_traceback(self, tb):
"""Преобразует трассировку в структурированный формат."""
frames = []
current = tb
while current:
frame_info = {
"filename": current.tb_frame.f_code.co_filename,
"name": current.tb_frame.f_code.co_name,
"lineno": current.tb_lineno,
}
# Добавляем локальные переменные (осторожно с конфиденциальными данными)
if hasattr(current.tb_frame, "f_locals"):
safe_locals = self.sanitize_dict(current.tb_frame.f_locals)
if safe_locals:
frame_info["locals"] = safe_locals

frames.append(frame_info)
current = current.tb_next

return frames

def sanitize_dict(self, data):
"""Очищает словарь от конфиденциальных данных и несериализуемых объектов."""
sensitive_keys = ["password", "token", "secret", "key"]
result = {}

for k, v in data.items():
# Пропускаем приватные атрибуты и функции
if k.startswith("_") or callable(v):
continue

# Маскируем чувствительные данные
if any(sensitive in k.lower() for sensitive in sensitive_keys):
result[k] = "********"
else:
# Преобразуем несериализуемые объекты в строки
try:
json.dumps({k: v})
result[k] = v
except (TypeError, OverflowError):
result[k] = str(v)

return result

# Настройка логгера для структурированного вывода
structured_handler = logging.FileHandler("structured_exceptions.json")
structured_handler.setFormatter(StructuredExceptionFormatter())
logger = logging.getLogger("structured")
logger.addHandler(structured_handler)
logger.setLevel(logging.DEBUG)

def process_with_cause_tracking():
try:
try:
result = fetch_data()
process_data(result)
except ConnectionError as e:
# Создаем новое исключение с оригинальным в качестве причины
raise ServiceUnavailableError(
"Не удалось получить данные из внешнего сервиса",
context={"service": "data_provider"}
) from e
except ServiceUnavailableError as e:
# Логируем всю цепочку исключений
logger.error(f"Ошибка обработки: {e}")

# Извлекаем и анализируем причину
cause = e.__cause__
if cause:
logger.error(f"Вызвано исключением: {cause}")

# В Python также есть __context__ для неявных причин
if hasattr(e, '__context__') and e.__context__:
logger.error(f"Контекст исключения: {e.__context__}")

import functools
import logging
import traceback
import json
import requests

class MonitoringClient:
"""Клиент для отправки данных об исключениях в систему мониторинга."""

def __init__(self, endpoint, api_key, environment="production"):
self.endpoint = endpoint
self.api_key = api_key
self.environment = environment
self.logger = logging.getLogger("monitoring")

def report_exception(self, exc_type, exc_value, exc_traceback, context=None):
"""Отправляет информацию об исключении в систему мониторинга."""
try:
payload = {
"exception_type": exc_type.__name__,
"message": str(exc_value),
"traceback": traceback.format_tb(exc_traceback),
"environment": self.environment,
"context": context or {},
"timestamp": datetime.now().isoformat(),
}

headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}

response = requests.post(
f"{self.endpoint}/api/exceptions",
data=json.dumps(payload),
headers=headers,
timeout=2 # Короткий таймаут, чтобы не блокировать приложение
)

if response.status_code != 200:
self.logger.warning(
f"Не удалось отправить исключение в мониторинг: {response.status_code}"
)
except Exception as e:
self.logger.error(f"Ошибка при отправке в мониторинг: {e}")

# Создание клиента мониторинга
monitor = MonitoringClient(
endpoint="https://monitoring.example.com",
api_key="your_api_key",
environment="production"
)

# Декоратор для отслеживания исключений
def monitored(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
monitor.report_exception(
exc_type, exc_value, exc_traceback,
context={"function": func.__name__, "args": args, "kwargs": kwargs}
)
raise
return wrapper

# Использование
@monitored
def critical_business_function(user_id, transaction_data):
# Важная бизнес-логика
pass