GUIDE 17 · BACKGROUND WORK6 min read

Background tasks and durable queues

Run small post-response tasks or persist jobs for a separate worker process.

IN THIS GUIDE

Work through each step in order. The examples use the generated LenAPI project structure and are designed to be adapted to your app.

Use short-lived response tasks

BackgroundTasks runs work after the response is sent. Sync callbacks are run in a worker thread; async callbacks are awaited. This is in-process and is not a durable queue.

Python

from lenapi import BackgroundTasks

def send_receipt(email: str):
    ...

@app.post("/orders")
def create_order(background_tasks: BackgroundTasks):
    order = save_order()
    background_tasks.add_task(send_receipt, order.email)
    return {"id": order.id}

Persist jobs in a worker queue

BackgroundQueue supports database, Redis, and process-cache backends. Configure max_attempts and retry_delay when creating the queue: max_attempts is the total number of executions, and retry_delay is the base delay in seconds. Failed jobs wait retry_delay × attempt number before the next attempt. Pass delay_seconds to enqueue() to schedule the first attempt for later; it is not passed to the task function. Database and Redis jobs are delivered at least once, so task handlers should be idempotent.

Python

from lenapi import BackgroundQueue

queue = BackgroundQueue(
    app.database,
    max_attempts=3,
    retry_delay=10,  # seconds between retries, multiplied by attempt number
)

@queue.task("send_welcome")
async def send_welcome(email: str):
    ...

job_id = await queue.enqueue(
    "send_welcome",
    "ada@example.com",
    delay_seconds=300,  # run in five minutes
)

Run the worker separately

Export the queue and register tasks in main.py. Start a separate worker process with the project CLI. The worker applies its queue migration before consuming jobs.

Python

python lenora.py background-worker

TIP

The cache/memory backend is ephemeral and cannot be shared between an API process and a separate worker process.