Metadata-Version: 2.5
Name: errand-jobs
Version: 0.2.1
Summary: Stateful background jobs — a zero-dependency engine with an optional FastAPI adapter.
Project-URL: Homepage, https://github.com/jmiguelmangas/errand
Project-URL: Repository, https://github.com/jmiguelmangas/errand
Project-URL: Issues, https://github.com/jmiguelmangas/errand/issues
Project-URL: Changelog, https://github.com/jmiguelmangas/errand/blob/main/CHANGELOG.md
Author-email: jmiguelmangas <jmmangas@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: asyncio,background-jobs,background-tasks,fastapi,jobs,scheduler,task-queue
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: coverage[toml]>=7; extra == 'dev'
Requires-Dist: fastapi>=0.100; extra == 'dev'
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-cov>=5; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Provides-Extra: fastapi
Requires-Dist: fastapi>=0.100; extra == 'fastapi'
Description-Content-Type: text/markdown

# errand

<p align="center">
  <img src="https://raw.githubusercontent.com/jmiguelmangas/errand/main/assets/logo.png" alt="errand logo" width="200">
</p>

<p align="center">
  <a href="https://pypi.org/project/errand-jobs/"><img src="https://img.shields.io/pypi/v/errand-jobs.svg" alt="PyPI version"></a>
  <a href="https://pypi.org/project/errand-jobs/"><img src="https://img.shields.io/pypi/pyversions/errand-jobs.svg" alt="Supported Python versions"></a>
  <a href="https://github.com/jmiguelmangas/errand/actions/workflows/ci.yml"><img src="https://github.com/jmiguelmangas/errand/actions/workflows/ci.yml/badge.svg" alt="CI status"></a>
  <a href="https://github.com/jmiguelmangas/errand/blob/main/LICENSE"><img src="https://img.shields.io/pypi/l/errand-jobs.svg" alt="License"></a>
</p>

> 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`](https://github.com/jmiguelmangas/errand/blob/main/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

```bash
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

```python
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:

```python
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

```python
@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:

```python
@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(...)`:

```python
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:

```python
# 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}
```

```python
# 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:

```python
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:

```python
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:

```python
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`](https://github.com/jmiguelmangas/errand/blob/main/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
