Task Badger Python SDK

On this page, we get you up and running with Task Badger's Python SDK.

Install

Install the SDK using uv or your favorite package manager:

uv pip install taskbadger

Configure

The SDK must be configured before you can use it to interact with the APIs.

import taskbadger

taskbadger.init(
    token="YOUR_API_KEY",
    tags={"environment": "production"},
)

When using a Project API Key, the organization and project are detected automatically from the key. You can find your API key on the project settings page in the Task Badger dashboard.

Name	Description
token	Project API key. The organization and project are extracted from the key automatically.
tags	Global tags which are added to all tasks.
systems	System integrations such as Celery
before_create	A function that is called before a task is created. See Before Create Callback
context_providers	Providers that attach extra data to a task when it errors. See Error Context Providers
organization_slug	The organization identifier. Only required for legacy API keys.
project_slug	The project identifier. Only required for legacy API keys.

If you attempt to use the SDK without configuring it you will get an error. To avoid this you can use the safe functions which will log any errors to the taskbadger logger.

Tip

Tags provided here will be applied to all tasks created using the SDK. If you need to add tags to individual tasks you can do so using the create and update methods or the task.add_tag method. Tags added manually will override the global tags.

Usage

The SDK provides a Task class which offers a convenient interface to the API.

Tasks are created by calling the Task.create method:

from taskbadger import Task

# create a new task with custom data and tags
task = Task.create(
    "task name",
    data={
        "custom": "data"
    },
    tags={"tenant": "acme"}
)

Alternatively a task may be retrieved via the tasks ID:

from taskbadger import Task

task = Task.get(task_id)

The task object provides methods for updating the properties of a task and adding custom data.

Listing tasks

list_tasks returns a TaskList, a single page of tasks that you can iterate over directly. The tasks it yields are ordinary Task objects, so they can be updated in place:

import taskbadger
from taskbadger import StatusEnum

for task in taskbadger.list_tasks(page_size=50):
    if task.status == StatusEnum.PENDING:
        task.canceled()

The next_ and previous attributes hold the URL of the adjacent page, or None if there isn't one. To fetch the next page, pass its cursor query parameter back to list_tasks:

from urllib.parse import parse_qs, urlparse

page = taskbadger.list_tasks(page_size=100)
while True:
    for task in page:
        ...
    if not page.next_:
        break
    cursor = parse_qs(urlparse(page.next_).query)["cursor"][0]
    page = taskbadger.list_tasks(page_size=100, cursor=cursor)

Changed in v2.5.1

list_tasks previously returned the generated PaginatedTaskList whose results were taskbadger.internal.models.Task objects, without the SDK's update() / safe_update() methods. It now returns a TaskList of taskbadger.Task objects.

A TaskList also has a length, so an empty page is falsy where PaginatedTaskList was always truthy. If you have code like if taskbadger.list_tasks(...):, note that it now tests whether the page has any tasks in it.

Parent and child tasks

Since v2.5.0

A task can be nested under another task by passing the parent's ID as the parent field:

from taskbadger import Task

parent = Task.create("import")
child = Task.create("import.chunk", parent=parent.id)

Tasks nest a single level deep, so parent must be the ID of a task that is not itself a child. A task's parent can not be changed once it has been set. Both rules are enforced by the API rather than by the SDK, so breaking either one surfaces as a failed request.

Use list_tasks to fetch the children of a task:

import taskbadger

for child in taskbadger.list_tasks(parent=parent.id):
    print(child.name, child.status)

Task.create and create_task only nest a task when you pass parent explicitly. The function decorator, Celery and Procrastinate integrations do it for you: a task enqueued while another tracked task is running is nested under it automatically. Because nesting is capped at one level, a task enqueued from within a child becomes a sibling of that child rather than a grandchild.

Each integration excludes a few cases from the automatic nesting — see the linked pages for what does and doesn't get nested.

Connection management

The SDK will open a new connection for each request and close it when the request is complete. For instances where you wish to make multiple requests you can use the taskbadger.Session context manager:

from taskbadger import Session

with Session() as session:
    task = Task.create("my task")
    task.update(status="success")

If you are using the function decorator or the Celery integration, session management is handled automatically within the body of the function or Celery task.

Scope

The SDK provides the taskbadger.current_scope context manager which can be used to set custom data and modify tags for the duration of the context. The content of the scope will be merged with any custom task data passed directly to any of the other API methods.

import socket
import taskbadger

with taskbadger.current_scope() as scope:
    scope["hostname"] = socket.gethostname()
    scope.tag({"tenant": "acme"})

A common use case for this is to add request scoped context in frameworks like Django or Flask using a custom middleware. Here's an example for Django:

import taskbadger

def taskbadger_scope_middleware(get_response):
    def middleware(request):
        with taskbadger.current_scope() as scope:
            scope["user"] = request.user.username
            scope.tag({"tenant": request.tenant.slug})
            return get_response(request)

    return middleware

Note

The data passed directly to the API will take precedence over the data in the current scope. If the same key is present in the current scope as well as the data passed in directly, the value in the data passed directly will be used. The same applies to tags.

Before Create Callback

The before_create parameter in the taskbadger.init function allows you to define a function that will be called before a task is created. This function will be passed the task data as a dictionary and should return the modified task data.

def before_create(task_data: dict) -> dict:
    data = task_data.setdefault("data", {})
    data["custom"] = "data"

    tags = task_data.setdefault("tags", {})
    tags["tenant"] = "acme"

    return task_data

Since v1.5.0

Error Context Providers

Since v2.5.0

Context providers attach extra data to a task when it errors, for example a link back to the system that reported the exception. They are consulted whenever a tracked task fails — via the function decorator, or the Celery or Procrastinate integrations — and whatever a provider returns is stored on the task data under the provider's identifier.

You can also call Task.error(exception=...) yourself, but read the caveat below first.

Task.error() on its own has no baseline

Those three integrations record the state of each provider as the task starts, so a provider can tell context belonging to this task from context left over from something unrelated.

Calling task.error(exception=...) directly does consult the providers, but nothing took that baseline, so they have nothing to compare against. With SentryContextProvider that means the task gets a link to whatever sentry_sdk.last_event_id() happens to be — quite possibly a stale, unrelated issue.

Prefer letting an integration own the error path. If you must call Task.error yourself and the link matters, pass the context in explicitly via data instead of relying on a provider.

Providers are registered with taskbadger.init:

import taskbadger
from taskbadger.context_providers.sentry import SentryContextProvider

taskbadger.init(
    token="YOUR_API_KEY",
    context_providers=[SentryContextProvider(organization_slug="acme")],
)

A provider that raises is logged to the taskbadger logger and skipped, so it can never break the task update.

Sentry

SentryContextProvider links a failed task to the Sentry issue for the same exception. It needs the sentry-sdk package, available via the sentry extra. If the package isn't installed the provider is a silent no-op — it adds no context and reports no error, so install the extra:

uv add 'taskbadger[sentry]'
# or: pip install 'taskbadger[sentry]'

import taskbadger
from taskbadger.context_providers.sentry import SentryContextProvider

taskbadger.init(
    token="YOUR_API_KEY",
    context_providers=[SentryContextProvider(organization_slug="acme")],
)

Failed tasks then carry the Sentry event ID in their data, plus a link to the issue when organization_slug is given:

{
  "exception": "bad input",
  "sentry": {
    "event_id": "5f8a...",
    "url": "https://sentry.io/organizations/acme/issues/?query=5f8a..."
  }
}

Pass base_url if you are running a self-hosted Sentry.

The exception value is str(exception), except on the Celery path, where Celery's own exception info is used and the value is a full traceback.

Note

The provider does not report the exception to Sentry itself. It assumes your application already does that (e.g. via a framework integration) and reads back the event ID, which avoids reporting the same exception twice.

To avoid linking to an unrelated event, the provider records Sentry's current event ID when the task starts and only attaches context if it has changed by the time the task errors. So no context is added when the exception never reaches Sentry, or when Sentry saw nothing new while the task ran.

Custom providers

To attach context from another system, subclass ContextProvider, set an identifier and implement capture_error_context:

from taskbadger.context_providers import ContextProvider


class RequestIdProvider(ContextProvider):
    identifier = "request"

    def capture_error_context(self, exception, snapshot=None):
        return {"id": get_current_request_id()}

Providers that read back state captured by another system, rather than capturing it themselves, should also implement snapshot. It is called when a tracked task starts and its return value is passed back as snapshot, so the provider can tell a fresh capture from a stale one left over from something unrelated.

Python Reference

taskbadger.Task

The Task class provides a convenient Python API to interact with Task Badger tasks.

get `classmethod`

get(task_id: str) -> Task

Get an existing task

create `classmethod`

create(
    name: str,
    status: StatusEnum = StatusEnum.PENDING,
    value: int = None,
    value_max: int = None,
    data: dict = None,
    max_runtime: int = None,
    stale_timeout: int = None,
    actions: list[Action] = None,
    monitor_id: str = None,
    tags: dict[str, str] = None,
    queue: str = None,
    external_id: str = None,
    parent: str = None,
) -> Task

Create a new task

See taskbadger.create_task for more information.

pre_processing

pre_processing()

Update the task status to pre_processing.

starting

starting()

Update the task status to processing and set the value to 0.

processing

processing(value: int = None)

Update the task status to processing and set the value.

post_processing

post_processing(value: int = None)

Update the task status to post_processing and set the value.

success

success(value: int = None)

Update the task status to success and set the value.

error

error(value: int = None, data: dict = None, exception: BaseException = None)

Update the task status to error and set the value and data.

If exception is given, it's passed to any configured context providers (e.g. Sentry, see taskbadger.context_providers) and the result merged into data. Called on its own (outside @track or the Celery/Procrastinate integrations), providers have no baseline to compare against, so e.g. SentryContextProvider will report whatever sentry_sdk.last_event_id() currently is.

canceled

canceled()

Update the task status to cancelled

update_status

update_status(status: StatusEnum)

Update the task status

increment_value

increment_value(amount: int)

Increment the task progress by adding the specified amount to the current value. If the task value is not set it will be set to amount.

update_value

update_value(
    value: int, value_step: int = None, rate_limit: int = None
) -> bool

Update task progress.

Parameters:

Name	Type	Description	Default
`value`	`int`	The new value to set.	required
`value_step`	`int`	The minimum change in value required to trigger an update.	`None`
`rate_limit`	`int`	The minimum interval between updates in seconds.	`None`

Returns:

Name	Type	Description
`bool`	`bool`	True if the task was updated, False otherwise

If either value_step or rate_limit is set, the task will only be updated if the specified conditions are met. If both are set, the task will be updated if either condition is met.

set_value_max

set_value_max(value_max: int)

Set the value_max.

update

update(
    name: str = None,
    status: StatusEnum = None,
    value: int = None,
    value_max: int = None,
    data: dict = None,
    max_runtime: int = None,
    stale_timeout: int = None,
    actions: list[Action] = None,
    tags: dict[str, str] = None,
    queue: str = None,
    external_id: str = None,
    parent: str = None,
    data_merge_strategy: Any = None,
)

Generic update method used to update any of the task fields.

This can also be used to add actions.

See taskbadger.update_task for more information.

add_actions

add_actions(actions: list[Action])

Add actions to the task.

Deprecated: per-task actions are deprecated in favor of project-level actions and will be removed in a future release.

tag

tag(tags: dict[str, str])

Add tags to the task.

ping

ping(rate_limit=None) -> bool

Update the task without changing any values. This can be used in conjunction with 'stale_timeout' to indicate that the task is still running.

Parameters:

Name	Type	Description	Default
`rate_limit`		The minimum interval between pings in seconds. If set this will only update the task if the last update was more than `rate_limit` seconds ago.	`None`

Returns:

Name	Type	Description
`bool`	`bool`	True if the task was updated, False otherwise

Low level functions

In addition to the taskbadger.Task class. There are also a number of functions that provide lower level access to the API:

taskbadger.get_task

get_task(task_id: str) -> Task

Fetch a Task from the API based on its ID.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	The ID of the task to fetch.	required

taskbadger.create_task

create_task(
    name: str,
    status: StatusEnum = StatusEnum.PENDING,
    value: int = None,
    value_max: int = None,
    data: dict = None,
    max_runtime: int = None,
    stale_timeout: int = None,
    actions: list[Action] = None,
    monitor_id: str = None,
    tags: dict[str, str] = None,
    queue: str = None,
    external_id: str = None,
    parent: str = None,
) -> Task

Create a Task.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the task.	required
`status`	`StatusEnum`	The task status.	`PENDING`
`value`	`int`	The current 'value' of the task.	`None`
`value_max`	`int`	The maximum value the task is expected to achieve.	`None`
`data`	`dict`	Custom task data.	`None`
`max_runtime`	`int`	Maximum expected runtime (seconds).	`None`
`stale_timeout`	`int`	Maximum allowed time between updates (seconds).	`None`
`actions`	`list[Action]`	Task actions. Deprecated: use project-level actions instead.	`None`
`monitor_id`	`str`	ID of the monitor to associate this task with.	`None`
`tags`	`dict[str, str]`	Dictionary of namespace -> value tags.	`None`
`queue`	`str`	Name of the queue the task is from.	`None`
`external_id`	`str`	Identifier from the originating system (e.g. Celery task ID) for correlating with logs.	`None`
`parent`	`str`	ID of the parent task. Tasks nest a single level deep, so this must be the ID of a task that is not itself a child. The Celery and Procrastinate integrations set this automatically for tasks enqueued from within a tracked task.	`None`

Returns:

Name	Type	Description
`Task`	`Task`	The created Task object.

taskbadger.update_task

update_task(
    task_id: str,
    name: str = None,
    status: StatusEnum = None,
    value: int = None,
    value_max: int = None,
    data: dict = None,
    max_runtime: int = None,
    stale_timeout: int = None,
    actions: list[Action] = None,
    tags: dict[str, str] = None,
    queue: str = None,
    external_id: str = None,
    parent: str = None,
) -> Task

Update a task. Requires only the task ID and fields to update.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	The ID of the task to update.	required
`name`	`str`	The name of the task.	`None`
`status`	`StatusEnum`	The task status.	`None`
`value`	`int`	The current 'value' of the task.	`None`
`value_max`	`int`	The maximum value the task is expected to achieve.	`None`
`data`	`dict`	Custom task data.	`None`
`max_runtime`	`int`	Maximum expected runtime (seconds).	`None`
`stale_timeout`	`int`	Maximum allowed time between updates (seconds).	`None`
`actions`	`list[Action]`	Task actions. Deprecated: use project-level actions instead.	`None`
`tags`	`dict[str, str]`	Dictionary of namespace -> value tags.	`None`
`queue`	`str`	Name of the queue the task is from.	`None`
`external_id`	`str`	Identifier from the originating system (e.g. Celery task ID) for correlating with logs.	`None`
`parent`	`str`	ID of the parent task. Can only be set on a task that doesn't already have a parent — the API rejects an attempt to change one.	`None`

Returns:

Name	Type	Description
`Task`	`Task`	The updated Task object.

taskbadger.list_tasks

list_tasks(
    page_size: int = None, cursor: str = None, parent: str = None
) -> TaskList

List tasks.

Parameters:

Name	Type	Description	Default
`page_size`	`int`	Number of results to return per page.	`None`
`cursor`	`str`	Pagination cursor.	`None`
`parent`	`str`	Only return the children of this task.	`None`

taskbadger.TaskList

A page of tasks as returned by taskbadger.list_tasks.

Iterating over a TaskList yields taskbadger.Task objects:

for task in taskbadger.list_tasks():
    print(task.name)

results `property`

results: list[Task]

The tasks in this page.

Context Provider Reference

taskbadger.context_providers

ContextProvider

Base class for pluggable providers that attach extra context to a task's data when it errors, e.g. so the TaskBadger UI can link out to an external system (Sentry, Rollbar, etc.).

Registered via init(context_providers=[...]) and consulted whenever a tracked task (via @track, the Celery/Procrastinate integrations) errors.

Implementations that read back state some other system captured on its own (rather than capturing it themselves, which risks duplicate reporting) should override snapshot to record a baseline when the task starts, so capture_error_context can tell a fresh capture from a stale one left over from something unrelated.

snapshot

snapshot()

Called when a tracked task starts, before user code runs. Return an opaque value to be passed back as snapshot to capture_error_context. Default: None (no baseline tracking).

capture_error_context

capture_error_context(exception: BaseException, snapshot=None) -> dict | None

Return extra context for exception, or None if there is nothing to add. The result is stored under data[self.identifier].

snapshot is whatever this provider's snapshot() returned when the task started.

taskbadger.context_providers.sentry.SentryContextProvider

Bases: ContextProvider

Links a failed task to the corresponding Sentry issue.

Reads back sentry_sdk.last_event_id() rather than capturing the exception itself, on the assumption the surrounding system already reports its own exceptions to Sentry (e.g. via a framework integration). To avoid linking to a stale event left over from something unrelated, a snapshot is taken when the task starts and the event id is only reported if it changed by the time the task errors.

Requires the sentry-sdk package; a no-op if it isn't installed.

Safe functions

For instances where you prefer not to handle errors you can use the following function which will handle all errors and log them to the taskbadger logger.

These can also be used safely in instances where the API has not been configured via taskbadger.init.

taskbadger.create_task_safe

create_task_safe(name: str, **kwargs: P.kwargs) -> Task | None

Safely create a task. Any errors are handled and logged.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the task.	required
`**kwargs`	`kwargs`	See taskbadger.create_task	`{}`

Returns:

Type	Description
`Task \| None`	The created task or None

taskbadger.update_task_safe

update_task_safe(task_id: str, **kwargs: P.kwargs) -> Task | None

Safely update a task. Any errors are handled and logged.

Parameters:

Name	Type	Description	Default
`task_id`	`str`	The ID of the task to update.	required
`**kwargs`	`kwargs`	See taskbadger.update_task	`{}`

Returns:

Type	Description
`Task \| None`	The updated task or None

Task Badger Python SDK

Install

Configure

Usage

Listing tasks

Parent and child tasks

Connection management

Scope

Before Create Callback

Error Context Providers

Sentry

Custom providers

Python Reference

taskbadger.Task

get classmethod

create classmethod

pre_processing

starting

processing

post_processing

success

error

canceled

update_status

increment_value

update_value

set_value_max

update

add_actions

tag

ping

Low level functions

taskbadger.get_task

taskbadger.create_task

taskbadger.update_task

taskbadger.list_tasks

taskbadger.TaskList

results property

Context Provider Reference

taskbadger.context_providers

ContextProvider

snapshot

capture_error_context

taskbadger.context_providers.sentry.SentryContextProvider

Safe functions

taskbadger.create_task_safe

taskbadger.update_task_safe

get `classmethod`

create `classmethod`

results `property`