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.1, 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 and run as detached background tasks — a slow or hanging hook can never hold a worker slot or delay the next job. 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.

One thing that's specific to InMemoryJobStore, not a universal guarantee: enqueue() returns a job that's already visible via get_job()/list_jobs() with no polling needed, because it can create the record synchronously. A store that needs real I/O to persist (a future Redis/Postgres store) can't offer that without blocking the event loop, so it falls back to creating the record asynchronously — a moment of eventual consistency right after enqueue() for that store, same as before this was fixed for the in-memory case.

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

The core feature set (job store, worker pool, status router, retries, dependency injection, scheduling, lifecycle hooks, bounded memory) is implemented, tested, and released — see CHANGELOG.md for what shipped in each version.

Not yet built: 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