Как работать со встроенными исключениями Python

# Стиль EAFP (предпочтителен в Python)
try:
    value = my_dict["key"]
    result = process_value(value)
except KeyError:
    result = default_value
 
# Стиль LBYL (характерен для других языков)
if "key" in my_dict:
    value = my_dict["key"]
    result = process_value(value)
else:
    result = default_value

def calculate_discount(price: float, discount_percent: float) -> float:
    """
    Рассчитывает сумму со скидкой.
    
    Args:
        price: Исходная цена
        discount_percent: Процент скидки (от 0 до 100)
        
    Returns:
        Цена со скидкой
    """
    if not (0 <= discount_percent <= 100):
        raise ValueError(f"Процент скидки должен быть от 0 до 100, получено {discount_percent}")
    
    return price * (1 - discount_percent / 100)

def transfer_money(source_account, target_account, amount):
    assert amount > 0, "Сумма перевода должна быть положительной"
    assert source_account.balance >= amount, "Недостаточно средств"
    
    source_account.withdraw(amount)
    target_account.deposit(amount)

def get_user_data(user_id):
    try:
        user = database.find_user(user_id)
        if not user:
            logger.warning(f"User {user_id} not found, using empty profile")
            return {"name": "Unknown", "email": "", "status": "inactive"}
        
        profile = {
            "name": user.name or "Unnamed",
            "email": user.email or "",
            "status": user.status if user.status in valid_statuses else "unknown"
        }
        return profile
    except DatabaseError as e:
        logger.error(f"Database error while fetching user {user_id}: {str(e)}")
        return {"name": "System Error", "email": "", "status": "error"}

BaseException
 ├── SystemExit
 ├── KeyboardInterrupt
 ├── GeneratorExit
 └── Exception
     ├── StopIteration
     ├── ArithmeticError
     │   ├── FloatingPointError
     │   ├── OverflowError
     │   └── ZeroDivisionError
     ├── AssertionError
     ├── AttributeError
     ├── BufferError
     ├── EOFError
     ├── ImportError
     │   └── ModuleNotFoundError
     ├── LookupError
     │   ├── IndexError
     │   └── KeyError
     └── ... (и много других)

try:
    # Сложная логика обработки данных
    process_data(huge_dataset)
except BaseException as e:
    logger.error(f"Произошла ошибка: {e}")
    # Продолжаем работу

try:
    process_data(huge_dataset)
except Exception as e:
    logger.error(f"Произошла ошибка: {e}")
    # Продолжаем работу

try:
    value = some_list[index] # Может вызвать IndexError
    other_value = some_dict[key] # Может вызвать KeyError
except LookupError as e:
    print(f"Ошибка доступа к данным: {e}")

class DataProcessingError(Exception):
    """Базовый класс для ошибок обработки данных."""
    pass
 
class DataValidationError(DataProcessingError):
    """Ошибка валидации входных данных."""
    pass
 
class DataTransformationError(DataProcessingError):
    """Ошибка преобразования данных."""
    pass
 
# Теперь можно перехватывать все ошибки обработки данных единообразно
try:
    validate_data(data)
    transform_data(data)
except DataProcessingError as e:
    print(f"Ошибка обработки данных: {e}")

try:
    import some_non_existent_module
except ModuleNotFoundError:
    print("Модуль не найден, устанавливаем...")
    install_module("some_non_existent_module")
except ImportError:
    print("Модуль найден, но возникла ошибка при импорте")

try:
    1 / 0
except Exception as exc:
    print(f"args: {exc.args}")
    print(f"__str__: {exc}")
    print(f"__repr__: {repr(exc)}")
    print(f"Трассировка:\n{exc.__traceback__}")

args: ('division by zero',)
__str__: division by zero
__repr__: ZeroDivisionError('division by zero')
Трассировка: <traceback object at 0x7f42b11ae8c0>

def log_exception(exc, logger):
    error_details = {
        "type": type(exc).__name__,
        "message": str(exc),
        "args": exc.args,
        "traceback": traceback.format_tb(exc.__traceback__)
    }
    
    # Если исключение имеет дополнительные атрибуты, сохраняем и их
    for attr in dir(exc):
        if not attr.startswith("__") and attr not in ["args", "with_traceback"]:
            try:
                value = getattr(exc, attr)
                if not callable(value):
                    error_details[attr] = repr(value)
            except Exception:
                pass
    
    logger.error(f"Exception details: {json.dumps(error_details, indent=2)}")

try:
    open("несуществующий_файл.txt")
except OSError as e:
    print(f"Код ошибки: {e.errno}")
    print(f"Сообщение ОС: {e.strerror}")
    print(f"Имя файла: {e.filename}")

class DataValidationError(Exception):
    def __init__(self, message, field_name=None, invalid_value=None):
        self.field_name = field_name
        self.invalid_value = invalid_value
        super().__init__(message)
    
    def __str__(self):
        base_msg = super().__str__()
        if self.field_name:
            return f"{base_msg} (поле: {self.field_name}, значение: {repr(self.invalid_value)})"
        return base_msg

try:
    raise ValueError("Оригинальная ошибка")
except ValueError as e:
    import sys
    _, _, original_tb = sys.exc_info()
    try:
        # Какой-то другой код, вызывающий ошибку
        1 / 0
    except:
        # Переиспользуем оригинальную трассировку
        raise e.with_traceback(original_tb)

try:
    for file_path in huge_file_list:
        process_file(file_path)
except Exception as e:  # Не перехватывает KeyboardInterrupt!
    logging.error(f"Обработка прервана: {e}")
except:  # А вот это перехватит ВСЁ, включая KeyboardInterrupt
    logging.error("Неизвестная ошибка")

try:
    for file_path in huge_file_list:
        process_file(file_path)
except Exception as e:
    logging.error(f"Обработка прервана: {e}")
except KeyboardInterrupt:
    logging.info("Процесс прерван пользователем")
    # Здесь можно выполнить действия по корректному завершению
    raise  # Повторно возбуждаем исключение для завершения программы

import sys
 
def validate_config():
    if not os.path.exists("config.ini"):
        print("Ошибка: файл конфигурации не найден!")
        sys.exit(1)  # Выход с ненулевым кодом — сигнал об ошибке
 
# В shell скрипте можно проверить:
# python check_config.py
# if [ $? -ne 0 ]; then
#     echo "Проверка конфигурации не пройдена"
# fi

try:
    # Основная логика
    if shutdown_requested:
        sys.exit(0)  # Нормальное завершение
    if critical_error_detected:
        sys.exit(2)  # Критическая ошибка
except Exception:
    sys.exit(1)  # Непредвиденная ошибка

import psutil
import os
 
def check_memory_usage(threshold_mb=100):
    process = psutil.Process(os.getpid())
    memory_info = process.memory_info()
    memory_mb = memory_info.rss / 1024 / 1024
    
    if memory_mb > threshold_mb:
        # Принудительно запускаем сборку мусора
        import gc
        gc.collect()
        
        # Проверяем снова
        memory_info = process.memory_info()
        memory_mb = memory_info.rss / 1024 / 1024
        
        if memory_mb > threshold_mb:
            # Если все еще выше порога, принимаем меры
            return False
    return True
 
# Использование в коде
while data_chunks:
    chunk = data_chunks.pop(0)
    process_chunk(chunk)
    
    if not check_memory_usage(500):
        # Сохраняем промежуточные результаты и освобождаем память
        save_intermediate_results()
        clear_caches()

# Код, написанный для Linux (до Python 3)
try:
    file = open("config.dat", "r")
    data = file.read()
except IOError:
    print("Не удалось открыть файл")
 
# А на Windows мог потребоваться дополнительный обработчик
except WindowsError:
    print("Специфическая ошибка Windows")

OSError
├── BlockingIOError
├── ChildProcessError
├── ConnectionError
│   ├── BrokenPipeError
│   ├── ConnectionAbortedError
│   ├── ConnectionRefusedError
│   └── ConnectionResetError
├── FileExistsError
├── FileNotFoundError
├── InterruptedError
├── IsADirectoryError
├── NotADirectoryError
├── PermissionError
└── TimeoutError

try:
    with open("sensitive_data.txt", "r") as file:
        content = file.read()
except FileNotFoundError:
    print("Файл не найден, создаю новый")
    initialize_new_file()
except PermissionError:
    print("Недостаточно прав для доступа к файлу")
    request_elevated_privileges()
except OSError as e:
    print(f"Другая ошибка ввода-вывода: {e}")

# Старый подход (до Python 3.3)
try:
    response = make_http_request(url, timeout=5)
except OSError as e:
    if "timed out" in str(e):
        retry_with_longer_timeout()
    elif "Connection refused" in str(e):
        try_alternative_server()
    else:
        handle_unknown_error(e)

# Современный подход
try:
    response = make_http_request(url, timeout=5)
except TimeoutError:
    retry_with_longer_timeout()
except ConnectionRefusedError:
    try_alternative_server()
except OSError as e:
    handle_unknown_error(e)

try:
    os.rename("source.dat", "destination.dat")
except OSError as e:
    print(f"Код ошибки: {e.errno}")
    print(f"Сообщение: {e.strerror}")
    print(f"Исходный файл: {e.filename}")
    print(f"Целевой файл: {e.filename2}")
    
    # На основе кода ошибки можно принять решение
    if e.errno == errno.EACCES:  # Отказ в доступе
        handle_permission_problem()
    elif e.errno == errno.ENOENT:  # Файл не найден
        handle_missing_file()

import errno
import os
 
try:
    os.mkdir("/path/to/folder")
except OSError as e:
    if e.errno == errno.EEXIST:
        print("Директория уже существует")
    elif e.errno == errno.EACCES:
        print("Недостаточно прав для создания директории")
    else:
        raise  # Пробрасываем неизвестные ошибки выше

# Классические примеры TypeError
"2" + 2       # TypeError: can only concatenate str (not "int") to str
len(42)       # TypeError: object of type 'int' has no len()
[1, 2, 3][1.5]  # TypeError: list indices must be integers or slices, not float

def validate_config(config):
    if not isinstance(config, dict):
        raise TypeError(f"Конфигурация должна быть словарем, получено {type(config).__name__}")
    
    if "parameters" not in config:
        raise KeyError("В конфигурации отсутствует секция 'parameters'")
    
    params = config["parameters"]
    if not isinstance(params, list):
        raise TypeError(f"'parameters' должен быть списком, получено {type(params).__name__}")
    
    for i, param in enumerate(params):
        if not isinstance(param, dict):
            raise TypeError(f"Параметр #{i} должен быть словарем, получено {type(param).__name__}")
        if "name" not in param or not isinstance(param["name"], str):
            raise TypeError(f"Параметр #{i} должен содержать строковое поле 'name'")

int("привет")   # ValueError: invalid literal for int() with base 10: 'привет'
[1, 2, 3].index(5)  # ValueError: 5 is not in list

def validate_age(age):
    if not isinstance(age, (int, float)):
        raise TypeError("Возраст должен быть числом")
    
    if age < 0 or age > 150:
        raise ValueError(f"Недопустимый возраст: {age}. Должен быть от 0 до 150")

def add_numbers(a: int, b: int) -> int:
    return a + b
 
# Это вызовет TypeError, несмотря на аннотации
result = add_numbers("2", 3)

from typeguard import check_type
 
def process_data(data: list[dict]) -> None:
    try:
        check_type('data', data, list[dict])
        # Продолжаем обработку, зная, что тип верный
    except TypeError as e:
        logging.error(f"Неверный формат данных: {e}")
        raise

try:
    result = process_complex_data(input_data)
except ValueError as e:
    # Обработка неверных значений
    handle_value_error(e)
except TypeError as e:
    # Обработка неверных типов
    handle_type_error(e)
except Exception as e:
    # Перехват других исключений
    handle_generic_error(e)

from pydantic import BaseModel, ValidationError, Field
 
class User(BaseModel):
    name: str
    age: int = Field(ge=0, lt=150)
    email: str
 
try:
    user = User(name="Иван", age=200, email="invalid_email")
except ValidationError as e:
    print(f"Ошибки валидации: {e}")

try:
    # Код, который может вызвать исключение
    risky_operation()
except SpecificError as e:
    # Обработка конкретного исключения
    handle_specific_error(e)
except (AnotherError, YetAnotherError) as e:
    # Обработка нескольких типов исключений
    handle_group_of_errors(e)
except Exception as e:
    # Обработка всех остальных исключений
    handle_unexpected_error(e)
else:
    # Выполняется, если в блоке try не возникло исключений
    cleanup_after_success()
finally:
    # Выполняется всегда, независимо от того, было исключение или нет
    release_resources()

def process_payment(payment_data):
    try:
        validate_payment_data(payment_data)  # Может вызвать ValidationError
        amount = calculate_amount(payment_data)  # Может вызвать ValueError
    except ValidationError as e:
        log_validation_error(e)
        return {"status": "error", "message": str(e)}
    except ValueError as e:
        log_calculation_error(e)
        return {"status": "error", "message": "Ошибка расчета суммы платежа"}
    else:
        # Выполнится только если все проверки пройдены успешно
        transaction_id = submit_to_payment_gateway(payment_data, amount)
        log_successful_payment(transaction_id)
        return {"status": "success", "transaction_id": transaction_id}
    finally:
        # Выполнится всегда
        cleanup_sensitive_data(payment_data)

# НЕПРАВИЛЬНО!
try:
    data = json.loads(input_string)
except Exception as e:
    print(f"Ошибка: {e}")
except json.JSONDecodeError as e:
    print(f"Некорректный JSON: {e}")  # Этот блок никогда не выполнится!

# ПРАВИЛЬНО
try:
    data = json.loads(input_string)
except json.JSONDecodeError as e:
    print(f"Некорректный JSON: {e}")
except Exception as e:
    print(f"Другая ошибка: {e}")

try:
    response = make_api_request(url, data)
except ConnectionRefusedError:
    # Сервер не доступен
    retry_or_use_fallback()
except ConnectionError:
    # Другие проблемы с соединением
    log_and_alert()
except TimeoutError:
    # Превышено время ожидания
    handle_timeout()
except Exception as e:
    # Неожиданная ошибка
    log_unexpected_error(e)
    raise  # Повторно возбуждаем исключение после логирования

def process_user_data(user_id):
    try:
        user = fetch_user(user_id)  # Может вызвать различные исключения
        return transform_user_data(user)
    except Exception as e:
        # Обработка на слишком высоком уровне абстракции!
        # Что произошло? Проблемы с БД? С сетью? С форматированием?
        log_error(f"Ошибка обработки пользователя {user_id}: {e}")
        return None

def process_user_data(user_id):
    try:
        user = fetch_user(user_id)
    except DatabaseConnectionError:
        log_db_error(f"Не удалось подключиться к БД при запросе пользователя {user_id}")
        return None
    except UserNotFoundError:
        log_info(f"Пользователь {user_id} не найден")
        return create_default_user()
    
    try:
        return transform_user_data(user)
    except ValidationError as e:
        log_warning(f"Ошибка валидации данных пользователя {user_id}: {e}")
        return sanitize_user_data(user)

file = open("huge_data.csv", "r")
try:
    data = process_file(file)
    save_results(data)
finally:
    file.close()  # Гарантируем закрытие файла

with open("huge_data.csv", "r") as file:
    data = process_file(file)
    save_results(data)
# Файл автоматически закроется при выходе из блока with

class Transaction:
    def __init__(self, db_connection):
        self.conn = db_connection
    
    def __enter__(self):
        self.conn.execute("BEGIN TRANSACTION")
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            # Если исключений не было, подтверждаем транзакцию
            self.conn.execute("COMMIT")
        else:
            # Если было исключение, откатываем изменения
            self.conn.execute("ROLLBACK")
            # Возвращаем False, чтобы исключение было проброшено дальше
            return False
 
# Использование:
with Transaction(database_connection):
    update_account_balance(account_id, -100)
    update_account_balance(recipient_id, 100)
    # Если одна из операций вызовет исключение, обе будут откачены

def process_config_file(file_path):
    try:
        with open(file_path) as f:
            config = json.load(f)
        return config
    except FileNotFoundError as e:
        raise ConfigError(f"Конфигурационный файл не найден: {file_path}") from e
    except json.JSONDecodeError as e:
        raise ConfigError(f"Ошибка в формате конфигурационного файла: {e}") from e

def retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except (ConnectionError, TimeoutError) as e:
                    attempts += 1
                    if attempts == max_attempts:
                        raise
                    logging.warning(
                        f"Попытка {attempts} не удалась: {e}. "
                        f"Повтор через {delay} сек."
                    )
                    time.sleep(delay)
        return wrapper
    return decorator
 
# Использование:
@retry(max_attempts=5, delay=2)
def fetch_data_from_api(url):
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

from contextlib import suppress
 
# Вместо:
try:
    os.remove('temp_file.txt')
except FileNotFoundError:
    pass  # Игнорируем случай, когда файл не существует
 
# Можно написать:
with suppress(FileNotFoundError):
    os.remove('temp_file.txt')

# Старый подход
try:
    os.remove('temp1.dat')
except FileNotFoundError:
    pass
 
try:
    os.remove('temp2.dat')
except FileNotFoundError:
    pass
 
# И так для каждого файла...

from contextlib import suppress
 
for filename in ['temp1.dat', 'temp2.dat', 'temp3.dat']:
    with suppress(FileNotFoundError):
        os.remove(filename)

async def task_guardian(coroutine, fallback_value=None):
    try:
        return await coroutine
    except Exception as e:
        logging.error(f"Ошибка в асинхронной задаче: {e}")
        return fallback_value
 
# Использование:
async def main():
    results = await asyncio.gather(
        task_guardian(fetch_data("api1.example.com")),
        task_guardian(fetch_data("api2.example.com")),
        task_guardian(fetch_data("api3.example.com")),
    )
    # Даже если некоторые запросы завершатся с ошибкой,
    # мы получим список с результатами или значениями по умолчанию

@retry(max_attempts=3)
def fetch_and_process_source(source_config):
    source_id = source_config["id"]
    logging.info(f"Обработка источника {source_id}")
    
    try:
        with timeout(30):  # Пользовательский контекстный менеджер для ограничения времени
            raw_data = fetch_data(source_config["url"])
    except TimeoutError:
        logging.error(f"Превышено время ожидания для источника {source_id}")
        return {"source_id": source_id, "status": "timeout"}
    except ConnectionError as e:
        logging.error(f"Ошибка подключения к источнику {source_id}: {e}")
        return {"source_id": source_id, "status": "connection_error"}
    
    try:
        processed_data = process_data(raw_data)
        with Transaction(db):  # Наш контекстный менеджер для транзакций
            save_to_database(processed_data)
    except DataProcessingError as e:
        logging.error(f"Ошибка обработки данных источника {source_id}: {e}")
        return {"source_id": source_id, "status": "processing_error"}
    except DatabaseError as e:
        logging.error(f"Ошибка сохранения в БД для источника {source_id}: {e}")
        return {"source_id": source_id, "status": "database_error"}
    
    logging.info(f"Источник {source_id} успешно обработан")
    return {"source_id": source_id, "status": "success", "records": len(processed_data)}

# Плохой пример
raise ValueError("Invalid input")
 
# Хороший пример
raise ValueError(
   f"Expected date in format YYYY-MM-DD, got '{date_str}'. "
   f"Please check the documentation at section 3.2 for valid formats."
)

def validate_user_age(age):
   if not isinstance(age, int):
       raise TypeError(
           f"Expected age to be an integer, got {type(age).__name__}. "
           "Please convert string values using int() before validation."
       )
   
   if age < 0 or age > 150:
       raise ValueError(
           f"Age must be between 0 and 150, got {age}. "
           "This validation is applied to prevent data entry errors."
       )

def process_order(order_data, user_id):
   try:
       # Обработка заказа
       validate_order(order_data)
       apply_discounts(order_data)
       charge_payment(order_data)
   except ValidationError as e:
       raise OrderProcessingError(
           f"Failed to process order for user {user_id}. "
           f"Validation error: {str(e)}. "
           f"Order data: {json.dumps(order_data, indent=2)}"
       ) from e
   except PaymentError as e:
       raise OrderProcessingError(
           f"Payment failed for user {user_id}, order #{order_data.get('id')}. "
           f"Error: {str(e)}. Amount: {order_data.get('total')}"
       ) from e

def log_exception(exc, logger, additional_context=None):
   error_info = {
       "type": exc.__class__.__name__,
       "message": str(exc),
       "timestamp": datetime.now().isoformat(),
       "traceback": traceback.format_exception(
           type(exc), exc, exc.__traceback__
       )
   }
   
   if additional_context:
       error_info["context"] = additional_context
   
   logger.error(f"Exception occurred: {json.dumps(error_info, indent=2)}")

class DatabaseQueryError(Exception):
   def __init__(self, message, query=None, parameters=None, underlying_error=None):
       self.query = query
       self.parameters = parameters
       self.underlying_error = underlying_error
       self.timestamp = datetime.now()
       
       # Формируем детальное сообщение
       detailed_msg = message
       if query:
           detailed_msg += f"\nQuery: {query}"
       if parameters:
           detailed_msg += f"\nParameters: {parameters}"
       if underlying_error:
           detailed_msg += f"\nCaused by: {type(underlying_error).__name__}: {underlying_error}"
       
       super().__init__(detailed_msg)
   
   def to_dict(self):
       """Конвертирует исключение в словарь для логирования или API-ответа"""
       return {
           "error_type": self.__class__.__name__,
           "message": str(self),
           "timestamp": self.timestamp.isoformat(),
           "query": self.query,
           "parameters": self.parameters,
           "underlying_error": str(self.underlying_error) if self.underlying_error else None
       }

try:
   user = authenticate(username, password)
except AuthenticationError as e:
   # Подробная версия для логов
   detailed_error = f"Authentication failed for user '{username}'. {str(e)}"
   logger.error(detailed_error)
   
   # Безопасная версия для пользователя
   raise HTTPError(status_code=401, detail="Invalid credentials")

from contextlib import ExitStack
 
def process_multiple_files(file_paths):
    with ExitStack() as stack:
        # Открываем все файлы сразу
        files = [stack.enter_context(open(path)) for path in file_paths]
        # Если при открытии любого файла возникнет исключение,
        # все уже открытые файлы будут корректно закрыты
        
        # Работаем со всеми файлами
        for file in files:
            process_file_content(file)

from contextlib import redirect_stdout
import io
 
def test_logging_function():
    f = io.StringIO()
    with redirect_stdout(f):
        log_important_info("Test message")
    
    output = f.getvalue()
    assert "Test message" in output

async def process_data_sources():
    tasks = [
        asyncio.create_task(fetch_from_source(url))
        for url in data_sources
    ]
    
    results = []
    for task in asyncio.as_completed(tasks):
        try:
            result = await task
            results.append(result)
        except ConnectionError:
            logging.error(f"Ошибка подключения к источнику")
            # Продолжаем с другими задачами
        except Exception as e:
            logging.error(f"Непредвиденная ошибка: {e}")
            # Решаем, нужно ли останавливать все задачи
            for t in tasks:
                if not t.done():
                    t.cancel()
            raise
    
    return results

async def resilient_task_processing():
    pending = set()
    try:
        for source in data_sources:
            try:
                task = asyncio.create_task(process_source(source))
                pending.add(task)
                task.add_done_callback(lambda t: pending.remove(t))
            except Exception as e:
                logging.error(f"Не удалось создать задачу для {source}: {e}")
        
        while pending:
            await asyncio.sleep(0.1)
            
    except asyncio.CancelledError:
        # Отмена всех незавершенных задач при отмене основной корутины
        for task in pending:
            task.cancel()
        # Ждем завершения отмененных задач
        await asyncio.gather(*pending, return_exceptions=True)
        raise

def exception_handler(fallback_value=None, retries=0, logger=None):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            while True:
                try:
                    return func(*args, **kwargs)
                except (ConnectionError, TimeoutError) as e:
                    attempts += 1
                    if attempts <= retries:
                        if logger:
                            logger.warning(f"Попытка {attempts}/{retries} не удалась: {e}")
                        time.sleep(1)  # Простая задержка перед повторной попыткой
                        continue
                    if logger:
                        logger.error(f"Все попытки исчерпаны: {e}")
                    return fallback_value
                except Exception as e:
                    if logger:
                        logger.error(f"Неожиданная ошибка: {e}")
                    return fallback_value
        return wrapper
    return decorator

@exception_handler(fallback_value=[], retries=3, logger=app_logger)
def fetch_user_transactions(user_id):
    response = requests.get(f"{API_URL}/users/{user_id}/transactions")
    response.raise_for_status()
    return response.json()

def async_exception_handler(fallback_value=None):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                logging.error(f"Ошибка в {func.__name__}: {e}")
                return fallback_value
        return wrapper
    return decorator
 
@async_exception_handler(fallback_value={})
async def fetch_user_profile(user_id):
    async with aiohttp.ClientSession() as session:
        async with session.get(f"{API_URL}/users/{user_id}") as response:
            response.raise_for_status()
            return await response.json()

def data_processor():
    try:
        for item in source_data:
            try:
                processed_item = process_item(item)
                yield processed_item
            except ItemProcessingError:
                # Пропускаем проблемные элементы
                continue
    except GeneratorExit:
        # Вызывается при закрытии генератора
        cleanup_resources()

class DatabaseTransaction:
    def __init__(self, connection):
        self.connection = connection
        
    def __enter__(self):
        self.cursor = self.connection.cursor()
        self.cursor.execute("BEGIN TRANSACTION")
        return self.cursor
        
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            # Если исключений не было, фиксируем транзакцию
            self.cursor.execute("COMMIT")
        else:
            # При исключении откатываем изменения
            self.cursor.execute("ROLLBACK")
            # Возвращаем False, чтобы исключение пробросилось дальше
            return False
        
        self.cursor.close()

try:
    process_data(user_input)
except:
    print("Что-то пошло не так")

try:
    process_data(user_input)
except ValueError as e:
    print(f"Неверный формат данных: {e}")
except ConnectionError as e:
    print(f"Проблема с подключением: {e}")
except Exception as e:
    print(f"Непредвиденная ошибка: {e}")
    # Логирование подробной информации
    logger.error(traceback.format_exc())

try:
    result = potentially_dangerous_operation()
except Exception:
    pass  # Просто игнорируем любые проблемы

try:
    os.remove('temp_file.txt')
except FileNotFoundError:
    # Игнорируем ситуацию, когда файл уже удален
    logger.debug("Файл temp_file.txt уже отсутствовал")

try:
    data = load_data_from_file(filename)
    result = process_data(data)
    save_result(result)
except Exception as e:
    print(f"Ошибка: {e}")

try:
    data = load_data_from_file(filename)
except FileNotFoundError:
    print(f"Файл {filename} не найден")
    return
except PermissionError:
    print(f"Нет прав для чтения файла {filename}")
    return
 
try:
    result = process_data(data)
except ValueError as e:
    print(f"Ошибка обработки данных: {e}")
    return
 
try:
    save_result(result)
except (IOError, OSError) as e:
    print(f"Ошибка сохранения результатов: {e}")

if value < 0:
    raise ValueError("Неверное значение")

if value < 0:
    raise ValueError(f"Значение должно быть положительным, получено: {value}")

# Плохой пример: использование исключений для проверки наличия ключа
def get_user_name(user_id):
    try:
        user = database.get_user(user_id)
        return user.name
    except KeyError:
        return "Unknown User"
 
# Лучше использовать обычные условные конструкции
def get_user_name(user_id):
    user = database.get_user(user_id)
    if user is None:
        return "Unknown User"
    return user.name

# Низкоуровневая функция
def read_config_file(path):
    try:
        with open(path) as f:
            return json.load(f)
    except FileNotFoundError:
        # Проброс специфического исключения
        raise ConfigError(f"Конфигурационный файл не найден: {path}")
    except json.JSONDecodeError as e:
        raise ConfigError(f"Ошибка парсинга конфигурации: {e}")
 
# Функция среднего уровня
def initialize_system(config_path):
    try:
        config = read_config_file(config_path)
        setup_services(config)
    except ConfigError as e:
        logger.error(f"Ошибка инициализации системы: {e}")
        raise SystemInitializationError(f"Не удалось запустить систему: {e}")
 
# Высокоуровневый обработчик
def main():
    try:
        initialize_system("config.json")
        run_application()
    except Exception as e:
        logger.critical(f"Критическая ошибка: {e}")
        sys.exit(1)

# Доменные исключения (внутренний слой)
class DomainException(Exception):
    """Базовый класс для всех исключений в домене"""
 
class InsufficientFundsException(DomainException):
    def __init__(self, account_id, required, available):
        self.account_id = account_id
        self.required = required
        self.available = available
        message = f"Недостаточно средств на счете {account_id}. Требуется: {required}, доступно: {available}"
        super().__init__(message)

# Исключения use cases (средний слой)
class UseCaseException(Exception):
    """Базовый класс для всех исключений в вариантах использования"""
    
class PaymentProcessingException(UseCaseException):
    def __init__(self, payment_id, reason):
        self.payment_id = payment_id
        self.reason = reason
        message = f"Не удалось обработать платеж {payment_id}: {reason}"
        super().__init__(message)

# Исключения интерфейса (внешний слой)
class APIException(Exception):
    """Базовый класс для исключений, возвращаемых через API"""
    def __init__(self, message, code, details=None):
        self.code = code
        self.details = details or {}
        super().__init__(message)
    
    def to_dict(self):
        """Конвертирует исключение в словарь для API-ответа"""
        return {
            "error": {
                "code": self.code,
                "message": str(self),
                "details": self.details
            }
        }

# Контроллер API (внешний слой)
def create_payment_endpoint(request):
    try:
        payment_data = validate_request(request)
        payment_id = payment_use_case.create_payment(payment_data)
        return {"payment_id": payment_id, "status": "success"}
    except ValidationError as e:
        raise APIException("Ошибка валидации", code="VALIDATION_ERROR", details=e.errors())
    except PaymentProcessingException as e:
        if isinstance(e.__cause__, InsufficientFundsException):
            # Трансформируем доменное исключение в API-исключение
            return APIException(
                "Недостаточно средств", 
                code="INSUFFICIENT_FUNDS",
                details={
                    "account_id": e.__cause__.account_id,
                    "required": e.__cause__.required,
                    "available": e.__cause__.available
                }
            )
        # Другие типы ошибок обработки платежа
        raise APIException("Ошибка обработки платежа", code="PAYMENT_ERROR")
    except Exception as e:
        logging.exception("Непредвиденная ошибка при создании платежа")
        raise APIException("Внутренняя ошибка сервера", code="INTERNAL_ERROR")

class AccountFrozenException(DomainException):
    """Счет заморожен и не может выполнять транзакции"""
    
class DailyTransferLimitExceededException(DomainException):
    """Превышен дневной лимит переводов"""
    
class SuspiciousActivityException(DomainException):
    """Обнаружена подозрительная активность по счету"""

class TransferService:
    def transfer(self, from_account, to_account, amount):
        # Проверки доменных правил
        if from_account.is_frozen:
            raise AccountFrozenException(from_account.id)
            
        daily_transfers = self.transfer_repository.get_daily_transfers(from_account.id)
        daily_total = sum(t.amount for t in daily_transfers)
        
        if daily_total + amount > from_account.daily_limit:
            raise DailyTransferLimitExceededException(
                account_id=from_account.id,
                limit=from_account.daily_limit,
                current_total=daily_total,
                requested=amount
            )
            
        # Проверка на подозрительную активность
        if self.fraud_detector.is_suspicious(from_account, to_account, amount):
            raise SuspiciousActivityException(from_account.id)
            
        # Выполнение перевода...

class DomainContext:
    def __init__(self):
        self._context = {}
        
    def add(self, key, value):
        self._context[key] = value
        return self
        
    def to_dict(self):
        return self._context.copy()
 
class ContextualDomainException(DomainException):
    def __init__(self, message, context=None):
        self.context = context or DomainContext()
        super().__init__(message)
        
    def add_context(self, key, value):
        self.context.add(key, value)
        return self

try:
    transfer_service.transfer(account1, account2, amount)
except DailyTransferLimitExceededException as e:
    # Обогащаем исключение информацией из текущего контекста
    e.add_context("user_id", current_user.id)
    e.add_context("transaction_type", "mobile_app_transfer")
    e.add_context("attempt_time", datetime.now().isoformat())
    # Пробрасываем дальше
    raise

# Инфраструктурный слой (сбор данных)
class MetricCollector:
    def collect(self, service_id):
        try:
            response = self.http_client.get(f"{self.base_url}/metrics/{service_id}")
            return self._parse_response(response)
        except requests.ConnectionError as e:
            raise MetricCollectionException(
                f"Не удалось подключиться к сервису {service_id}"
            ) from e
        except requests.Timeout as e:
            raise MetricCollectionTimeoutException(
                f"Превышено время ожидания ответа от сервиса {service_id}"
            ) from e
        except Exception as e:
            raise MetricCollectionException(
                f"Ошибка при сборе метрик с сервиса {service_id}"
            ) from e
 
# Доменный слой (аналитика)
class AnomalyDetector:
    def detect_anomalies(self, metrics):
        try:
            # Анализ метрик...
            if self._is_anomaly(metrics):
                return AnomalyEvent(metrics, severity=self._calculate_severity(metrics))
            return None
        except KeyError as e:
            # Отсутствует ключевая метрика
            missing_metric = str(e)
            raise AnalysisException(
                f"Отсутствует необходимая метрика: {missing_metric}"
            )
        except ValueError as e:
            # Некорректное значение метрики
            raise AnalysisException(
                f"Некорректное значение метрики: {str(e)}"
            )
 
# Слой приложения (оркестрация процессов)
class MonitoringService:
    def __init__(self, collector, detector, notifier):
        self.collector = collector
        self.detector = detector
        self.notifier = notifier
        
    def monitor_service(self, service_id):
        try:
            metrics = self.collector.collect(service_id)
            anomaly = self.detector.detect_anomalies(metrics)
            
            if anomaly:
                self.notifier.send_notification(anomaly)
                return {"status": "anomaly_detected", "severity": anomaly.severity}
            return {"status": "ok"}
            
        except MetricCollectionException as e:
            logging.error(f"Ошибка сбора метрик: {e}")
            # Повторная попытка через резервный канал
            return self._fallback_monitoring(service_id, e)
            
        except AnalysisException as e:
            logging.error(f"Ошибка анализа: {e}")
            return {"status": "analysis_error", "error": str(e)}
            
        except NotificationException as e:
            logging.error(f"Ошибка отправки уведомления: {e}")
            # Аномалия обнаружена, но уведомление не отправлено
            return {"status": "notification_error", "anomaly_detected": True}