ERRAND

MINIMAL DURABLE TASK QUEUE FOR PYTHON

> uv add git+https://github.com/RverseHQ/errand

THE DISPATCH DESK

errand.db
from errand import Job, Queue

queue = Queue("errand.db")


@queue.handler("summarize", concurrency=2)
async def summarize(job: Job) -> str:
    return await llm(job.payload["text"])


@queue.handler("greet")
async def greet(job: Job) -> str:
    return f"hello {job.payload['name']}"

A SHORT LIST OF GUARANTEES

SUBMIT AND READ BACK

import asyncio

from errand import Job, Queue, Status

queue = Queue("errand.db")


@queue.handler("greet")
async def greet(job: Job) -> str:
    return f"hello {job.payload['name']}"


async def main() -> None:
    async with queue:
        id = queue.submit("greet", {"name": "errand"})
        await asyncio.sleep(0)

        job = queue.get(id)

        assert job.status is Status.DONE
        assert job.result == "hello errand"


asyncio.run(main())

FAILURE IS A RESULT

@queue.handler("divide")
async def divide(job: Job) -> float:
    return job.payload["a"] / job.payload["b"]


async with queue:
    id = queue.submit("divide", {"a": 1, "b": 0})
    await asyncio.sleep(0)

    job = queue.get(id)

    assert job.status is Status.FAILED
    assert job.error == "ZeroDivisionError: division by zero"

RECOVERY

async def crash() -> str:
    queue = Queue("errand.db")

    @queue.handler("work")
    async def work(job: Job) -> None:
        await asyncio.sleep(3600)

    async with queue:
        id = queue.submit("work", None)
        await asyncio.sleep(0)

        assert queue.get(id).status is Status.RUNNING

    return id


async def restart(id: str) -> None:
    queue = Queue("errand.db")

    @queue.handler("work")
    async def work(job: Job) -> str:
        return "finished"

    async with queue:
        await asyncio.sleep(0)

        assert queue.get(id).result == "finished"


asyncio.run(restart(asyncio.run(crash())))

COMPOSING PRIMITIVES

def submit_once(payload: dict) -> str:
    for job in queue.running():
        if job.kind == "summarize" and job.payload == payload:
            return job.id
    return queue.submit("summarize", payload)

ONE SMALL SURFACE

API Purpose
Queue(path)
@queue.handler(kind, concurrency=None)
queue.submit(kind, payload)
queue.get(id)
queue.running()
async with queue
Job
Status

WEB SERVICES

@asynccontextmanager
async def lifespan(app: FastAPI):
    async with queue:
        yield


app = FastAPI(lifespan=lifespan)


@app.post("/tasks")
async def create(text: str) -> str:
    return queue.submit("summarize", {"text": text})


@app.get("/tasks/{id}")
async def read(id: str) -> Job:
    job = queue.get(id)
    if job is None:
        raise HTTPException(404)
    return job

ROADMAP