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.

API stability: localqueue is evolving quickly. Minor releases may include breaking changes to the Python API. Persisted database compatibility is treated separately and documented for each release.

• 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 · Static typing · Dead letters · Event bus · Benchmarks · Guarantees · Backpressure · Diagnostics · API · Storage compatibility · Migrating to 1.3 · Operational envelope · Changelog · Development

Installation

Using uv:

uv add localqueue

Using pip:

python -m pip install localqueue

localqueue requires Python 3.11 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 DeliveryPolicy, SimpleQueue


delivery = DeliveryPolicy(lease_seconds=30, max_retries=3)
with SimpleQueue("./data", delivery=delivery) 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 DeliveryPolicy, SimpleQueue

delivery = DeliveryPolicy(lease_seconds=30, max_retries=3)
with SimpleQueue("./data", delivery=delivery) 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. Select DurabilityMode.DURABLE 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 DeliveryPolicy, SimpleQueue, Worker


class InvalidDeployment(Exception):
    pass


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


with SimpleQueue(
    "./data",
    delivery=DeliveryPolicy(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.

Static typing

Payload types flow through Serializer[PayloadT], SimpleQueue[PayloadT], Job[PayloadT], and Worker[PayloadT]. EventBus class patterns likewise give handlers the concrete event subtype. These are static relationships: the serializer reconstructs queue payloads at runtime, and Pydantic validates EventBus events. JsonSerializer does not reconstruct arbitrary annotated classes.

See the static typing guide for a custom serializer, typed workers, concrete sync/async event handlers, string/wildcard handlers, and installed-wheel verification.

Dead letters

queue.list_failed() returns immutable typed records with stable FailureReason, exact raw_payload bytes, and per-record decode errors. Replay one record with queue.retry_failed(record.id). EventBus exposes the same workflow through bus.subscription("payments").list_failed() and .retry_failed(id) without internal queue names. See the dead-letter guide for at-least-once replay safety.

Event bus

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

uv add "localqueue[bus]"

import asyncio

from localqueue.bus import (
    BaseEvent,
    CsvRow,
    CsvSource,
    EventBus,
    RetryPolicy,
    event,
)


@event(identity=("import_id", "external_id"))
class UserCreationRequested(BaseEvent):
    event_name = "user.creation-requested"

    import_id: str
    external_id: str
    name: str


bus = EventBus("./data")


@bus.handler(UserCreationRequested)
async def create_user(event: UserCreationRequested, ctx) -> None:
    await api.create_user(
        external_id=event.external_id,
        name=event.name,
        idempotency_key=ctx.event_id,
    )


creator = bus.subscription(UserCreationRequested.event_name)
creator.config.concurrency = 50
creator.config.retry = RetryPolicy.exponential(max_attempts=8)


@bus.source(CsvSource("users.csv"), checkpoint="users:2026-07")
def users(row: CsvRow) -> UserCreationRequested:
    return UserCreationRequested(
        import_id="2026-07",
        external_id=row["external_id"],
        name=row["name"],
    )


users.config.batch_size = 5_000
users.config.max_pending = 100_000


async def main() -> None:
    try:
        result = await bus.execute(users)
        result.raise_for_failures()
        print(result.items_committed, result.deliveries_acknowledged)
    finally:
        bus.close()


asyncio.run(main())

execute() resumes after interruption, starts registered local handlers when needed, and also allows worker processes sharing the database to participate. It waits for this execution's root and descendant deliveries—not global bus idleness. Failed deliveries are terminal and reported in ExecutionResult; raise_for_failures() makes failure explicit. External effects remain at-least-once, so use a stable idempotency key. The caller always owns bus.close().

Lower-level dispatch() plus standalone bus.run() producer/worker deployments remain supported. Each subscription is a durable queue, so local worker processes act as a consumer group. See docs/event-bus.md.

Delivery guarantees

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

DurabilityMode.RELAXED is the default and selects SQLite synchronous=NORMAL, prioritizing throughput. Recent transactions may be lost after abrupt operating-system, host, kernel, or power failure. DurabilityMode.DURABLE selects synchronous=FULL and prioritizes stronger protection for recent commits. Neither mode is an absolute guarantee across every filesystem, kernel, drive cache, controller, or hardware failure.

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

Configuration

from localqueue import DeliveryPolicy, DurabilityMode, SimpleQueue


queue = SimpleQueue(
    "./data",
    name="emails",
    delivery=DeliveryPolicy(
        lease_seconds=60,
        max_retries=3,
    ),
    durability=DurabilityMode.RELAXED,
    serializer=None,
    max_pending_jobs=None,
)

require_subscribers remains a boolean on EventBus: it is one stable dispatch decision, not a group of related invariants. serializer, EventBus topology, and event registry remain explicit strategy objects.

Migrating configuration in v1.3

The old Python constructor parameters are removed without aliases or deprecation shims:

# before
SimpleQueue(path, fsync=True)

# after
SimpleQueue(path, durability=DurabilityMode.DURABLE)

# before
SimpleQueue(path, lease_seconds=30, max_retries=5)

# after
SimpleQueue(
    path,
    delivery=DeliveryPolicy(
        lease_seconds=30,
        max_retries=5,
    ),
)

The same delivery= and durability= arguments apply to EventBus. Public configuration attributes move under the corresponding semantic object:

# before                                  # after
queue.lease_seconds                       queue.delivery.lease_seconds
queue.max_retries                         queue.delivery.max_retries
bus.lease_seconds                         bus.delivery.lease_seconds
bus.max_retries                           bus.delivery.max_retries
bus.fsync                                 bus.durability

The removed attributes have no properties, aliases, or compatibility shims. This is a Python API break only; it does not change the SQLite schema or serialized payload format.

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.