Skip to content

Task Badger Python SDK

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

Install

Install the SDK using pip or your favorite pacakge manager: 

pip install --upgrade taskbadger

Configure

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

import taskbadger

taskbadger.init(
    organization_slug="my-org",
    project_slug="my-project",
    token="***"
)

Details about these configuration parameters can be found here

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.

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, Action, EmailIntegration

# create a new task with custom data and an action definition
task = Task.create(
    "task name",
    data={
        "custom": "data"
    },
    actions=[
        Action(
            trigger="*/10%,success,error",
            integration=EmailIntegration(to="me@example.com")
        )
    ]
)

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, adding custom data and adding actions.

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

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
            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.

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

Create a new task

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)

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

canceled 

canceled()

Update the task status to cancelled

update_status 

update_status(status: StatusEnum)

Update the task status

increment_progress 

increment_progress(amount: int)

Increment the task progress. If the task value is not set it will be set to amount.

update_progress 

update_progress(value: int)

Update task progress.

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, data_merge_strategy: Any = None)

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

This can also be used to add actions.

add_actions 

add_actions(actions: List[Action])

Add actions to the task.

ping 

ping()

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

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

 None
monitor_id str

ID of the monitor to associate this task with.

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

 None

Returns:

Name Type Description
Task Task

The updated Task object.

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) -> Optional[Task]

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
Optional[Task]

The created task or None

taskbadger.update_task_safe 

update_task_safe(task_id: str, **kwargs: P.kwargs) -> Optional[Task]

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
Optional[Task]

The updated task or None