Task Badger Python Client

This is the official Python SDK for Task Badger.

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

Integration Tests

Getting Started

Install

pip install taskbadger

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

pip install 'taskbadger[cli]'

Client Usage

import taskbadger
from taskbadger.systems import CelerySystemIntegration

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

CLI Usage

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

Request timeout

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.

Procrastinate Integration

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,
    )],
)

Known limitations

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.

Keeping long-running tasks fresh

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.

Skipping tracking for a single Celery call

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

Metadata

Release files for taskbadger 2.6.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for taskbadger 2.6.0
File	Size	Uploaded
taskbadger-2.6.0.tar.gz	48.5 kB	Aug 11, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for taskbadger 2.6.0
File	Interpreter	ABI	Platform	Reset
taskbadger-2.6.0-py3-none-any.whl 84.1 kB Aug 11, 2026	Python 3	none	any	Details

Total release size: 132.6 kB

Release files / taskbadger-2.6.0.tar.gz

Download URL	taskbadger-2.6.0.tar.gz
Size	48.5 kB
Tags	Source
SHA-256 checksum How to use checksums	`cdbb49f792cca703b1877a025bfa8c0252e8b67e55411c574f81b977af9326bb`
BLAKE2b-256 checksum How to use checksums	`bf7e86276827270d8c2c4e04c4e0137598105462d638e19fe4c0fc4bd9a06369`
Upload date	Aug 11, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`uv/0.12.3 {"installer":{"name":"uv","version":"0.12.3","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}`

Release files / taskbadger-2.6.0-py3-none-any.whl

Download URL	taskbadger-2.6.0-py3-none-any.whl
Size	84.1 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`4337116d9810bca235de65c811f0545c06d92309b98e2020967aa51af5e73077`
BLAKE2b-256 checksum How to use checksums	`267faf79f9d6d38ee4ae1791ad71e11c9317a7f6a7c73fd1edaf0ab1ed339d3d`
Upload date	Aug 11, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`uv/0.12.3 {"installer":{"name":"uv","version":"0.12.3","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}`