errand

Stateful background jobs — the missing middle ground between FastAPI's BackgroundTasks and Celery. A zero-dependency engine with an optional, first-class FastAPI adapter.

BackgroundTasks is fire-and-forget: you can't tell whether a task started, is running, finished, or failed. The next step up is Celery/ARQ + Redis — a broker, a worker runtime, and a pile of ops. errand fills the valley in between: in-process jobs with tracked state, retries, scheduling, and dependency injection inside tasks — using nothing but the Python standard library.

The engine imports zero third-party packages. FastAPI is an optional extra: install it and you get a drop-in lifespan and a status router; skip it and the engine still runs anywhere. Because FastAPI is user-supplied and only its most stable public surface is touched (APIRouter, the .dependency attribute on Depends), a FastAPI release won't leave errand stranded.

Status: 0.2.0, published. Job store, worker pool, status router, retries with backoff, dependency injection, scheduling, lifecycle hooks, and bounded in-memory growth are all implemented, tested (100% coverage), and live on PyPI — see CHANGELOG.md.

The PyPI name errand turned out to be taken by an unrelated, actively maintained package, so this project publishes as errand-jobs (pip install errand-jobs) while the import name stays errand_jobs — the GitHub repo and project name remain errand.

Why

• State you can query. Every job has an id and a status (PENDING → RUNNING → SUCCEEDED / FAILED), with timestamps, result, and error captured. An optional router exposes it over HTTP out of the box.
• No new infrastructure. Pure asyncio. Runs inside your app process (your FastAPI app, or any async program). Start with the in-memory store; swap in a durable store later without touching your task code.
• Retries with backoff. Fixed or exponential, configured per task.
• Scheduling built in. interval, at, and a cron subset — no separate beat process.
• Real dependency injection. Use the same Depends(...) callables you use in routes, including yield-based resources with proper teardown.
• Sync tasks don't block the loop. Plain def tasks run in a thread.

Non-goals

Not a distributed task queue. If you need multi-machine workers, guaranteed delivery across a broker, or millions of jobs, use Celery/ARQ. errand targets the single-process, "I just need to know if it worked" case that most apps actually have.

Install

pip install errand-jobs            # engine only, zero dependencies
pip install "errand-jobs[fastapi]" # + the lifespan and status router
# or
uv add "errand-jobs[fastapi]"

Requires Python 3.10+. FastAPI is only needed for the adapter (the .router); the engine runs standalone. The distribution is errand-jobs; the import name is errand_jobs.

Quickstart

from fastapi import FastAPI
from errand_jobs import Errand

tasks = Errand()                       # in-memory store, 4 workers
app = FastAPI(lifespan=tasks.lifespan)   # starts/drains workers + scheduler
app.include_router(tasks.router, prefix="/jobs")  # optional status API


@tasks.task(max_retries=3, retry_backoff="exponential")
async def send_welcome_email(user_id: int) -> None:
    ...  # slow work


@app.post("/signup")
async def signup() -> dict:
    job = tasks.enqueue(send_welcome_email, user_id=42)
    return {"job_id": job.id}

Check on it:

GET /jobs/{job_id}
→ {"id": "...", "name": "send_welcome_email", "status": "RUNNING",
   "attempts": 1, "created_at": "...", "started_at": "...", ...}

Dependency injection in tasks

The same pattern you use in routes, teardown included:

from fastapi import Depends
from errand_jobs import Errand

tasks = Errand()


async def get_db():
    db = Session()
    try:
        yield db
    finally:
        db.close()


@tasks.task
async def reindex(db=Depends(get_db)) -> None:
    ...  # db is torn down after the task, success or failure

Scheduling

@tasks.schedule(cron="0 * * * *")     # hourly
async def hourly_cleanup() -> None:
    ...


@tasks.schedule(interval_seconds=30)  # every 30s
async def heartbeat() -> None:
    ...

Scheduled runs are tracked exactly like enqueued jobs.

Lifecycle hooks

Get visibility into job outcomes without wrapping every task:

@tasks.on_success
def log_success(job: Job) -> None:
    print(f"{job.name} succeeded: {job.result_repr}")


@tasks.on_failure
def alert_on_failure(job: Job) -> None:
    print(f"{job.name} failed: {job.error}")


@tasks.on_retry
def log_retry(job: Job) -> None:
    print(f"{job.name} retrying (attempt {job.attempts})")

Each decorator can be used multiple times; every registered hook fires, in registration order, with an immutable snapshot of the job as of that exact transition. Hooks may be sync or async — keep them fast, they run inline on the event loop, not in a thread. A hook that raises is logged and doesn't affect the job or any other hook.

Using errand with sync frameworks (Flask, Django)

FastAPI gets first-class treatment: pass tasks.lifespan to FastAPI(...) and the worker pool starts and drains with the app, no glue code needed. Flask and Django are WSGI-based and don't own an event loop the way an ASGI app does, so errand's asyncio engine needs a small bridge — run one event loop in a background thread for the app's lifetime, and hop onto it from each view with asyncio.run_coroutine_threadsafe(...):

import asyncio
import threading

from errand_jobs import Errand

tasks = Errand()


@tasks.task
def resize_image(path: str) -> str:
    ...  # slow work


_loop = asyncio.new_event_loop()
threading.Thread(target=_loop.run_forever, daemon=True).start()
asyncio.run_coroutine_threadsafe(tasks.startup(), _loop).result()

enqueue() schedules an internal task on the running loop, so it must be called from a coroutine running on _loop — wrap it rather than calling tasks.enqueue(...) straight from the view:

# Flask
@app.post("/upload")
def upload():
    async def _enqueue():
        return tasks.enqueue(resize_image, request.form["path"])

    job = asyncio.run_coroutine_threadsafe(_enqueue(), _loop).result()
    return {"job_id": job.id}

# Django
def upload_view(request):
    async def _enqueue():
        return tasks.enqueue(resize_image, request.POST["path"])

    job = asyncio.run_coroutine_threadsafe(_enqueue(), _loop).result()
    return JsonResponse({"job_id": job.id})

get_job()/list_jobs() are already coroutines, so the same call works directly with no wrapper:

job = asyncio.run_coroutine_threadsafe(tasks.get_job(job_id), _loop).result()

On process exit (e.g. via atexit), drain in-flight jobs the same way:

asyncio.run_coroutine_threadsafe(tasks.shutdown(), _loop).result()

Backends

The core ships with InMemoryJobStore. The JobStore interface is the single seam for durability — a Redis or Postgres store can be added later as an optional extra without changing any task code.

InMemoryJobStore keeps every job record until the process exits or something prunes it — unbounded growth in a long-running process. For that case, pass prune_after (seconds) to Errand(...) and it prunes terminal jobs (SUCCEEDED/FAILED/CANCELLED) automatically once their finished_at is older than that, on a background check (at most every 60s). Jobs still PENDING/RUNNING are never touched, however old:

tasks = Errand(prune_after=86400)  # drop finished jobs after 24h

Off by default — a short-lived process, or one backed by a durable store, usually doesn't need it.

Roadmap

All of TASKS.md's milestones (M0–M7) are implemented, tested, and merged — this is a complete 0.1.0, not a work in progress. TASKS.md is kept as the historical build plan; contributor and architecture notes live in DESIGN.md and CLAUDE.md; release notes are in CHANGELOG.md.

Explicitly deferred past 0.1.0 (called out as such in DESIGN.md): a durable JobStore (Redis/Postgres — the interface is already the seam for it), remote enqueue/cancel over HTTP, and jitter on retry backoff.

License

MIT