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 · Benchmarks · Guarantees · Backpressure · Diagnostics · API · Storage compatibility · Operational envelope · 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,
    max_pending_jobs=None,
)

Bounded backlog and backpressure

Limit one logical queue by job count and choose whether producers wait:

from localqueue import Full, SimpleQueue


queue = SimpleQueue("./data", max_pending_jobs=10_000)
try:
    queue.put(payload, block=False)
except Full:
    ...

queue.put(payload, timeout=5.0)

Pending means ready + processing: delayed jobs, active leases, and expired unreclaimed leases count; ACKed and failed jobs do not. The native capacity check, deduplication calculation, and complete insert share one BEGIN IMMEDIATE transaction, so participating processes using the same configured limit cannot oversubscribe it. put_many() is all or nothing, and duplicates that do not create rows consume no new slots, even above the limit. A batch whose deduplicated new-row count exceeds the limit raises Full immediately because it can never fit.

This limits logical backlog, not SQLite/WAL bytes, disk space, or retained terminal records. The setting belongs to each SimpleQueue object and is not persisted: all producers for a logical queue must use the same limit, while an unlimited or direct-SQL client can bypass the contract. EventBus fanout remains unlimited. See the backpressure reference for polling, timeout, transition, multiprocess, and error details.

API overview

Runtime diagnostics

Capture a typed, immutable operational snapshot without opening SQLite directly:

import json

from localqueue import SimpleQueue


queue = SimpleQueue("./data")
report = queue.diagnostics()

print(report)
print(json.dumps(report.to_dict(), indent=2))
queue.close()

Diagnostics schema v2 reports logical counts, configured capacity, pending jobs, available slots, and ages for the selected queue together with effective SQLite settings and best-effort shared database/WAL/SHM sizes. It is read-only and does not reclaim leases, checkpoint WAL, or run an integrity check. See the runtime diagnostics reference for field types, snapshot semantics, clock limitations, and the boundary with integrity/backup maintenance.

Integrity checks and online backups

Run explicit SQLite maintenance without reaching into the native connection:

from localqueue import SimpleQueue


queue = SimpleQueue("./data")
integrity = queue.check_integrity()
if integrity.ok:
    backup = queue.backup("./backups/2026-07-22")
    print(backup.to_dict())
queue.close()

Backups use SQLite's Online Backup API and remain consistent while other connections produce and consume jobs. The destination directory is reserved exclusively and must not already exist. See the maintenance reference for result fields, locking and latency, filesystem requirements, error semantics, independent verification, and recovery steps.

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.

Benchmarks

The canonical single-process benchmark is run with the optional benchmark extra:

uv sync --extra benchmark
uv run python -m localqueue.benchmark --profile smoke --output smoke.json
uv run python -m localqueue.benchmark --profile multiprocess-ci --output multiprocess.json

See the benchmark guide for NORMAL/FULL durability and report interpretation. Benchmark output is evidence about a particular host, filesystem, and run; it is not a product performance promise.

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

Run the configured strict type check for the production package:

uv run pyrefly check --progress-bar no

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.