Background tasks

Table of contents

1. Introduction
2. Writing a task
3. Enqueueing a task
   1. Waiting for the commit
4. Checking a run
5. Reporting progress
   1. Showing progress in a command
6. Choosing where tasks run
7. Running a worker
8. Testing tasks
9. Advanced
10. Reference
    1. @task
    2. enqueue(target, kwargs, *, key=None, after_commit=None, timeout=None, app=None)
    3. TaskRun
    4. Settings
11. See also

---

Introduction

Work that should not hold a request, such as sending a mail, converting an upload or building a report, runs as a task. You mark a function with @task and enqueue it with the ids it needs. Where it runs, in the caller, in a thread or in a worker container, is the deployment’s choice, and your code stays the same.

from splent_framework.tasks import enqueue, task


@task(timeout=60, retries=2)
def build_report(report_id):
    ...


run = enqueue(build_report, {"report_id": report.id})

---

Writing a task

@task(timeout=60, retries=2, backoff=10)
def build_report(report_id):
    report = Report.query.get(report_id)
    report.pdf = render_pdf(report)
    db.session.commit()

The function stays a plain function, so a command or a seeder can still call it directly. Put it at the top level of its module. A worker finds it by its dotted path, and @task refuses a nested function or a method.

Make running it twice harmless. A retry or a second enqueue runs it again, so pass ids, read the rows inside and write the result whole.

Enqueueing a task

run = enqueue(build_report, {"report_id": report.id}, key=report.id)

Arguments must be JSON values in every mode, so a test fails on what a queue would refuse. key merges this run into one of the same task and key that has not started yet, and returns that one. Two uploads of the same file in a row then convert it once.

Waiting for the commit

db.session.add(report)
db.session.flush()                                 # report.id is known
enqueue(build_report, {"report_id": report.id})   # waits for the commit
db.session.commit()                                # now it starts

A task enqueued after a write you have not committed waits, as waiting. It starts at the commit, and is dropped if you roll back. Inside a savepoint it follows the savepoint. With nothing pending it starts at once.

Pass after_commit=True to wait for a commit in any open transaction, or False to never wait.

Checking a run

from splent_framework.tasks import get_run

run = get_run(run_id)
run.status     # waiting, queued, running, succeeded, failed, timed_out or dropped
run.attempts   # 1
run.error      # "ValueError: no such report"

get_run answers a run as it stands now, or None once it is forgotten. The traceback of a failure goes to the log, not to the run.

Reporting progress

from splent_framework.tasks import report_progress, task


@task(timeout=600)
def import_files(batch_id):
    files = batch_files(batch_id)
    for done, file in enumerate(files, start=1):
        ingest(file)
        report_progress(done, len(files), file.name)

report_progress(done, total=None, note=None) stores how far the task is on its run.

run = get_run(run_id)
run.progress          # 0.75, or None until the task names a total
run.progress_note     # "notes_2026.csv"

run.as_dict() includes it, ready for a page that asks every few seconds. Outside a task report_progress does nothing, so the same function runs in a request or a command.

Showing progress in a command

import click

from splent_framework.tasks.runs import reporting


def show(done, total, note):
    click.echo(f"{done}/{total or '?'} {note or ''}")


with reporting(show, every=1):
    import_files(batch_id)

When you call the task yourself, reporting(reporter, every=0.0) sends each report inside the block to your function, at most once every every seconds, and always the last one.

---

Choosing where tasks run

TASKS_MODE=rq

TASKS_MODE	The task runs	Use it for
sync	In the caller, before enqueue returns, retries included	Tests. The testing configuration uses it
thread	In a pool of threads of the web process. The default	Development, or a product without a worker
rq	In an RQ queue on Redis, drained by a worker	Production, and any task that reads untrusted input

In thread mode, waiting work lives in memory and is lost when the process stops. A thread cannot be stopped at its timeout either, so a hung task keeps its thread. The run is marked overdue and logged. Only an RQ worker stops an attempt at its timeout.

The app reads the mode once, when it starts. An unknown mode, or rq without the rq package or without a Redis, stops the app right there. In rq mode the readiness probe checks that Redis too, as tasks.

Running a worker

python -m splent_framework.tasks.worker           # drain the queue
python -m splent_framework.tasks.worker --burst   # stop when the queue is empty
python -m splent_framework.tasks.worker --check   # exit 0 while the worker beats

rq mode needs splent_framework[rq] in the product’s image and a container that runs the worker. It builds the app of the product SPLENT_APP names, with every feature, and runs RQ with its scheduler, which requeues each retry when its wait is over. Each job runs in a process of its own and is killed when it outlives its timeout.

Use --check as the container’s healthcheck. It builds no app and reads a heartbeat file the worker writes, TASKS_WORKER_HEARTBEAT_FILE, by default splent-tasks-worker.heartbeat in the temporary directory.

Testing tasks

The testing configuration runs every task in the caller. A test that enqueues sees the effect when enqueue returns, and a task that waits for the commit runs at the test’s commit. To test thread mode, build an app whose config sets TASKS_MODE, and call wait() before you check.

---

Advanced

When a run that waited for the commit cannot be queued, because Redis is down, it cannot undo the commit. It is logged and recorded as failed. A run that starts at once raises instead, so you know the work was not queued.

In rq mode the progress reaches Redis at most twice a second, so a task may report after every file. The last report always arrives. A note longer than 200 characters is cut.

---

Reference

@task

Argument	Default	Means
timeout	TASKS_DEFAULT_TIMEOUT	Seconds one attempt may take
retries	0	Attempts after the first one fails, at most 10
backoff	10	Seconds before the first retry, doubled after each one, at most an hour

enqueue(target, kwargs, *, key=None, after_commit=None, timeout=None, app=None)

Argument	Means
target	The @task function, or its dotted name. An unregistered dotted name is imported, and a plain function found that way runs with the defaults
kwargs	The arguments, JSON values
key	Merges into a run of the same task and key that has not started
after_commit	None waits only after an uncommitted write. True waits in any open transaction. False never waits
timeout	Replaces the task’s for this run

It returns a TaskRun. wait(timeout=None) blocks until the tasks this process started are over, in thread mode, and answers False when the timeout passed first.

TaskRun

Field	Means
status	waiting, queued (also between attempts), running, succeeded, failed, timed_out or dropped
attempts, max_attempts	The attempts made, and the most it gets
enqueued_at, started_at, finished_at	In UTC
error	The last error’s type and message
overdue	The last attempt took, or is taking, longer than its timeout
progress_done, progress_total, progress_note, progress	What the task reported. progress goes from 0 to 1

In sync and thread mode the process keeps the last 500 runs. In rq mode the job in Redis is the record, kept for TASKS_RUN_TTL seconds after it ends.

Settings

Setting	Default	Means
TASKS_MODE	thread, sync in testing	sync, thread or rq
TASKS_THREAD_WORKERS	2	Tasks that run at once in thread mode
TASKS_DEFAULT_TIMEOUT	180	Seconds an attempt may take when its task names none
TASKS_REDIS_URL	REDIS_TASKS_URL, then REDIS_URL	The Redis of the queue. Keep it apart from a cache, which may be flushed
TASKS_QUEUE	The product’s name	The queue, so two products can share a Redis
TASKS_RUN_TTL	86400	Seconds Redis keeps an ended run

Set them in your product’s config.py. A variable of the same name in the .env wins, since whether a worker exists is the deployment’s call.

---

See also

• Application initialization, where the mode is checked
• Health probes, where rq mode adds its check
• Configuration, every variable the framework reads