This is the official Python SDK for Task Badger.

For full documentation go to https://docs.taskbadger.net/python/.

pip install taskbadger

To use the taskbadger command-line tool, install the cli extra:

pip install 'taskbadger[cli]'

import taskbadger
from taskbadger.systems import CelerySystemIntegration

taskbadger.init(
    token="***",
    systems=[CelerySystemIntegration()],
    tags={"environment": "production"}
)

$ export TASKBADGER_API_KEY=***
$ taskbadger run "nightly-backup" -- ./backup.sh

API requests time out after 5 seconds by default. Override it with the timeout argument (seconds), or with the TASKBADGER_HTTP_TIMEOUT environment variable, which the CLI also honours:

taskbadger.init(token="***", timeout=30)

Pass an httpx.Timeout for finer control, or httpx.Timeout(None) to disable timeouts entirely.

The SDK includes optional support for the Procrastinate task queue.

Install with the extra:

pip install 'taskbadger[procrastinate]'

Opt a single task into tracking with the track decorator:

import procrastinate
from taskbadger.procrastinate import track, current_task

app = procrastinate.App(connector=...)

@track
@app.task(queue="default")
async def add(a, b):
    return a + b

@track(name="report", value_max=100, tags={"env": "prod"})
@app.task
async def report(rows):
    tb = current_task()
    for i, row in enumerate(rows):
        await process(row)
        if i % 10 == 0:
            tb.update(value=i)

To auto-track every task on an App, register the system integration:

import taskbadger
from taskbadger.systems.procrastinate import ProcrastinateSystemIntegration

taskbadger.init(
    token="***",
    systems=[ProcrastinateSystemIntegration(
        app=app,
        auto_track_tasks=True,
        includes=[r"myapp\..*"],
        excludes=[r"myapp\.cleanup\..*"],
        record_task_args=True,
    )],
)

• task.configure(...).defer(...) is not tracked. Procrastinate's configure() returns a separate JobDeferrer whose methods bypass our wrapper. Use task.defer(...) directly for tracked deferrals. Tasks deferred via configure().defer() will run normally but will not appear in TaskBadger.
• task.batch_defer* is not tracked. Same reason as configure().defer().
• Tasks added via app.add_tasks_from(blueprint) after ProcrastinateSystemIntegration is constructed are not auto-instrumented. Construct the integration after all blueprints are registered, or apply @track to those tasks explicitly.

A task with a stale_timeout is marked stale by Task Badger if it goes too long without an update. Set heartbeat_interval (seconds) to have the SDK ping the task for you while it runs, rather than updating it from the task body.

For Procrastinate, on the task or on ProcrastinateSystemIntegration(...):

@track(heartbeat_interval=60)
@app.task
async def slow_job():
    ...

For Celery, on CelerySystemIntegration(...), on the task, or per call with slow_job.apply_async(taskbadger_heartbeat_interval=60):

@app.task(base=taskbadger.Task, taskbadger_heartbeat_interval=60)
def slow_job():
    ...

Unless stale_timeout is given explicitly it is set to twice the interval. All running tasks are pinged from a single background thread, started the first time a task with a heartbeat runs.

Pass taskbadger_track=False to leave one execution untracked. This overrides auto-tracking as well as the taskbadger.Task base class:

noisy_job.apply_async(args, taskbadger_track=False)

Canvas primitives don't go through apply_async on the task itself, so pass it in the headers:

noisy_job.map(items).apply_async(headers={"taskbadger_track": False})