tarsq

The background job runtime for Python. Tasks run in their own subprocess. Timeouts are hard kills. Crashed workers restart themselves.

Features

• @task decorator to register handlers — sync and async
• @schedule decorator for cron-based recurring tasks
• dispatch() to enqueue jobs from anywhere in your app
• status() to track job progress
• Each task runs in its own subprocess — timeouts enforced with a hard process kill
• Automatic retries with exponential backoff
• Per-worker context via on_startup
• Crashed worker auto-restart
• Stuck job recovery on startup

Requirements

• Python 3.11+
• Redis

Installation

pip install tarsq

Quick start

1. Define your tasks (myapp/tasks.py):

from tarsq import task

@task("send_email", timeout=10, max_retries=2)
def send_email(ctx, payload):
    print(f"Sending email to {payload['to']}")

@task("resize_image")
async def resize_image(ctx, payload):
    print(f"Resizing image at {payload['url']}")

2. Configure the worker (myapp/worker_settings.py):

from tarsq import WorkerSettings

class MyWorkerSettings(WorkerSettings):
    app = "myapp.tasks"
    workers = 4

3. Start the worker:

tarsq --settings myapp.worker_settings.MyWorkerSettings

Or without a settings class:

tarsq --app myapp.tasks --workers 4

4. Dispatch jobs from anywhere in your application:

from tarsq import dispatch, status

job_id = dispatch("send_email", payload={"to": "user@example.com"})

job = status(job_id)
print(job.status)  # queued | in_progress | completed | failed

The @task decorator

@task("send_email", timeout=30, max_retries=3)
def send_email(ctx, payload):
    ...

Both sync and async functions are supported. Each task receives:

• ctx — dict populated by on_startup, containing shared resources
• payload — the dict passed to dispatch()

Scheduled tasks

Use @schedule to register tasks that run automatically on a cron schedule:

from tarsq import schedule

@schedule("daily_report", cron="every day at 9am")
def daily_report(ctx, payload):
    generate_report()

@schedule("cleanup", cron="0 2 * * *")
def cleanup(ctx, payload):
    purge_old_records()

Built-in presets:

Standard 5-field cron expressions are also accepted.

WorkerSettings

Inheritance from tarsq's WorkerSettings is optional — only define the attributes you need.

Context & on_startup

on_startup runs inside each worker process after it spawns. Use it to initialise per-process resources and store them in ctx. Every task handler receives this ctx.

from tarsq import WorkerSettings

def startup(ctx):
    ctx["db"] = SessionLocal  # store the factory, not a live session

class MyWorkerSettings(WorkerSettings):
    app = "myapp.tasks"
    on_startup = startup

@task("create_user")
def create_user(ctx, payload):
    db = ctx["db"]()  # create a session inside the task
    ...
    db.close()

Note: ctx is pickled when passed to the task subprocess. Do not store live resources (db sessions, open connections) directly in ctx — store factories or classes instead. tarsq will warn you at startup if it detects unpicklable values.

Job status

from tarsq import status

job = status(job_id)

job.job_id      # UUID string
job.task        # task name
job.status      # "queued" | "in_progress" | "completed" | "failed"
job.retries     # number of retry attempts so far
job.created_at  # ISO 8601 timestamp
job.updated_at  # ISO 8601 timestamp of last status change

Returns None if no job with the given ID exists.

Environment variables

Variables can be set in a .env file in your project root.

CLI reference

tarsq [--settings <path>] [--app <module>] [--workers <n>]

CLI args take priority over WorkerSettings.

License

MIT