import taskbadger

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

import taskbadger
from celery import Celery

app = Celery("tasks")

@app.task(bind=True, base=taskbadger.Task)
def my_task(self):
    task = self.taskbadger_task
    for i in range(1000):
        do_something(i)
        if i % 100 == 0:
            task.update(value=i, value_max=1000)
    task.success(value=1000)

from taskbadger import Task, Action, EmailIntegration, WebhookIntegration

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

# update the task status to 'processing' and set the value to 0
task.started()
try:
   for i in range(100):
      do_something(i)
      if i!= 0 and i % 10 == 0:
         # update the progress of the task
         task.update_progress(i)
except Exception as e:
    # record task errors
    task.error(data={
        "error": str(e)
    })
    raise

# record task success
task.success()