oopsys-python 0.1.0

Catch exceptions, log them, and optionally report to an agent — without crashing your app.

oopsys-python

oops, system crashed :)

oopsys-python ловит ошибки за вас. Вы навешиваете один декоратор на точку входа (или на «единицу работы»), и дальше любое исключение превращается в аккуратный структурированный лог и (опционально) отчёт, отправленный агенту по HTTP. Приложение при этом не падает.

Цель — убрать ручные try/except и человеческий фактор: вы не забудете обработать ошибку, потому что её обрабатывает oopsys.

• Работает одинаково для sync и async (определяется автоматически).
• Логирование через structlog / kitstructlog.
• Конфигурация через pydantic-settings, только переменные с префиксом OOPSYS_.
• Без агента библиотека полностью рабочая — просто никуда не отправляет.

---

Установка

Репозиторий: github.com/morington/oopsys-python

PyPI (рекомендуется)

pip install oopsys-python

uv add oopsys-python

В pyproject.toml зависимости:

dependencies = ["oopsys-python>=0.1.0"]

Импорт в коде: from oopsys_python import guard, configure, ... (подчёркивание, не дефис).

GitHub (ветка main или тег)

pip install "oopsys-python @ git+https://github.com/morington/oopsys-python.git"

uv add "oopsys-python @ git+https://github.com/morington/oopsys-python.git"

Конкретная версия по тегу:

pip install "oopsys-python @ git+https://github.com/morington/oopsys-python.git@v0.1.0"

---

Публикация на PyPI (для сопровождающих)

uv build
uv publish   # нужен токен: UV_PUBLISH_TOKEN или настройка ~/.pypirc

Либо релиз на GitHub — workflow .github/workflows/publish.yml соберёт и опубликует пакет, если в secrets репозитория задан PYPI_API_TOKEN.

---

Быстрый старт

from oopsys_python import configure, guard


@guard
async def main() -> None:
    ...  # ваша логика


if __name__ == "__main__":
    import asyncio

    configure()          # один раз: читает .env и настраивает логи
    asyncio.run(main())

configure() создаёт настройку из окружения и инициализирует логгеры. После этого guard, timeit, ignore, install берут её автоматически.

---

Что входит в библиотеку

Объект	Что делает
configure(config=None, *, transport=None, setup_logging=True)	Настраивает oopsys один раз для всего приложения. Обычно вызывается в точке входа.
@guard	Главный декоратор. Перехватывает ошибки функции, логирует/отправляет, не даёт упасть. Sync и async.
@timeit	Логирует время выполнения функции. Только в development, уровень DEBUG.
@ignore	Исключает функцию из обработки oopsys — её ошибка летит как обычно.
install()	Глобальная «сеть»: ловит вообще всё необработанное (потоки, asyncio) как critical.
Configuration	Настройки (OOPSYS_*).
Oopsys	Низкоуровневый класс, если нужен явный экземпляр.

---

@guard — основное

Вешаешь на функцию. Дальше два сценария:

Всё ок — функция отработала как обычно, возвращает свой нормальный результат.

Внутри вылетела ошибка — oopsys ловит её, пишет в лог (и в агент, если включён), не роняет программу и возвращает из функции то, что ты указал в fallback. Снаружи это выглядит так, будто функция просто вернула это значение — без raise, без traceback.

fallback — что функция «вернёт», если упала

Самое главное: fallback — это ответ функции при ошибке.

Не колбэк, не «обработчик» — обычное значение, как return 42, только oopsys подставляет его сам, когда внутри было исключение.

@guard(fallback=None)
async def fetch(url: str) -> dict | None:
    r = await client.get(url)
    return r.json()

data = await fetch("https://bad-host")   # сеть упала
# data is None  — не исключение, а fallback
# ошибка уже в логе oopsys

if data is None:
    ...   # твоя реакция: повторить, пропустить, подождать

Если fallback не указать — по умолчанию None.

Другие варианты — любое значение, которое тебе удобно проверять дальше:

@guard(fallback=None)              # «нет результата» — if x is None
@guard(fallback=False)             # функция bool — при ошибке False
@guard(fallback=[])                # при ошибке пустой список (см. ниже)
@guard(fallback={"ok": False})     # «заглушка» как при своей бизнес-ошибке

Успех — fallback не трогается, возвращается реальный return из тела функции.

Ошибка — всегда один и тот же fallback, неважно ValueError это или ConnectError. Какая именно ошибка была — смотри в логе (error_type, detail).

Нельзя написать fallback=lambda: [] в надежде, что lambda вызовется — вернётся сама lambda. Нужно готовое значение: [], None, False и т.д.

Осторожно с fallback=[]: при каждой ошибке вернётся один и тот же список в памяти. Если его меняешь — лучше fallback=None и делай [] у себя в коде после проверки.

При reraise=True fallback не используется — ошибка снова летит наружу (удобно в dev, см. ниже).

Остальные параметры @guard

• critical=True — в логе и в агенте пометить как «приложение упало» (уровень critical), но процесс всё равно не падает, если не reraise.
• reraise=True — залогировать и пробросить ошибку как без oopsys.

KeyboardInterrupt и SystemExit oopsys не ловит — Ctrl+C работает как обычно.

Куда вешать guard

Не нужно декорировать все 20 000 функций. Достаточно границ, где работа начинается заново:

• точка входа main;
• корутина-воркер, которую вы крутите в цикле;
• задача asyncio.create_task(worker());
• джоба планировщика (APScheduler и т.п.);
• обработчик запроса / события.

Принцип: вешайте guard на повторяющуюся единицу работы, чтобы её падение не убивало весь процесс.

---

@timeit — скорость выполнения

from oopsys_python import timeit


@timeit
def heavy() -> None:
    ...

Логирует execution time ... ms=... hf=0:00:00.6915 на уровне DEBUG и только в development-режиме (OOPSYS_IS_DEVELOPMENT=true). В production это полный no-op, поэтому декоратор можно смело оставлять в коде.

Совмещается с guard (порядок любой):

@guard(fallback=None)
@timeit
async def fetch_fact() -> str:
    ...

---

@ignore — исключить функцию

Иногда нужно, чтобы конкретная функция падала «сама по себе», без вмешательства oopsys. Навесьте @ignore — её исключение пройдёт сквозь guard и install нетронутым.

from oopsys_python import guard, ignore


@ignore
def must_fail() -> None:
    raise RuntimeError("обрабатываю сам")


@guard
def main() -> None:
    must_fail()   # это исключение oopsys НЕ перехватит — оно вылетит наружу

Нужно редко, но возможность есть.

---

install() — глобальная сеть

guard защищает то, что вы обернули. install() — страховка на случай, если где-то осталось необработанное исключение. Он вешает хуки на главный поток, рабочие потоки и event loop, и любое «утёкшее» исключение логирует как critical и отправляет агенту.

from oopsys_python import configure, install


def bootstrap() -> None:
    configure()
    install()   # дальше ничего не пропадёт молча

Важно: install() сообщает о падении, но не «оживляет» программу постфактум — чтобы приложение продолжало работать, нужен guard на единице работы. Ctrl+C / SystemExit и функции с @ignore он пропускает.

---

Конфигурация (.env)

Читаются только переменные с префиксом OOPSYS_, остальные игнорируются:

OOPSYS_IS_DEVELOPMENT=false       # dev-режим: включает timeit, влияет на reraise
OOPSYS_SERVICE_NAME=my-service    # имя сервиса в отчётах
OOPSYS_RERAISE=false              # true: логировать и пробрасывать ошибку дальше

OOPSYS_AGENT__ENABLED=true        # слать ли отчёты агенту
OOPSYS_AGENT__HOST=localhost
OOPSYS_AGENT__PORT=8080
OOPSYS_AGENT__PATH=/reports
OOPSYS_AGENT__TIMEOUT=3.0

Можно задать всё и в коде:

from oopsys_python import Configuration, configure

configure(Configuration(is_development=True, service_name="parser"))

Лог vs обычное падение в development

OOPSYS_RERAISE (или guard(reraise=True)) управляет поведением:

• reraise=false (по умолчанию) — ошибка логируется и гасится, приложение работает дальше. Так нужно в production.
• reraise=true — ошибка логируется (уровень error) и пробрасывается — видите настоящий traceback. Удобно во время разработки и отладки.

---

Что уходит агенту

При каждой пойманной ошибке формируется JSON-отчёт:

{
  "severity": "error",
  "service": "my-service",
  "environment": "production",
  "exception_type": "ValueError",
  "message": "invalid literal for int()",
  "traceback": "Traceback (most recent call last): ...",
  "timestamp": "2026-06-01T12:00:00Z",
  "context": {"callable": "compute"}
}

• severity: error (приложение живо) или critical (упало бы).
• Отправка — обычный HTTP POST на OOPSYS_AGENT__*.
• Если агент выключен или недоступен — отчёт тихо пропускается (лог debug), приложение не страдает.

---

Сценарии

1. Долгоживущий воркер / парсер

@guard(fallback=None)
@timeit
async def fetch_fact() -> str:
    ...

async def main() -> None:
    while True:
        fact = await fetch_fact()
        if fact is None:        # ошибка уже залогирована/отправлена
            await asyncio.sleep(5)
            continue
        logger.info("fact", fact=fact)
        await asyncio.sleep(15)

Сеть упала — цикл не падает, просто ждёт и повторяет. (см. examples/parser.py)

2. Планировщик (APScheduler)

@guard(critical=False)
async def job() -> None:
    ...

scheduler.add_job(job, "interval", seconds=30)

Падение одной джобы не роняет планировщик; следующий запуск идёт по расписанию.

3. Фоновые задачи

@guard
async def worker() -> None:
    ...

asyncio.create_task(worker())   # ошибка задачи будет поймана, а не "потеряна"

4. CLI / скрипт

@guard(critical=True)
def main() -> None:
    ...

if __name__ == "__main__":
    configure()
    main()

---

Примеры в репозитории

uv run python examples/protected_sync.py    # sync: guard + timeit
uv run python examples/protected_async.py   # async: guard + ignore + install
uv run python examples/parser.py            # реальный HTTP-воркер в цикле

---

Низкоуровневый доступ

Если нужен явный экземпляр без глобальной настройки:

from oopsys_python import Configuration, Oopsys

oops = Oopsys(Configuration(service_name="x"))
report = oops.capture(exc, critical=True)   # вручную: лог + отправка