Background jobs and cron for Python, backed by SQLite, with no dependencies.

pip install tinyjobs

That's the whole setup. No broker, no result backend, no daemon. The queue is a file.

Celery is the right tool for a lot of systems, but it asks for a Redis or RabbitMQ instance before it will run a single job. This is for the projects that never needed that: a Django app that sends emails, a scraper with a nightly refresh, a CLI that offloads slow work. When you outgrow SQLite you switch the URL to Redis and your task code doesn't change.

tasks.py

from tinyjobs import TinyJobs

app = TinyJobs("sqlite:///jobs.db")

@app.task async def generate_pdf(document_id: int) -> str: ...

@app.task(max_attempts=5, timeout=300) def transcode(video_id: int) -> None: ...

Enqueue from anywhere, including plain sync code:

from tasks import generate_pdf

job = generate_pdf.enqueue(123) print(job.id) # 01a061bf-c08a-738d-...

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.refresh() # updates this instance in place, and returns it job.status # queued -> running -> success 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.

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.

Terminal jobs are pruned on a schedule. Successes and cancellations are kept for a week, failures for thirty days, because failures are the ones you come back to:

app = TinyJobs("sqlite:///jobs.db", retention=RetentionPolicy(failed=90 * 86400)) app = TinyJobs("sqlite:///jobs.db", retention=KEEP_EVERYTHING) # prune nothing

tinyjobs worker --app tasks:app --retain-success 7d --retain-failed 30d tinyjobs worker --app tasks:app --no-purge

tinyjobs purge --app tasks:app --before 30d --dry-run

Every worker checks whether a sweep is due, and exactly one of them wins, because claiming the sweep is a single conditional UPDATE on a marker row. So running forty workers still means one sweep per interval, with no leader to elect. Queued and running jobs are never touched at any horizon, and each sweep deletes in small batches with a cap, so it never holds the write lock long enough to stall a claim.

Note that pruning does not shrink the file. SQLite reuses the freed pages, so the database plateaus rather than shrinking. Handing space back to the filesystem needs VACUUM, which wants exclusive access, so it stays a manual thing.

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.

Or from async code, which is the native path:

job = await generate_pdf.aenqueue(123)

Run 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

Look 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

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

Cron

@app.task(cron="*/15 * * * *") async def refresh_cache() -> None: ...

@app.task(cron="30 2 * * *", timezone="Europe/Rome") def nightly_report() -> None: ...

There is no extra process to run. Every tinyjobs worker also runs the scheduler, and however many workers you start, each firing creates exactly one job: the schedule moves on with a single conditional UPDATE, and only the worker whose update matched writes the job, in the same transaction. Kill any of them, even with kill -9, and the rest carry on.

What it creates is an ordinary job, with the task's retries, queue and priority.

tinyjobs schedules --app tasks:app

NAME CRON TIMEZONE NEXT RUN LAST RUN LAST JOB

---

tasks.nightly_report 30 2 * * * Europe/Rome 2026-10-11 02:30:00 - - tasks.refresh_cache */15 * * * * UTC 2026-10-10 14:15:00 2026-10-10 14:09:00 01a125b7-6b62-7135-b9a0-b7707bf39c1c

tinyjobs worker --app tasks:app --no-scheduler # runs jobs, fires nothing tinyjobs scheduler --app tasks:app # fires schedules, runs nothing

When nothing was running at the moment a schedule was due, misfire decides what happens on restart:

@app.task(cron="0 * * * *") # skip, the default @app.task(cron="0 * * * *", misfire="run_once") # one catch-up run @app.task(cron="0 0 * * *", misfire="run_all", overlap=True) # every missed day, oldest first

A firing only counts as missed when it is more than a minute late (--misfire-grace). A schedule won't start a run while the previous one is still queued or running, unless overlap=True.

A malformed expression, an unknown timezone, 0 0 30 2 * (which never fires) or a cron task that needs arguments all raise at import, not at 2 a.m.

@app.on_job_failure def alert(job, error, will_retry): if not will_retry: sentry_sdk.capture_exception(error)

@app.on_job_success def timing(job, result, duration): statsd.timing(job.task, duration)

on_job_start(job) completes the set. Hooks run in the worker and can be def or async def. One that raises is logged and ignored, so a broken metrics client never fails a job. A def hook runs on the worker's event loop, so keep it quick.

tinyjobs stats --app tasks:app tinyjobs check --app tasks:app --max-wait 5m # exits 1 on any problem

QUEUE QUEUED RUNNING SUCCESS FAILED CANCELLED OLDEST DUE

---

default 2 2 0 0 0 16s mail 1 0 0 0 0 16s

running past their lease: 2

A job past its lease means the worker holding it stopped reporting, usually because it died. check fails on that, on queued jobs for a task the app doesn't register (a missing --import), on a due job waiting longer than --max-wait, and on a cron schedule more than a minute overdue, which means no worker or scheduler is running. Both are app.stats() and app.check() in code.

from fastapi import FastAPI from tasks import app as jobs, generate_pdf

api = FastAPI(lifespan=jobs.lifespan(worker=True))

@api.post("/documents/{document_id}/pdf") async def render(document_id: int) -> dict: job = await generate_pdf.aenqueue(document_id) return {"job": job.id}

uvicorn main:api is the whole deployment. The lifespan connects when the server starts, runs a worker inside it, and on shutdown lets in-flight jobs finish before closing. Plain def endpoints can call enqueue() too, on SQLite or on Redis.

worker=True takes the options of app.worker(), such as concurrency, queues and shutdown_timeout. Under uvicorn --workers 4 each process runs its own worker on the same queue, and cron still fires once. If you already have a lifespan, nest it:

@asynccontextmanager async def lifespan(api: FastAPI): async with jobs.serve(worker=True): yield

Shutdown waits up to shutdown_timeout (30 s) for running jobs, and a second Ctrl-C does not cut it short. Keep it under your orchestrator's grace period: docker stop allows 10 s, and past it the process is killed and its jobs wait out their leases instead of going straight back to the queue.

Starlette takes the same lifespan. Leave out worker=True and the server only enqueues, with tinyjobs worker --app tasks:app running the jobs elsewhere.

The worker shares the process with your API: an async task that blocks the loop stalls requests too, and CPU-heavy def tasks compete for the GIL. That is the point where a separate worker pays off. Numbers in Measured.

Django 6 ships django.tasks, a standard API for background tasks, without a worker to run them. tinyjobs is that worker. Your tasks stay plain Django code:

settings.py

INSTALLED_APPS = [..., "tinyjobs.contrib.django"]

TASKS = { "default": { "BACKEND": "tinyjobs.contrib.django.TinyJobsBackend", "OPTIONS": {"url": "sqlite:///jobs.db"}, } }

shop/tasks.py

from django.tasks import task

@task def send_receipt(order_id: int) -> None: ...

send_receipt.enqueue(42)

python manage.py tinyjobs worker

manage.py tinyjobs takes every tinyjobs command, such as worker, jobs, stats and check, against the configured backend, with --backend to pick a TASKS alias other than default. OPTIONS takes the arguments of TinyJobs(...): a redis:// URL, "eager": True for tests. Queues, priorities, run_after, async tasks, takes_context and get_result() all work, and the task_enqueued, task_started and task_finished signals fire. A result shows its run_after until the job is first picked up.

Tasks are found by their import path, and only inside your INSTALLED_APPS: a job can never make the worker import code your project does not already trust, and enqueueing a task defined elsewhere raises InvalidTask. Django has no retries, so a failed task stays failed. Enqueue from inside a transaction with transaction.on_commit, or the worker may run before the row it needs is committed.

app = TinyJobs("memory://", eager=True)

job = generate_pdf.enqueue(123) # runs before enqueue returns job.status # success job.result() # '/tmp/123.pdf'

Or keep the queue and run it when you choose, with a clock you control:

from tinyjobs import FrozenClock

clock = FrozenClock(time.time()) app = TinyJobs("memory://", clock=clock)

flaky.enqueue() app.drain() # runs everything due, highest priority first; returns how many ran clock.advance(10) # the retry's backoff app.drain()

memory:// is the SQLite backend on a private in-memory database, so tests run the same claim, dedup and retries as production. Delays and backoff are respected in eager mode too: advance the clock and drain. A failing task is recorded on the job, not raised, as it would be in a worker.

Two things are worth knowing before you hit them:

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. Pass timeouts to the library you are calling, or write the task as async def where cancellation actually works.

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) bu