Metadata-Version: 2.4
Name: fronta
Version: 0.3.0
Summary: Distributed task queue for asyncio tasks and sandboxed processes
Keywords: task queue,postgresql,asyncio,sandbox,bubblewrap,mcp,workers
Author: Ondrej
Author-email: Ondrej <ondrej@ilcik.com>
License-Expression: MIT
License-File: LICENSE
License-File: THIRD_PARTY_NOTICES.md
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Framework :: FastAPI
Classifier: Framework :: Pydantic :: 2
Classifier: Intended Audience :: Developers
Classifier: Operating System :: MacOS :: MacOS X
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Dist: click>=8.1.1,<9
Requires-Dist: psycopg-pool>=3.2.8,<4
Requires-Dist: psycopg[binary,pool]>=3.2.11,<4
Requires-Dist: pydantic>=2.12,<3
Requires-Dist: pydantic-settings>=2.5,<3
Requires-Dist: fastapi>=0.110,<1 ; extra == 'server'
Requires-Dist: jsonschema[format-nongpl]>=4.20,<5 ; extra == 'server'
Requires-Dist: mcp>=2.1,<3 ; extra == 'server'
Requires-Dist: uvicorn>=0.31.1,<1 ; extra == 'server'
Requires-Python: >=3.12
Project-URL: Homepage, https://github.com/dreo/fronta
Project-URL: Repository, https://github.com/dreo/fronta
Project-URL: Issues, https://github.com/dreo/fronta/issues
Project-URL: Changelog, https://github.com/dreo/fronta/blob/main/CHANGELOG.md
Provides-Extra: server
Description-Content-Type: text/markdown

# Fronta

Task queue on PostgreSQL for Python. Workers run `async def` handlers in-process and executables
in bubblewrap sandboxes; an optional server exposes the queue over REST and MCP with a small
dashboard.

Fronta keeps its tables in a `fronta` schema of your own PostgreSQL 16+ database; there is no
broker. Workers claim rows with `SELECT … FOR UPDATE SKIP LOCKED`, hold leases renewed by
heartbeats, and record every state change in one fenced transaction, so the tasks of a crashed or
stalled worker are reaped and retried while attempts remain. Priorities, scheduled runs, dedupe
keys, retries with jittered backoff and concurrency limits (per task type and per key) are enforced
in the database; attempt timeouts and cancellation by the worker.

**Status:** alpha. The API and the schema can change between minor versions before 1.0 (see
`CHANGELOG.md`). Linux or macOS, Python 3.12–3.14, PostgreSQL 16+. Sandboxed process workers
require Linux.

## Install

```bash
uv add fronta               # SDK + worker
uv add "fronta[server]"     # + REST/MCP server and dashboard
```

`pip install fronta` works the same. The SDK, server, and workers containing only asyncio tasks are
supported on Linux and macOS. A worker containing any process task needs Linux, `bwrap`
(bubblewrap), `prlimit` (util-linux), and unprivileged user namespaces.

## Example

```python
# app/tasks.py
import fronta
from pydantic import BaseModel


class Resize(BaseModel):
    image_id: int
    width: int


@fronta.task("resize", input=Resize, max_attempts=5, attempt_timeout=120)
async def resize(ctx: fronta.Context, job: Resize) -> dict[str, int]:
    await ctx.progress({"stage": "download"})
    ...  # idempotent work that honors CancelledError
    return {"bytes": 12345}


worker = fronta.Worker([resize])
```

```bash
export FRONTA_DSN=postgresql://user:pass@host/db   # the role must be able to create the schema
fronta db init                    # creates schema `fronta`; safe to repeat
fronta worker app.tasks:worker    # runs until SIGTERM/SIGINT
```

```python
# enqueue.py: any process that reaches the database
import asyncio

import fronta
from app.tasks import Resize, resize


async def main() -> None:
    await fronta.open_pool()  # once, at application start
    try:
        task_id = await resize.enqueue(Resize(image_id=7, width=800), priority=5, key="resize-7")
        print(task_id)
    finally:
        await fronta.close_pool()  # at application shutdown


asyncio.run(main())
```

`enqueue(..., conn=conn)` joins a non-autocommit psycopg transaction instead of using the pool;
with an autocommit connection, Fronta wraps the row and its notifications in one transaction.
`key` dedupes: while a task with the same key is queued or running, `enqueue` returns its id;
once that task has finished, the same key enqueues a new one.

A handler gets the validated input and a `Context` (`task_id`, `attempt`, `log`, `progress()`,
`enqueue()`, `cancelled`, `state` from the worker lifespan). It must handle
`asyncio.CancelledError` and be safe to run twice: after a lost lease the task runs again.

## Lifecycle events

External workflow code can react to committed task state transitions through the SDK:

```python
import fronta
from app.tasks import Notify, notify


async def run_workflow() -> None:
    async with fronta.subscribe_events() as events:
        async for event in events:
            if event.type != "resize" or event.state is not fronta.State.SUCCEEDED:
                continue
            source = await fronta.get_task(event.id)
            if source.input["image_id"] != 7:
                continue
            await notify.enqueue(
                Notify(source_id=event.id, result=source.result),
                key=f"workflow:{event.id}:notify",
            )
```

Each `TaskEvent` has `id`, `type`, and `state`. Events cover committed state transitions, including
enqueue, claim, retry, success, failure, release, cancellation, and reaping. Heartbeats, progress,
a running task's cancellation request, rolled-back enqueues, and dedupe hits do not broadcast a
state transition. `get_task()` returns the latest durable row, which may already be newer than the
event being handled.

The stream is a live PostgreSQL notification feed: it has no replay, acknowledgement, or silent
reconnect. A disconnect is raised to the consumer, which should reconnect and reconcile from its
own durable state. Running one workflow instance prevents competing consumers, but cannot prevent
a missed event during startup or a crash. An enqueue `key` only dedupes while the child is queued
or running; a durable workflow should record handled source events in its own table and enqueue the
child with the same psycopg transaction via `enqueue(..., conn=conn)`.

On Linux, a sandboxed process task with a placeholder executable:

```python
class Convert(BaseModel):
    source: str


convert = fronta.process_task(
    "convert",
    ["/usr/bin/convert-tool", "--from-stdin"],  # reads the JSON input on stdin
    input=Convert,
    sandbox=fronta.Sandbox(memory_bytes=512 << 20, cpu_time_s=60, max_pids=16),
    max_concurrency=4,
)
```

The process runs in a private tmpfs `/work` without network; its result is
`{"exit_code", "stdout", "stderr", "truncated"}`. Exit code 0 means the task succeeded; anything
else fails the attempt.

## Server

```bash
FRONTA_SERVER_TOKEN=... fronta server      # 127.0.0.1:8000
```

REST under `/api/v1` (task types, enqueue, get, list, cancel), MCP at `/mcp`, dashboard at `/`.
The Python SDK enqueues, reads one task, and subscribes to lifecycle events; listing and
cancellation go through the server. `FRONTA_SERVER_TOKEN` is required (the server never runs open:
even on loopback a browser could be made to cancel or enqueue tasks); every REST and MCP request
sends it as `Authorization: Bearer <token>`. Put a TLS-terminating reverse proxy in front of it
outside a private network. Endpoints, inputs and error codes:
[docs/reference.md](https://github.com/dreo/fronta/blob/main/docs/reference.md#server).

## Deploy

One database, any number of workers, optionally a server. Each process reads `FRONTA_*`
environment variables; `FRONTA_DSN` is the only required one. Run workers under a supervisor that
restarts them: a worker exits 0 after a graceful stop and 70 when a handler ignores cancellation or
blocks the event loop.

```ini
# /etc/systemd/system/fronta-worker.service
[Unit]
Description=Fronta worker
After=network-online.target

[Service]
User=app
WorkingDirectory=/srv/app
# FRONTA_DSN=... and other FRONTA_* variables; readable by root only (mode 0600)
EnvironmentFile=/etc/fronta/worker.env
ExecStart=/srv/app/.venv/bin/fronta worker app.tasks:worker
# SIGTERM goes to the worker only, which stops its sandboxes itself; SIGKILL to everything
KillMode=mixed
Restart=always
RestartSec=2
TimeoutStopSec=120

[Install]
WantedBy=multi-user.target
```

`systemctl enable --now fronta-worker`. On `SIGTERM` the worker stops claiming, lets running
attempts finish for `FRONTA_GRACE_S`, then stops the rest (another grace period for cooperative
cancellation, then a kill) and records every outcome before it exits. That takes at most
`2 × FRONTA_GRACE_S + 6 × FRONTA_KILL_TIMEOUT_S` (91 s with the defaults 30 s and 5 s) unless the
database is unreachable or a sandbox cannot be killed: then the worker keeps trying rather than
lose an outcome, and systemd's `SIGKILL` at `TimeoutStopSec` ends it. That loses no data:
sandboxes die with the worker and unrecorded attempts are retried when their lease expires.

Throughput is bounded by the database's commit rate (at least two durable commits per task, plus
heartbeats and progress): about 150 no-op tasks/s in total on a laptop PostgreSQL with fsync; a
claim costs ~5 ms on a 70k-row queue. Configuration, retry policy, guarantees and the
measurements: [docs/reference.md](https://github.com/dreo/fronta/blob/main/docs/reference.md).

## Not covered

- Exactly-once side effects: a worker stalled past its lease may still be running while the task
  is retried elsewhere; the stale attempt's writes to Fronta are rejected, its other effects are not.
- A built-in workflow engine, chains, or periodic tasks (only `run_at`). A singleton external
  consumer can implement workflows from lifecycle events.
- Schema migrations before 1.0: a release that changes the schema needs `fronta db init` on a
  fresh schema.
- Windows.
- Sandboxed process tasks on macOS. A worker containing one fails its startup check with a clear
  platform error; run that worker on Linux.

## Development

```bash
uv sync --all-extras
docker run -d --name fronta-test-pg -e POSTGRES_USER=fronta -e POSTGRES_PASSWORD=fronta \
  -e POSTGRES_DB=fronta -p 127.0.0.1:5439:5432 postgres:16
export FRONTA_TEST_DSN=postgresql://fronta:fronta@127.0.0.1:5439/fronta
make check       # lint, format, types, architecture, deps (also the git pre-commit hook)
make checkall    # check + the full test suite + pip-audit
```

CI runs the full `make checkall` gate once on every pull request and push to `main`; compatibility
legs cover Python 3.12–3.14, lower dependency bounds, real Linux process sandboxes, and the
portable SDK, asyncio worker, and server on macOS without repeating the stress/browser tiers. To
release: set the version (`uv version X.Y.Z`), add the
CHANGELOG section, merge, then push the tag `vX.Y.Z` from that `main` commit; the gate runs again,
the package goes to PyPI and a GitHub release is created. `SPEC.md` is the contract.

## License

MIT. The dashboard bundles Alpine.js (MIT); see `THIRD_PARTY_NOTICES.md`.
