Python integration

This guide shows how to wire Drumbeats into a Python project. By the end you have a single helper that wraps any job with start / success / failure pings — no per-job boilerplate, no per-environment changes.

The patterns below use requests (the standard for synchronous Python). The same shape works with httpx for async code; replace requests.get with await client.get where noted.

Quick example

import requests

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

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

import requests

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

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

POST the failure with a payload so the error message lands on the incident timeline. Success can stay GET — it does not need a body.

Centralized setup

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

BASE = os.environ.get("DRUMBEATS_BASE_URL", "https://api.drumbeats.io/v1")

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 ping — never raises, never blocks the caller."""
    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=3)
        else:
            requests.post(url, params=params, json={"payload": payload}, timeout=3)
    except requests.RequestException:
        # Side-channel; do not propagate.
        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 as exc:
                _ping(monitor_id, "failure", run_id=run_id, payload=repr(exc))
                raise
        return wrapper
    return decorator

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

BASE = os.environ.get("DRUMBEATS_BASE_URL", "https://api.drumbeats.io/v1")

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 ping — never raises, never blocks the caller."""
    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=3)
        else:
            requests.post(url, params=params, json={"payload": payload}, timeout=3)
    except requests.RequestException:
        # Side-channel; do not propagate.
        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 as exc:
                _ping(monitor_id, "failure", run_id=run_id, payload=repr(exc))
                raise
        return wrapper
    return decorator

Apply the decorator to any job function:

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()

Add a new monitor: drop the UUID into MONITORS, apply @with_monitor("key"). That is the entire change.

Error handling

A few cases need explicit care:

• Celery tasks. Apply the decorator inside the task function — @with_monitor outside of @celery.task confuses Celery's binding. Use @celery.task first, then call with_monitor("...")(self.run) inside.
• Async code (asyncio). The synchronous decorator works for asyncio if you asyncio.run(...) at the boundary. For a fully-async pipeline, replace the body with httpx.AsyncClient and decorate with the async-aware variant (see below).
• Process-level crashes. A worker that's killed by SIGKILL never sends failure — pair the monitor's max_duration_seconds config with the start ping so Drumbeats catches the hung run regardless.

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}, 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}, timeout=3)
    except httpx.HTTPError:
        pass

Production patterns

• Time-bound pings. The timeout=3 argument on requests.get / requests.post keeps a slow Drumbeats request from blocking your job. Three seconds is well under any sensible alerting threshold.
• Cap payload size. Drumbeats stores up to 25KB per payload at no extra beat cost (see Beats and usage). Truncate stack traces before sending: payload[:20_000].
• Progress logs. For long-running jobs, send log pings between work units. They appear inline on the incident timeline so the on-call engineer can see exactly where the job hung.
• Multi-env via slug. Use /v1/s-ping/{project_id}/{slug}/{event} if the same job runs in staging and prod. One slug, two project IDs, two monitor IDs — see Ping API: scheduled pings.

Framework-specific notes

• Django + Celery. Apply with_monitor inside the task function so the decorator wraps the actual work, not the task setup.
• Airflow. Pass the monitor ID via op_kwargs and call the wrapper inside the PythonOperator's python_callable. The Airflow run ID makes a great run_id.
• FastAPI background tasks. The wrapper works as-is for BackgroundTasks — they run on the same event loop as the request.

Related guides

• Monitor types — pick Cron, Heartbeat, or Event-driven before wiring pings.
• Ping API — endpoint and parameter reference.
• Production hardening — retries, timeouts, observability for any language.
• Alternatives — how Drumbeats compares to Cronitor and Healthchecks.io for Python stacks.