localqueue

A persistent, multiprocess-safe local queue for Python, backed by SQLite and Rust.

localqueue gives scripts and workers durable jobs with ACK/NACK, leases, bounded retries, and dead-letter handling—without a server, daemon, or external service.

• Durable: jobs survive process restarts in a local SQLite database.
• Safe under concurrency: multiple Python threads and processes can share a queue on the same machine.
• Failure-aware: expired leases and transient failures are retried; exhausted jobs move to a dead-letter state.
• Fenced: every delivery has a unique receipt, so stale workers cannot acknowledge a newer delivery.
• Flexible: use multiple named queues, optional job deduplication, automatic worker heartbeats, and custom serializers.
• Event-driven: optional durable pub/sub with explicit static topology, atomic fan-out, consumer groups, retries, and dead-letter handling.

Installation · Quick start · Worker · Event bus · Guarantees · API · Changelog · Development

Installation

Using uv:

uv add localqueue

Using pip:

python -m pip install localqueue

localqueue requires Python 3.10 or newer.

Upgrading from 0.5.0 requires code and storage changes because 1.x is a backward-incompatible reimplementation. See Migrating from 0.5.0.

Quick start

Save the two labeled blocks as producer.py and worker.py, then run the producer before starting the worker.

# producer.py
from localqueue import SimpleQueue


with SimpleQueue("./data", lease_seconds=30, max_retries=3) as queue:
    queue.put(
        {"task": "send-email", "to": "hello@example.com"},
        job_id="welcome-email-42",
    )

The producer can stop here. put() commits the job to ./data/localqueue.db; it is not an in-memory dictionary.

# worker.py — run this later, in another process
from localqueue import SimpleQueue

with SimpleQueue("./data", lease_seconds=30, max_retries=3) as queue:
    job = queue.get()

    try:
        print(f"Sending email to {job.data['to']}")
    except Exception as error:
        queue.nack(job, last_error=str(error))
    else:
        queue.ack(job)

The path passed to SimpleQueue is a directory. The queue creates and manages localqueue.db inside it. Payloads are JSON-serialized by default. Jobs survive normal process restarts; use fsync=True when the stronger SQLite durability setting documented in Delivery guarantees is required.

Worker

Worker handles the ACK/NACK lifecycle for you. A successful handler is acknowledged, an unexpected exception is retried, and an exception listed in permanent_errors is sent directly to the dead-letter state.

from localqueue import SimpleQueue, Worker


class InvalidDeployment(Exception):
    pass


def deploy(job):
    print(f"Deploying {job.data['app']}@{job.data['revision']}")


with SimpleQueue("./data", lease_seconds=30, max_retries=3) as queue:
    worker = Worker(
        queue,
        deploy,
        permanent_errors=(InvalidDeployment,),
        heartbeat_interval=10,
    )
    worker.run()

For long-running handlers, heartbeat_interval renews the lease in the background. It must be shorter than lease_seconds; one-third of the lease is recommended. A handler can also renew it explicitly with job.extend_lease(seconds), using a positive duration.

Event bus

Optional pub/sub on top of the same durable queues (requires the bus extra):

uv add "localqueue[bus]"

# shared.py
from localqueue.bus import BaseEvent, BusTopology


class UserCreated(BaseEvent):
    event_name = "user.created"

    user_id: str


TOPOLOGY = BusTopology({"email": [UserCreated]})

# producer.py
from localqueue.bus import EventBus

from shared import TOPOLOGY, UserCreated


bus = EventBus("./data", name="app", topology=TOPOLOGY)
bus.dispatch(UserCreated(user_id="123"))  # atomic fan-out, committed
bus.close()

# email_worker.py
import asyncio

from localqueue.bus import EventBus

from shared import TOPOLOGY, UserCreated


bus = EventBus("./data", name="app", topology=TOPOLOGY)
email = bus.subscription("email")


@email.handler(UserCreated)
async def send_welcome(event: UserCreated) -> None:
    ...


asyncio.run(bus.run())  # consume subscriptions handled by this process

Each subscription is a durable queue (__bus__:{bus}:{subscription}), so workers in multiple processes act as consumer groups. Handlers get the same retry, lease, and dead-letter semantics as regular jobs. The static topology decides where events are persisted; local handlers decide what the current process consumes. Producers do not import handlers, and workers do not participate in dispatch. See docs/event-bus.md.

Delivery guarantees

put() ──> ready ──> leased ──> acked
                     │
                     ├── nack() or expired lease ──> ready
                     │
                     └── fail() or retry limit ────> failed
                                                        │
                                           retry_failed() ──> ready

By default, SQLite uses synchronous=NORMAL, which protects against normal process crashes but may lose recent transactions after an operating-system or power failure. Pass fsync=True to use synchronous=FULL for stronger durability.

See Delivery guarantees for the complete durability, lease, retry, fencing, and deduplication contract.

Configuration

queue = SimpleQueue(
    "./data",
    name="emails",
    lease_seconds=60,
    max_retries=3,
    fsync=False,
    serializer=None,
)

API overview

Architecture

The public API is a small Python facade over a native PyO3 extension. Rust owns the transactional state machine, while SQLite WAL mode provides local persistence and coordinates competing writers.

Python application
       │
       ▼
SimpleQueue / Worker       Python API and serialization
       │
       ▼
Rust native extension      leases, receipts, retries, transactions
       │
       ▼
localqueue.db              SQLite WAL, one machine

All queue state lives in one SQLite table. Claims and multi-step transitions use immediate transactions so concurrent processes cannot reserve the same delivery. This is local infrastructure for a shared database on one machine, not a distributed message broker.

Development

Build the extension in a local virtual environment and run the test suite:

uv sync --extra dev
uv run maturin develop
uv run pytest

Rust quality checks:

cargo fmt --all --check
cargo clippy --locked --all-targets --all-features -- -D warnings

Build a release wheel:

uv run maturin build --release --locked

License

Licensed under the Apache License 2.0.