Guide

Python integration

Wire Drumbeats into Python with a decorator, plus notes for Celery, Airflow, and Django.

import requests

API = "https://api.drumbeats.io/v1/ping/<monitor-id>"

requests.get(f"{API}/start", timeout=3)
try:
    run_my_job()
    requests.get(f"{API}/success", timeout=3)
except Exception as exc:
    requests.post(f"{API}/failure", json={"payload": str(exc)}, timeout=3)
    raise

import requests

API = "https://api.drumbeats.io/v1/ping/<monitor-id>"

requests.get(f"{API}/start", timeout=3)
try:
    run_my_job()
    requests.get(f"{API}/success", timeout=3)
except Exception as exc:
    requests.post(f"{API}/failure", json={"payload": str(exc)}, timeout=3)
    raise

POST the failure with a payload so the error text reaches the incident timeline. Success needs no body.

Examples use requests. There is an httpx variant below for async code.

Wrap it once#

import functools
import os
import uuid
from typing import Callable, ParamSpec, TypeVar

import requests

BASE = os.environ.get("DRUMBEATS_BASE_URL", "https://api.drumbeats.io/v1")
PING_TIMEOUT = 3
MAX_PAYLOAD = 20_000

MONITORS = {
    "daily_backup": "11111111-2222-3333-4444-555555555555",
    "hourly_sync": "66666666-7777-8888-9999-aaaaaaaaaaaa",
    "newsletter_send": "bbbbbbbb-cccc-dddd-eeee-ffffffffffff",
}

P = ParamSpec("P")
R = TypeVar("R")


def _ping(monitor: str, event: str, run_id: str | None = None, payload: str | None = None) -> None:
    """Best effort. Never raises, never blocks longer than PING_TIMEOUT."""
    url = f"{BASE}/ping/{monitor}/{event}"
    params = {"run_id": run_id} if run_id else {}
    try:
        if payload is None:
            requests.get(url, params=params, timeout=PING_TIMEOUT)
        else:
            requests.post(
                url,
                params=params,
                json={"payload": payload[-MAX_PAYLOAD:]},
                timeout=PING_TIMEOUT,
            )
    except requests.RequestException:
        pass


def with_monitor(key: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
    monitor_id = MONITORS[key]

    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            run_id = f"{key}-{uuid.uuid4()}"
            _ping(monitor_id, "start", run_id=run_id)
            try:
                result = fn(*args, **kwargs)
                _ping(monitor_id, "success", run_id=run_id)
                return result
            except Exception:
                import traceback
                _ping(monitor_id, "failure", run_id=run_id, payload=traceback.format_exc())
                raise
        return wrapper
    return decorator

import functools
import os
import uuid
from typing import Callable, ParamSpec, TypeVar

import requests

BASE = os.environ.get("DRUMBEATS_BASE_URL", "https://api.drumbeats.io/v1")
PING_TIMEOUT = 3
MAX_PAYLOAD = 20_000

MONITORS = {
    "daily_backup": "11111111-2222-3333-4444-555555555555",
    "hourly_sync": "66666666-7777-8888-9999-aaaaaaaaaaaa",
    "newsletter_send": "bbbbbbbb-cccc-dddd-eeee-ffffffffffff",
}

P = ParamSpec("P")
R = TypeVar("R")


def _ping(monitor: str, event: str, run_id: str | None = None, payload: str | None = None) -> None:
    """Best effort. Never raises, never blocks longer than PING_TIMEOUT."""
    url = f"{BASE}/ping/{monitor}/{event}"
    params = {"run_id": run_id} if run_id else {}
    try:
        if payload is None:
            requests.get(url, params=params, timeout=PING_TIMEOUT)
        else:
            requests.post(
                url,
                params=params,
                json={"payload": payload[-MAX_PAYLOAD:]},
                timeout=PING_TIMEOUT,
            )
    except requests.RequestException:
        pass


def with_monitor(key: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
    monitor_id = MONITORS[key]

    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(fn)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            run_id = f"{key}-{uuid.uuid4()}"
            _ping(monitor_id, "start", run_id=run_id)
            try:
                result = fn(*args, **kwargs)
                _ping(monitor_id, "success", run_id=run_id)
                return result
            except Exception:
                import traceback
                _ping(monitor_id, "failure", run_id=run_id, payload=traceback.format_exc())
                raise
        return wrapper
    return decorator

from monitoring.drumbeats import with_monitor

@with_monitor("daily_backup")
def run_daily_backup() -> None:
    dump_database()
    upload_to_s3()

from monitoring.drumbeats import with_monitor

@with_monitor("daily_backup")
def run_daily_backup() -> None:
    dump_database()
    upload_to_s3()

Note payload[-MAX_PAYLOAD:] rather than payload[:MAX_PAYLOAD]. A Python traceback puts the actual exception on the last line, so keeping the tail is what you want. Drumbeats truncates from the front if you send more than your plan allows, which would throw away exactly the part you need.

The async variant#

import httpx

async def _aping(
    client: httpx.AsyncClient,
    monitor: str,
    event: str,
    run_id: str,
    payload: str | None = None,
) -> None:
    url = f"{BASE}/ping/{monitor}/{event}"
    try:
        if payload is None:
            await client.get(url, params={"run_id": run_id}, timeout=3)
        else:
            await client.post(
                url,
                params={"run_id": run_id},
                json={"payload": payload[-MAX_PAYLOAD:]},
                timeout=3,
            )
    except httpx.HTTPError:
        pass

import httpx

async def _aping(
    client: httpx.AsyncClient,
    monitor: str,
    event: str,
    run_id: str,
    payload: str | None = None,
) -> None:
    url = f"{BASE}/ping/{monitor}/{event}"
    try:
        if payload is None:
            await client.get(url, params={"run_id": run_id}, timeout=3)
        else:
            await client.post(
                url,
                params={"run_id": run_id},
                json={"payload": payload[-MAX_PAYLOAD:]},
                timeout=3,
            )
    except httpx.HTTPError:
        pass

What happens when it breaks#

Situation	What Drumbeats sees	What to do
The function raises	`failure` with the traceback attached	Nothing. The decorator handles it
`SystemExit` or `KeyboardInterrupt`	Nothing. Neither inherits from `Exception`	Catch `BaseException` in the wrapper if you need these reported
The worker is `SIGKILL`ed or OOM-killed	A `start` with no finish	Set `max_duration_seconds` so the hang is caught server-side
A `multiprocessing` child dies	Nothing, unless the parent raises	Ping from the parent, after joining the pool
A generator job is never fully consumed	`success` fires while work remains	Wrap the consumption, not the generator function

How you get alerted#

A failure opens a FAILED incident once failure_tolerance is reached, and pages every notification group on the monitor. The traceback appears as a preview in the alert and in full on the timeline, which usually means you can diagnose without opening a shell.

Framework notes#

Celery. Apply the decorator to the function body, inside the task, rather than outside @celery.task. Stacking it above the task decorator interferes with Celery's binding.

Airflow. Pass the monitor key through op_kwargs and wrap inside the python_callable. Airflow's own run ID makes an excellent run_id, since it links the Drumbeats run straight back to the DAG run.

Django. Management commands wrap cleanly with the decorator. For django-crontab or similar, wrap the command's handle method.

FastAPI background tasks. The wrapper works unchanged. They run on the same event loop, so use the async variant if the job itself is async.

Next#

Production hardening for the language-agnostic patterns. Event-driven pings for concurrent workers. Monitor types if you have not picked one yet. Alternatives if you are still choosing a vendor.