Создаем кастомные исключения в Python: полное руководство для разработчиков

class MyCustomException(Exception):
pass

def process_data(data):
if not data:
raise MyCustomException("Данные отсутствуют")
# Дальнейшая обработка данных

try:
process_data([])
except MyCustomException as e:
print(f"Произошла ошибка: {e}")

class CustomException(BaseClassException):
"""Документация для вашего исключения."""

def __init__(self, message="", *args, **kwargs):
# Вызов конструктора родительского класса
super().__init__(message, *args)
# Дополнительные атрибуты и логика

class InvalidFileFormatError(ValueError):
"""Исключение, возникающее при обработке файла с некорректным форматом."""

def __init__(self, filename, expected_format, *args):
self.filename = filename
self.expected_format = expected_format
message = f"Файл '{filename}' имеет некорректный формат. Ожидается: {expected_format}"
super().__init__(message, *args)

# Базовое исключение для всего приложения
class AppError(Exception):
"""Базовый класс для всех исключений приложения."""

# Исключения для модуля обработки данных
class DataError(AppError):
"""Базовое исключение для ошибок обработки данных."""

class DataFormatError(DataError):
"""Исключение для ошибок формата данных."""

class DataValidationError(DataError):
"""Исключение для ошибок валидации данных."""

# Исключения для модуля сетевого взаимодействия
class NetworkError(AppError):
"""Базовое исключение для сетевых ошибок."""

class ConnectionTimeoutError(NetworkError):
"""Исключение при превышении времени ожидания соединения."""

class ApplicationError(Exception):
"""Базовое исключение для всех ошибок приложения."""

def __init__(self, message="", error_code=None, *args):
self.error_code = error_code
super().__init__(message, *args)

def __str__(self):
if self.error_code:
return f"[Код {self.error_code}] {super().__str__()}"
return super().__str__()

class DatabaseError(ApplicationError):
"""Исключения, связанные с работой базы данных."""

class ValidationError(ApplicationError):
"""Исключения, связанные с валидацией данных."""

class AuthenticationError(ApplicationError):
"""Исключения, связанные с аутентификацией."""

class ConnectionPoolExhaustedError(DatabaseError):
"""Возникает, когда пул соединений базы данных исчерпан."""

def __init__(self, pool_size, wait_time=None):
self.pool_size = pool_size
self.wait_time = wait_time
message = f"Пул соединений (размер {pool_size}) исчерпан"
if wait_time:
message += f". Время ожидания: {wait_time} сек."
super().__init__(message, error_code="DB-001")

class InvalidCredentialsError(AuthenticationError):
"""Возникает при попытке аутентификации с неверными учетными данными."""

def __init__(self, username, attempts_left=None):
self.username = username
self.attempts_left = attempts_left
message = f"Неверные учетные данные для пользователя '{username}'"
if attempts_left is not None:
message += f". Осталось попыток: {attempts_left}"
super().__init__(message, error_code="AUTH-002")

class ValidationError(ApplicationError):
"""Исключения, связанные с валидацией данных."""

def __init__(self, message="", field=None, value=None, constraints=None):
self.field = field
self.value = value
self.constraints = constraints or {}
error_details = ""
if field:
error_details += f" Поле: {field}."
if value is not None:
error_details += f" Значение: {value}."
if constraints:
constraints_str = ", ".join(f"{k}={v}" for k, v in constraints.items())
error_details += f" Ограничения: {constraints_str}."

full_message = message + error_details
super().__init__(full_message, error_code="VAL-001")

def to_dict(self):
"""Преобразует исключение в словарь для API-ответов."""
return {
"error_code": self.error_code,
"message": str(self),
"field": self.field,
"constraints": self.constraints
}

def authenticate_user(username, password):
user = get_user_by_username(username)
if not user:
raise InvalidCredentialsError(username)

if not check_password(user, password):
attempts_left = get_attempts_left(user)
raise InvalidCredentialsError(username, attempts_left)

return user

def validate_user_data(user_data):
if not 'email' in user_data:
raise ValidationError("Отсутствует обязательное поле", field="email")

email = user_data['email']
if not is_valid_email(email):
raise ValidationError(
"Некорректный формат email", 
field="email",
value=email,
constraints={"format": "username@domain.tld"}
)

try:
user = authenticate_user(username, password)
user_data = get_user_data(user_id)
validate_user_data(user_data)
process_user_data(user_data)
except InvalidCredentialsError as e:
log.warning(f"Попытка входа с неверными учетными данными: {e}")
return render_login_page(error_message=str(e))
except ValidationError as e:
log.error(f"Ошибка валидации данных: {e}")
return json_response({"status": "error", "details": e.to_dict()})
except DatabaseError as e:
log.critical(f"Ошибка базы данных: {e}")
notify_admin(e)
return render_error_page(500, "Внутренняя ошибка сервера")
except ApplicationError as e:
log.error(f"Ошибка приложения: {e}")
return render_error_page(400, str(e))

class ResourceNotFoundError(Exception):
"""Исключение, возникающее при попытке доступа к несуществующему ресурсу."""

def __init__(self, resource_type, resource_id, message=None):
self.resource_type = resource_type
self.resource_id = resource_id

if message is None:
message = f"{resource_type} с идентификатором {resource_id} не найден"

super().__init__(message)

def get_user(user_id):
user = database.find_user(user_id)
if user is None:
raise ResourceNotFoundError("User", user_id)
return user

try:
user = get_user(123)
except ResourceNotFoundError as e:
print(f"Ошибка: {e}")
print(f"Тип ресурса: {e.resource_type}, ID: {e.resource_id}")

class ValidationError(Exception):
"""Исключение при ошибках валидации данных."""

def __init__(self, field, value, constraints=None, message=None):
self.field = field
self.value = value
self.constraints = constraints or {}

if message is None:
message = f"Ошибка валидации поля '{field}'"

self.validation_errors = {
"field": field,
"received_value": value,
"constraints": self.constraints
}

super().__init__(message)

def as_dict(self):
"""Возвращает информацию об ошибке в виде словаря."""
return {
"error": str(self),
"validation_details": self.validation_errors
}

@app.route('/api/users', methods=['POST'])
def create_user():
try:
data = request.get_json()
validate_email(data.get('email'))
# Продолжение обработки...
except ValidationError as e:
return jsonify(e.as_dict()), 400

import traceback
import logging

logger = logging.getLogger(__name__)

class DatabaseError(Exception):
"""Исключение для ошибок взаимодействия с базой данных."""

def __init__(self, operation, details=None, query=None):
self.operation = operation
self.details = details
self.query = query
self.traceback = traceback.format_exc()

message = f"Ошибка базы данных при выполнении операции '{operation}'"
if details:
message += f": {details}"

super().__init__(message)

# Автоматическое журналирование ошибки
logger.error(
f"DatabaseError: {message}\n"
f"Query: {query}\n"
f"Stack trace: {self.traceback}"
)

class context_info:
"""Контекстный менеджер для добавления информации к исключениям."""

def __init__(self, **context):
self.context = context

def __enter__(self):
return self

def __exit__(self, exc_type, exc_val, exc_tb):
if exc_val and hasattr(exc_val, 'add_context'):
exc_val.add_context(**self.context)
return False # Не подавляем исключение

class ContextualError(Exception):
"""Исключение с возможностью добавления контекстной информации."""

def __init__(self, message):
super().__init__(message)
self.context = {}

def add_context(self, **kwargs):
"""Добавляет контекстную информацию к исключению."""
self.context.update(kwargs)

def __str__(self):
base_message = super().__str__()
if not self.context:
return base_message

context_str = ", ".join(f"{k}={v}" for k, v in self.context.items())
return f"{base_message} [Контекст: {context_str}]"

def process_user_data(user_data):
with context_info(user_id=user_data.get('id'), action='update'):
if not validate_data(user_data):
raise ContextualError("Ошибка валидации данных пользователя")
# Продолжение обработки...

try:
process_user_data({'name': 'John'})
except ContextualError as e:
print(f"Произошла ошибка: {e}")
# Вывод: "Произошла ошибка: Ошибка валидации данных пользователя [Контекст: user_id=None, action=update]"

# Вместо этого
def withdraw(account, amount):
if amount <= 0:
raise ValueError("Amount must be positive")
if account.balance < amount:
raise ValueError("Insufficient funds")
account.balance -= amount

# Лучше так
class NegativeAmountError(ValueError):
"""Raised when attempting to operate with a negative amount."""
pass

class InsufficientFundsError(ValueError):
"""Raised when an account has insufficient funds for an operation."""

def __init__(self, account_id, requested, available):
self.account_id = account_id
self.requested = requested
self.available = available
message = f"Account {account_id} has insufficient funds: requested {requested}, available {available}"
super().__init__(message)

def withdraw(account, amount):
if amount <= 0:
raise NegativeAmountError(f"Cannot withdraw {amount}, amount must be positive")
if account.balance < amount:
raise InsufficientFundsError(account.id, amount, account.balance)
account.balance -= amount

try:
process_transaction(transaction_data)
except InvalidAccountError as e:
# Специфичная обработка ошибки аккаунта
log_account_error(e)
notify_user_about_account(e.account_id)
return error_response("account_invalid", str(e))
except InvalidAmountError as e:
# Специфичная обработка ошибки суммы
log_amount_error(e)
return error_response("amount_invalid", str(e))
except TransactionError as e:
# Обработка других ошибок транзакций
log_transaction_error(e)
return error_response("transaction_failed", str(e))
except Exception as e:
# Запасной вариант для непредвиденных ошибок
log_unexpected_error(e)
return error_response("unexpected_error", "An unexpected error occurred")

def api_process_transaction(transaction_data):
try:
return process_transaction(transaction_data)
except InvalidAccountError as e:
# Преобразуем внутреннее исключение в API-исключение
raise APIError(
status_code=400,
error_code="INVALID_ACCOUNT",
message=str(e),
details={"account_id": e.account_id}
) from e
except InvalidAmountError as e:
raise APIError(
status_code=400,
error_code="INVALID_AMOUNT",
message=str(e),
details={"requested_amount": e.amount}
) from e

class ErrorHandler:
def __init__(self, error_response_func):
self.error_response_func = error_response_func
self.error_mappings = {}

def register(self, exception_class, status_code, error_code):
self.error_mappings[exception_class] = (status_code, error_code)
return self

def __enter__(self):
return self

def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type is None:
return False

for exception_class, (status_code, error_code) in self.error_mappings.items():
if isinstance(exc_val, exception_class):
response = self.error_response_func(
status_code=status_code,
error_code=error_code,
message=str(exc_val),
details=getattr(exc_val, 'details', None)
)
# Как-то обрабатываем response, например, возвращаем
return True # Подавляем исключение
return False # Пропускаем исключение дальше

def handle_transaction(transaction_data):
error_handler = ErrorHandler(create_error_response)
error_handler.register(InvalidAccountError, 400, "INVALID_ACCOUNT")
error_handler.register(InvalidAmountError, 400, "INVALID_AMOUNT")
error_handler.register(TransactionError, 500, "TRANSACTION_FAILED")

with error_handler:
return process_transaction(transaction_data)

def process_transaction(transaction_data):
"""
Process a financial transaction.

Args:
transaction_data (dict): Transaction details including 'account_id', 'amount', etc.

Returns:
dict: Processed transaction result.

Raises:
InvalidAccountError: If the account doesn't exist or is inactive.
InvalidAmountError: If the amount is negative or exceeds limits.
InsufficientFundsError: If the account doesn't have enough funds.
TransactionError: For other transaction-related errors.
"""
# Реализация...