Jobs¶
enqueue gives you back a snapshot of the row it just wrote, not a live handle, so the Job you are
holding keeps saying queued: the worker that runs your task is a different process writing to the
database, and nothing reaches back into your object. Ask for the current state when you want it:
job = generate_pdf.enqueue(123)
print(job.id) # 01a061bf-c08a-738d-...
job.refresh() # updates this instance in place, and returns it
job.status # queued -> running -> success; held by a limit, waiting on another job
job.attempts # 3, if it took three tries
job.last_error # 'ConnectionError: refused', when it failed
app.get(job.id) # or a fresh object, if you prefer not to mutate
Attribute access never does I/O on its own. refresh() is the one place a round trip happens, which
keeps for job in jobs: print(job.status) from quietly becoming a thousand queries.
From async code, which is the native path:
job = await generate_pdf.aenqueue(123)
Results¶
If the task returns something you want back, ask for it to be kept:
@app.task(store_result=True)
async def generate_pdf(document_id: int) -> str:
return f"/tmp/{document_id}.pdf"
job = generate_pdf.enqueue(123)
...
job.refresh().result() # '/tmp/123.pdf'
Results are off by default because most jobs are run for their side effects, and storing return
values nobody reads is write amplification on the hot path. A stored result lives as long as its job
unless result_ttl says otherwise, and max_result_bytes keeps a big one out of the queue:
app = TinyJobs("sqlite:///jobs.db", max_result_bytes=64 * 1024)
@app.task(store_result=True, result_ttl=3600)
async def generate_pdf(document_id: int) -> str: ...
Within a few minutes of the TTL a worker that registers the task drops the result and keeps the
job, so jobs show still tells what happened. A result over the cap fails the job with
ResultTooLarge and is not retried, since the function ran and would produce the same result
again; put the file in object storage and return its key.
Waiting for a job¶
If you want to block until a job is finished, wait polls for you:
job = generate_pdf.enqueue(123)
print(job.wait(timeout=30).result()) # '/tmp/123.pdf'
await job.awaited(timeout=30) # from async code
It returns the job whatever the outcome, rather than raising, so a failed job leaves you holding
job.status, job.attempts and job.last_error to look at. Only the timeout raises, as
WaitTimeout, which is also a TimeoutError if you would rather catch that.
The async twin is called awaited and not await for the boring reason that await is a keyword.
Waiting on a job from the request that created it means you did not need a background job. It is meant for scripts, tests and glue code.
One job after another¶
A job can be told to run only once another has succeeded:
report = build_report.enqueue(42)
email_report.with_options(after=report).enqueue(42)
The second job is stored as waiting, where no worker can take it. When the first succeeds the
store queues the second as it records the success. When the first fails for good or is cancelled,
the store cancels the second and every job waiting behind it, and their last_error says so. On
Postgres a job sent inside a transaction that was still open when the first one ended misses that
step, as a few in a thousand do when both happen in the same instant, and the next reaper pass,
--reap-interval apart, settles it instead. An attempt that failed and will be retried changes nothing: the jobs behind keep waiting
for the last one.
after takes the Job or its id, works on app.send and as tinyjobs send --after too, and
combines with the other options. A
delay still counts from when the job was sent, so a job whose time has passed runs as soon as the
one before it is done. Any number of jobs can wait on the same job, and each can have jobs waiting
on it, so a line or a tree is a few enqueue calls. A job waits on one job only: there is no
"after all of these", and nothing to declare up front.
The job before it has to be in the store already, or enqueue raises
tinyjobs.JobNotFoundError. If it has succeeded by then, the new job is queued at once, and if it
has failed or been cancelled the new job is stored as cancelled, so a job sent after a fast one is
never lost to the race. The reaper pass that settles the Postgres case above is a worker's: where
only app.drain() runs, such a job waits until a worker does.
The second job gets nothing from the first. Send both what they need, or send the first one's id and read its result:
@app.task(store_result=True)
def build_report(account_id: int) -> str: ...
@app.task
def email_report(report_job: str) -> None:
path = app.get(report_job).result()
report = build_report.enqueue(42)
email_report.with_options(after=report).enqueue(report.id)
jobs retry puts a job back by the state of the job before it: queued if that one succeeded or is
no longer in the store, waiting if it is in flight again, and refused while it is still failed or
cancelled. So after a failure, retry the job that failed and then the ones behind it; retrying the
first does not bring the others back by itself. jobs cancel on any job of the line cancels what
waits behind it.
Running a worker¶
tinyjobs worker --app tasks:app --concurrency 8
tinyjobs is installed as a command, but python -m tinyjobs does the same thing and is handy when
you have not activated a virtualenv:
python3.13 -m tinyjobs worker --app tasks:app --concurrency 8
Looking at what happened¶
tinyjobs jobs list --app tasks:app
tinyjobs jobs list --app tasks:app --status failed
tinyjobs jobs show <job-id> --app tasks:app
jobs show prints every attempt, not just the last one:
attempts
1 failed 4ms laptop/8821/9389541d ConnectionError
2 failed 3ms laptop/8821/9389541d ConnectionError
3 success 12ms laptop/8821/9389541d
Timeouts on sync tasks are advisory¶
You can't kill a running Python thread. When a threaded task exceeds its timeout the job is marked
failed and the worker moves on, but the thread keeps going until the function returns, and a warning
names the task. In a worker that thread counts against concurrency until it returns, so the worker
claims one job fewer and never holds a job it has no thread for. A drain or an eager call runs the
next job on a new thread beside it. The process exits without waiting for such threads. The timeout
counts from when the job's thread starts, so a job is never failed before it runs. Pass timeouts to
the library you are calling, write the task as async def where cancellation actually works, or
declare it isolated, where the timeout is hard.
Arguments are serialized as JSON¶
Extended with tags for datetime, date, Decimal, UUID, Path, set, tuple, bytes and
non-string dict keys. Pickle is available behind TinyJobs(..., allow_pickle=True) but it's off by
default, because deserializing a pickle runs arbitrary code and the Postgres and Redis backends reach
over a network. Dataclasses and enums need one line of registration:
app.codec.register_dataclass(Money)
app.codec.register_enum(Priority)
An example¶
There's a runnable example in examples/demo.py covering retries, timeouts, priorities, delays,
deduplication and cron:
python -m examples.demo
tinyjobs worker --app examples.demo:app --concurrency 4