Sentry. snitchbot.
Crashes, load, anomalies —
delivered to Telegram.

import snitchbot

snitchbot.init("orders-api")

# Unhandled exceptions are captured automatically,
# including stack, thread, and origin.

async def list_orders(user_id: int) -> list:
    return await svc.fetch_all(user_id)

# Somewhere down the stack:
#   raise DatabaseConnectionError("refused to ...")
# -> snitchbot captures and sends the alert below.

import asyncio
import time
import snitchbot


@snitchbot.watch_slow(threshold_ms=100)
async def fetch_user_profile(user_id: int) -> dict:
    await asyncio.sleep(0.25)  # 250 ms > threshold
    return {"name": "Alice"}


@snitchbot.watch_slow(threshold_ms=500)
def generate_report() -> str:
    time.sleep(0.6)  # sync, also captured
    return "report-data"


async def main():
    snitchbot.init("slow-demo")
    await fetch_user_profile(42)
    generate_report()

import snitchbot
from snitchbot import AnomalyConfig, WatchdogConfig

snitchbot.init("watchdog-demo")

# Zero-config: watchdog is on, threshold 500 ms,
# auto-escalates to error at 2 s, critical at 5 s.

# Full config with custom thresholds:
snitchbot.init(
    "watchdog-demo",
    anomaly=AnomalyConfig(
        watchdog=WatchdogConfig(
            threshold_ms=500,            # 🟠 warning
            error_threshold_ms=2000,      # 🔴 error
            critical_threshold_ms=5000,   # 🟣 critical
            escalation_window="1m",
            cooldown_sec=5,
        ),
    ),
)

import snitchbot
from snitchbot import (
    AnomalyConfig,
    RssAnomalyConfig,
    CpuAnomalyConfig,
    FdAnomalyConfig,
    ThreadAnomalyConfig,
)

snitchbot.init(
    "anomaly-demo",
    sample_interval_sec=5,
    anomaly=AnomalyConfig(
        rss=RssAnomalyConfig(
            duration="1m", baseline_duration="30m",
            max_mb=450,          # 🔴 ceiling
            spike_ratio=1.5,      # 🟠 +50% vs baseline
            min_spike_mb=50,      # and ≥ 50 MB absolute
        ),
        cpu=CpuAnomalyConfig(
            duration="2m", baseline_duration="20m",
            max_percent=90,       # 🔴 ceiling
            spike_ratio=2.5,       # 🟠 spike
            min_spike_delta=30,    # ≥ 30 pp
        ),
        fds=FdAnomalyConfig(
            max_fds=800,          # 🔴 ulimit guard
            spike_ratio=1.5,       # 🔴 fd leak
            drop_ratio=0.5,        # 🟠 pool collapse
        ),
        threads=ThreadAnomalyConfig(
            max_threads=100,
            spike_ratio=1.5,
        ),
    ),
)

import snitchbot

snitchbot.init("orders-api")

# ▶ lifecycle("startup", reason="init")  — sent immediately

# ...your service does its thing...

# On any of these paths, a shutdown event is emitted:
#  · Clean exit      -> reason="clean_exit", exit_code=0
#  · SIGTERM / SIGINT -> reason="sigterm" / "sigint"
#  · Uncaught crash   -> reason="crash" (+ traceback)
#  · Thread crash     -> reason="thread_crash"

# Nothing to call, nothing to decorate.

import snitchbot

snitchbot.init("notify-demo")

# Warning with extras — renders as a meta-table
snitchbot.notify(
    "Starting checkout process",
    severity="warning",
    extras={"cart_size": 3, "user": "Alice"},
)

# Error with live traceback
try:
    _ = 1 / 0
except ZeroDivisionError:
    snitchbot.notify(
        "Division failed in payment calculator",
        severity="error",
        exc_info=True,
    )

import asyncio
import snitchbot


@snitchbot.watch_slow(threshold_ms=100)
async def call_payment_api(amount: float) -> str:
    await asyncio.sleep(0.2)
    return "txn-12345"


async def handle_request(request_id: str, user_id: int):
    with snitchbot.request_context(
        trace_id=request_id,
        user_id=user_id,
        action="checkout",
    ):
        snitchbot.notify(
            "User started checkout",
            extras={"cart_size": 3},
        )
        await call_payment_api(99.99)  # inherits ctx

import logging
import snitchbot

snitchbot.init("log-demo")
snitchbot.setup_logging()  # WARNING+ -> Telegram

# Or, for structlog users:
# processor = snitchbot.setup_structlog()
# structlog.configure(processors=[..., processor])

logger = logging.getLogger("myapp")

# Extras become a meta-table in the alert
logger.warning(
    "Cache miss rate too high",
    extra={"miss_pct": 42},
)

# exc_info=True attaches the traceback
try:
    _ = 1 / 0
except ZeroDivisionError:
    logger.error("Calculation failed", exc_info=True)

# Inside request_context — trace_id attached automatically
with snitchbot.request_context(trace_id="req-abc-123"):
    logger.warning("Slow DB query in checkout")

import snitchbot

snitchbot.init("web-demo")
snitchbot.setup_logging()

# ── FastAPI ───────────────────────────────────
from fastapi import FastAPI
from snitchbot.integrations.fastapi import install

app = FastAPI()
install(app)

# ── Flask ─────────────────────────────────────
# from flask import Flask
# from snitchbot.integrations.flask import install
# app = Flask(__name__); install(app)

# ── Litestar ──────────────────────────────────
# from litestar import Litestar
# from snitchbot.integrations.litestar import install
# app = Litestar(route_handlers=[...]); install(app)


@app.post("/checkout")
async def checkout(cart_value: int = 100):
    snitchbot.notify("Large checkout",
                     extras={"cart_value": cart_value})
    return {"status": "processing"}


@app.get("/search")
async def search(query: str):
    raise ValueError("Unknown search backend")