Почему этому уроку отдельное место

В DE-работе нормальная ситуация: API упал, файл не открылся, JSON оказался HTML’ом, БД отказала в соединении. Программа должна корректно реагировать, а не падать с непонятным стеком. Junior, который умеет правильно обрабатывать ошибки, работает на проде в два раза стабильнее джуна, который этому не научился. Поэтому ошибкам — последний и большой урок этого модуля.

Что такое исключение

def divide(a: int, b: int) -> float:
    if b == 0:
        raise ValueError("Деление на ноль не определено")
    return a / b

print(divide(10, 0))
# Traceback (most recent call last):
#   File ..., line N, in <module>
#     print(divide(10, 0))
#   File ..., line N, in divide
#     raise ValueError("Деление на ноль не определено")
# ValueError: Деление на ноль не определено

Иерархия встроенных исключений

В Python почти все исключения — наследники одного дерева. Знать всю иерархию наизусть не нужно, но знать главные классы — обязательно:

Главное правило: Exception — это базовый класс для прикладных ошибок. BaseException ловит ещё и SystemExit/KeyboardInterrupt, что почти всегда не то, что вам нужно. Пишите except Exception, не except BaseException.

try / except — базовый шаблон

try:
    risky_operation()
except SomeError:
    handle_it()

Несколько ветвей:

try:
    payload = json.loads(text)
    user = payload["user"]
    age = int(payload["age"])
except json.JSONDecodeError:
    log.error("Битый JSON")
except KeyError as e:
    log.error(f"Не хватает ключа: {e}")
except ValueError as e:
    log.error(f"Не получается распарсить число: {e}")

as e — привязывает объект исключения к имени e. Внутри блока можно получить сообщение через str(e), repr(e), или сами поля исключения.

except без класса ловит всё, что наследуется от BaseException. Это

# плохо
try:
    do_something()
except:                  # <-- ловит ВСЁ
    pass                 # <-- ещё и молча

# плохо
try:
    do_something()
except Exception:
    pass                 # <-- молча проглатывает любые ошибки

Вторая форма (с явным Exception) тоже плохая, если внутри pass или нет логирования. Никогда не «глотайте» исключения молча. Минимум — залогируйте.

else и finally

Полная форма блока:

try:
    f = open("data.csv")
    data = f.read()
except FileNotFoundError:
    data = None
else:
    # выполнится, если try НЕ упал
    print("Файл прочитан")
finally:
    # выполнится ВСЕГДА — упало в try или нет
    if "f" in locals() and not f.closed:
        f.close()

• else — выполняется, если в try не было исключения. Полезно, чтобы не путать «ошибочный путь» с «успешным после try».
• finally — выполняется всегда, даже если внутри try или except есть return. Используется для cleanup: закрыть файл, освободить лок, отдать соединение в пул.

Чаще для cleanup используется

# идиоматично — with сам закроет файл
try:
    with open("data.csv") as f:
        data = f.read()
except FileNotFoundError:
    data = None

raise — поднять исключение

Чтобы поднять своё:

def parse_age(s: str) -> int:
    if not s:
        raise ValueError("Возраст не указан")
    try:
        age = int(s)
    except ValueError:
        raise ValueError(f"Возраст должен быть числом, получили: {s!r}")

    if age < 0 or age > 150:
        raise ValueError(f"Возраст вне разумного диапазона: {age}")
    return age

Несколько встроенных классов, которые удобно использовать:

• ValueError — значение неподходящее (плохая строка, число вне диапазона).
• TypeError — тип не тот.
• KeyError — ключа нет.
• FileNotFoundError — нет файла.
• RuntimeError — общая «что-то пошло не так» для нестандартных случаев.
• NotImplementedError — функция-заглушка, ещё не реализована.

raise from — сохранить контекст

Когда вы перебрасываете исключение из другого — используйте raise ... from, чтобы Python показал обе ошибки:

import json

def load_config(path: str) -> dict:
    try:
        with open(path) as f:
            return json.load(f)
    except json.JSONDecodeError as e:
        raise ConfigError(f"Битый JSON в {path}") from e

В traceback будет видно:

json.JSONDecodeError: Expecting value: line 1 column 1
...
The above exception was the direct cause of the following exception:
...
ConfigError: Битый JSON в /etc/config.json

Это бесценно при дебаге — видно, что сломалось (JSONDecodeError) и где в нашем коде это поймано (ConfigError).

Если хотите подавить контекст (показать только новое исключение, без оригинала) — raise X from None.

Custom exceptions для домена

Для серьёзного проекта стоит сделать собственную иерархию исключений ETL. Это позволяет:

1. В одном except отловить все «наши» ошибки.
2. В логах сразу видеть, что это ошибка нашего домена, а не баг чужой библиотеки.
3. В разных слоях кода обрабатывать разные ошибки по-разному.

# etl/errors.py

class ETLError(Exception):
    """Базовая ошибка нашего ETL."""


class ExtractError(ETLError):
    """Не удалось получить данные из источника."""


class APIError(ExtractError):
    """Ошибка ответа API."""

    def __init__(self, status_code: int, message: str) -> None:
        super().__init__(f"API вернул {status_code}: {message}")
        self.status_code = status_code


class TransformError(ETLError):
    """Не удалось распарсить или валидировать запись."""


class LoadError(ETLError):
    """Не удалось загрузить в хранилище."""

Использование:

from etl.errors import APIError, ETLError

def fetch_users() -> list[dict]:
    response = requests.get("https://api.example.com/users", timeout=30)
    if response.status_code != 200:
        raise APIError(response.status_code, response.text[:200])
    return response.json()


def main() -> None:
    try:
        users = fetch_users()
        process(users)
    except APIError as e:
        # специфическая обработка — API уведомили, продолжаем со старыми данными
        log.warning(f"API недоступен: {e}. Использую вчерашний снапшот.")
        users = load_yesterday_snapshot()
        process(users)
    except ETLError as e:
        # любая другая ETL-ошибка — алерт и стоп
        log.error(f"Ошибка ETL: {e}", exc_info=True)
        raise
    except Exception:
        # что-то совсем непонятное — паника
        log.exception("Неожиданная ошибка")
        raise

exc_info=True в log.error добавляет traceback в лог. log.exception(...) делает то же автоматически.

Минимальный custom exception — просто наследник Exception, никаких дополнительных полей. Тело pass или docstring. Этого хватает в 80% случаев.

Anti-patterns

Несколько типовых ошибок junior’ов:

1. Pokemon catch — «поймать всех»

# плохо
try:
    process(record)
except:                      # <-- bare except
    pass                     # <-- ещё и молча

Что плохо:

• Прячет реальные баги (TypeError, AttributeError, NameError — никогда не увидите).
• Ловит даже Ctrl+C (KeyboardInterrupt).
• Маскирует системные ошибки (SystemExit, MemoryError).

Хорошо:

try:
    process(record)
except (ValidationError, KeyError) as e:
    log.warning(f"Невалидная запись {record.get('id')}: {e}")

Знайте, какие ошибки ожидаете. Остальные пусть всплывают.

2. Молчаливое игнорирование

# плохо
try:
    risky()
except Exception:
    pass

Даже если вы поймали правильный класс — pass без логирования — пуля в ногу. Через месяц у вас «ETL отрабатывает, но загружено 0 записей», и непонятно почему.

Минимум:

try:
    risky()
except Exception as e:
    log.warning(f"risky() упало: {e}")

Лучше — структурное логирование (мы коснёмся в модуле 9).

3. except + return + finally с return

Сложный кейс, на котором ломаются даже опытные:

def f():
    try:
        return 1
    finally:
        return 2   # <-- ВСЕГДА вернёт 2, перекроет 1!

finally выполняется после return из try. Если в finally тоже return — он перекроет первый. Никогда не делайте return в finally — это почти всегда баг.

4. Catch чужих ошибок не там, где знаете, что делать

# плохо: в extract обрабатываем ошибки, которые умеет обработать только main
def extract():
    try:
        return requests.get("...")
    except Exception:
        return None         # потерян контекст, дальше непонятно, почему пусто

Лучше — поднимать наверх, обрабатывать на верхнем уровне, где видно весь контекст:

def extract() -> list[dict]:
    response = requests.get("...", timeout=30)
    response.raise_for_status()      # raises HTTPError
    return response.json()


def main():
    try:
        data = extract()
    except requests.HTTPError as e:
        log.error(f"API недоступен: {e}")
        send_alert(...)
        sys.exit(1)

Правило: ловите там, где знаете, что делать. Не там, где исключение случилось.

ExceptionGroup (3.11+) — кратко awareness

В Python 3.11 появилось понятие

import asyncio

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fetch("api1"))
        tg.create_task(fetch("api2"))   # обе могут упасть параллельно

try:
    asyncio.run(main())
except* ConnectionError as eg:        # except* — звёздочка!
    log.error(f"Сетевые ошибки: {eg.exceptions}")
except* ValueError as eg:
    log.error(f"Ошибки парсинга: {eg.exceptions}")

except* (со звёздочкой) — новый синтаксис для разбора ExceptionGroup. Вам как junior’у это редко встретится — async-код пишут реже. Но знать про существование — нужно, чтобы при чтении кода 2024+ года не пугаться.

DE-кейс: ETL с обработкой ошибок API, БД и парсинга

Соберём канонический ETL-скрипт с правильной обработкой:

import logging
import sys
import requests
import psycopg

log = logging.getLogger("etl")


class ETLError(Exception):
    pass


class APIError(ETLError):
    pass


class ParseError(ETLError):
    pass


class LoadError(ETLError):
    pass


def fetch_users(url: str) -> list[dict]:
    try:
        response = requests.get(url, timeout=30)
    except requests.Timeout as e:
        raise APIError(f"Таймаут запроса к {url}") from e
    except requests.RequestException as e:
        raise APIError(f"Сетевая ошибка для {url}: {e}") from e

    if response.status_code != 200:
        raise APIError(f"API вернул {response.status_code}")

    try:
        return response.json()
    except ValueError as e:
        # бывает: API вернул HTML вместо JSON
        snippet = response.text[:200]
        raise APIError(f"Невалидный JSON: {snippet!r}") from e


def parse_users(raw: list[dict]) -> list[dict]:
    clean: list[dict] = []
    for r in raw:
        try:
            clean.append(
                {
                    "id": int(r["id"]),
                    "name": r["name"].strip(),
                    "email": r.get("email", "").lower() or None,
                }
            )
        except (KeyError, ValueError, AttributeError) as e:
            log.warning(f"Невалидная запись {r}: {e}")
            continue
    return clean


def load_users(users: list[dict], dsn: str) -> int:
    try:
        with psycopg.connect(dsn, connect_timeout=10) as conn:
            with conn.cursor() as cur:
                for u in users:
                    cur.execute(
                        "INSERT INTO users(id, name, email) VALUES (%s, %s, %s) "
                        "ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name, email=EXCLUDED.email",
                        (u["id"], u["name"], u["email"]),
                    )
    except psycopg.OperationalError as e:
        raise LoadError(f"Не удалось подключиться к БД: {e}") from e
    return len(users)


def main() -> int:
    try:
        raw = fetch_users("https://api.example.com/users")
        users = parse_users(raw)
        if not users:
            log.warning("Все записи отфильтрованы — нечего грузить")
            return 0
        n = load_users(users, dsn="postgresql://...")
        log.info(f"Загружено {n} записей")
        return 0

    except APIError as e:
        log.error(f"Источник недоступен: {e}", exc_info=True)
        return 2     # exit code 2 — внешняя проблема, можно перезапустить

    except LoadError as e:
        log.error(f"БД недоступна: {e}", exc_info=True)
        return 3

    except Exception:
        log.exception("Неожиданная ошибка")
        return 1     # 1 — bug в коде


if __name__ == "__main__":
    sys.exit(main())

Что здесь правильно:

• Каждый слой (fetch_users, parse_users, load_users) бросает свои доменные исключения.
• parse_users ловит ошибки на уровне отдельной записи и пропускает её, не падая весь батч.
• main ловит свои доменные классы и возвращает разные exit codes — это позволяет оркестратору (Airflow, cron) отличать transient-ошибки от багов.
• log.exception пишет с traceback.
• raise ... from e сохраняет контекст оригинальной ошибки.

Упражнение

В файле safe_fetch.py:

# 1. Напишите функцию safe_int(s: str) -> int | None,
# которая пытается распарсить строку как int.
# При ошибке возвращает None и пишет warning в log.

# 2. Напишите кастомное исключение DataValidationError(Exception).
# Добавьте к нему поле field_name.

# 3. Напишите функцию validate_user(raw: dict) -> dict,
# которая поднимает DataValidationError, если в raw нет id или name.

# 4. Напишите функцию fetch_json_safe(url: str) -> list[dict] | None,
# которая:
#   - возвращает список словарей при успехе;
#   - при HTTP-ошибке возвращает None и логирует;
#   - при невалидном JSON логирует с raise from и возвращает None;
#   - при любой другой ошибке - пробрасывает наверх.

# Для тестирования используйте url = "https://httpbin.org/status/500" (500 ошибка)
# и url = "https://httpbin.org/html" (валидный HTTP, но HTML вместо JSON).

Критерии:

• В (1) функция ловит только ValueError, не Exception.
• В (3) DataValidationError поднимается с понятным сообщением.
• В (4) разные виды ошибок обрабатываются разными except’ами.
• Нигде нет bare except или except: pass.

Итог модуля

За этот модуль мы прошли от голого синтаксиса (переменные, типы) до настоящего «миниприложения» (ETL с модулями и обработкой ошибок). Этого достаточно, чтобы:

• Прочитать любой Python-скрипт junior-уровня и понять, что он делает.
• Написать собственный скрипт от парсинга CSV до отправки в БД.
• Не допускать главных ловушек: mutable defaults, bare except, путаница is и ==, забытый encoding.

Дальше — модуль 4 «Идиомы Python»: ленивая обработка через генераторы, контекст-менеджеры (with), декораторы для retry-логики. Это то, что превращает скрипт в production-код.