TaskTiger Documentation

To get started with TaskTiger, follow these three steps:

1. Define your tasks: Create a Python file (e.g., tasks.py) containing the functions you want to run asynchronously.
2. Queue the task: Use a TaskTiger instance and its .delay() method to add the task to the Redis queue.
3. Run a worker: Execute the tasktiger command in your terminal. Ensure your PYTHONPATH includes the directory containing your task definitions so the worker can import them.

Note: TaskTiger uses Redis as its backend.

# tasks.py
def my_task():
    print('Hello')

# In your application code:
import tasktiger, tasks
tiger = tasktiger.TaskTiger()
tiger.delay(tasks.my_task)

# Run the worker
% PYTHONPATH=. tasktiger

TaskTiger supports Rollbar integration to log task errors. To enable this, initialize Rollbar and add a StructlogRollbarHandler (from tasktiger.rollbar) to the TaskTiger logger. The handler accepts a prefix string for all reported messages.

import logging
import rollbar
import sys
from tasktiger import TaskTiger
from tasktiger.rollbar import StructlogRollbarHandler

tiger = TaskTiger(setup_structlog=True)

rollbar.init(ROLLBAR_API_KEY, APPLICATION_ENVIRONMENT, 
             allow_logging_basic_config=False)
rollbar_handler = StructlogRollbarHandler('TaskTiger')
rollbar_handler.setLevel(logging.ERROR)
tiger.log.addHandler(rollbar_handler)

tiger.run_worker_with_args(sys.argv[1:])

TaskTiger allows you to specify task behavior using either a @tiger.task decorator or the tiger.delay() method.

• Task Decorator: Used to define default behavior for a task. Options set here are the defaults for that task type.
• tiger.delay(): Used when queueing a task. Options passed to delay() will override any settings defined in the task decorator.

Important: When using delay(), the task must be defined in a module other than the one being executed (it cannot be in the __main__ module).

@tiger.task(queue='myqueue', unique=True)
def my_task():
    print('Hello')

# The task will be queued in "otherqueue", even though the task decorator
# says "myqueue".
tiger.delay(my_task, queue='otherqueue')

You can deploy TaskTiger and its required Redis backend using Docker Compose. The setup defines two services:

1. redis: Uses the redis:7.0.9 image and exposes port 6379.
2. tasktiger: Builds from the local Dockerfile and connects to the Redis service.

To connect the TaskTiger service to Redis, the REDIS_HOST environment variable is set to redis.

version: "3.7"
services:
  redis:
    image: redis:7.0.9
    expose:
      - 6379
  tasktiger:
    build:
      context: .
      dockerfile: Dockerfile
    environment:
      REDIS_HOST: redis
    volumes:
      - .:/src
    depends_on:
      - redis

To prevent the error queue from growing indefinitely in Redis, use the purge_errored_tasks method. This can be automated by creating a periodic task using the periodic helper.

from tasktiger import TaskTiger, periodic
import datetime

tiger = TaskTiger()

@tiger.task(schedule=periodic(hours=1))
def purge_errored_tasks():
    tiger.purge_errored_tasks(
        limit=1000,
        last_execution_before=(
            datetime.datetime.utcnow() - datetime.timedelta(weeks=12)
        )
    )

If you need to set up a specific environment (e.g., via a manage.py script) before launching workers, use the run_worker_with_args method. This allows you to pass command-line arguments through your script to the TaskTiger worker.

import sys
from tasktiger import TaskTiger

try:
    command = sys.argv[1]
except IndexError:
    command = None

if command == 'tasktiger':
    tiger = TaskTiger(setup_structlog=True)
    # Strip the "tasktiger" arg when running via manage, so we can run e.g. 
    # ./manage.py tasktiger --help
    tiger.run_worker_with_args(sys.argv[2:])
    sys.exit(0)

While simple functions can be queued using tiger.delay(func), you can use the @tiger.task() decorator to enable a more concise, Celery-compatible syntax.

When a task is decorated, you can call .delay() directly on the function object itself rather than through the TaskTiger instance.

# tasks.py
import tasktiger
tiger = tasktiger.TaskTiger()

@tiger.task()
def my_task(name, n=None):
    print('Hello', name)

# Queuing the task:
# Option 1: Standard syntax
tiger.delay(my_task, args=('John',), kwargs={'n': 1})

# Option 2: Celery-compatible syntax (requires @tiger.task())
my_task.delay('John', n=1)

You can access information about the task currently being executed from within the task function itself:

• Non-batch tasks: Use tiger.current_task to get the current Task instance.
• Batch tasks: Use tiger.current_tasks to get a list of all Task instances currently being processed in that batch (in the same order they were passed).

from tasktiger import TaskTiger

tiger = TaskTiger()

@tiger.task(batch=True)
def my_task(args):
    for task in tiger.current_tasks:
        print(task.n_executions())