Metadata-Version: 2.4
Name: localqueue
Version: 1.4.1
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Dist: pydantic>=2,<3 ; extra == 'benchmark'
Requires-Dist: persist-queue==1.1.0 ; extra == 'benchmark-compare'
Requires-Dist: pydantic>=2,<3 ; extra == 'bus'
Requires-Dist: typing-extensions>=4.12,<5 ; extra == 'bus'
Requires-Dist: hypothesis>=6,<7 ; extra == 'dev'
Requires-Dist: import-linter>=2,<3 ; extra == 'dev'
Requires-Dist: pytest>=8,<10 ; extra == 'dev'
Requires-Dist: pytest-cov>=7,<8 ; extra == 'dev'
Requires-Dist: pytest-timeout>=2,<3 ; extra == 'dev'
Requires-Dist: maturin>=1.14 ; extra == 'dev'
Requires-Dist: pydantic>=2,<3 ; extra == 'dev'
Requires-Dist: packaging>=24,<26 ; extra == 'dev'
Requires-Dist: mypy>=1.18,<2 ; extra == 'dev'
Requires-Dist: pyright>=1.1,<2 ; extra == 'dev'
Requires-Dist: ruff>=0.15,<1 ; extra == 'dev'
Requires-Dist: pyrefly>=1.1,<2 ; extra == 'dev'
Requires-Dist: pre-commit>=4,<5 ; extra == 'dev'
Requires-Dist: python-semantic-release>=10.6,<11 ; extra == 'dev'
Requires-Dist: zensical>=0.0.51,<0.1 ; extra == 'docs'
Provides-Extra: benchmark
Provides-Extra: benchmark-compare
Provides-Extra: bus
Provides-Extra: dev
Provides-Extra: docs
License-File: LICENSE
Summary: A persistent, multiprocess-safe local queue for Python, backed by SQLite and Rust.
License-Expression: Apache-2.0
Requires-Python: >=3.11
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Homepage, https://github.com/brunoportis/localqueue
Project-URL: Issues, https://github.com/brunoportis/localqueue/issues
Project-URL: Repository, https://github.com/brunoportis/localqueue

# localqueue

[![PyPI](https://img.shields.io/pypi/v/localqueue.svg)](https://pypi.org/project/localqueue/)
[![Python](https://img.shields.io/pypi/pyversions/localqueue.svg?cacheSeconds=300)](https://pypi.org/project/localqueue/)
[![CI](https://github.com/brunoportis/localqueue/actions/workflows/ci.yml/badge.svg)](https://github.com/brunoportis/localqueue/actions/workflows/ci.yml)
[![codecov](https://codecov.io/gh/brunoportis/localqueue/graph/badge.svg)](https://app.codecov.io/gh/brunoportis/localqueue)
[![License](https://img.shields.io/badge/license-Apache--2.0-blue.svg)](https://github.com/brunoportis/localqueue/blob/main/LICENSE)

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](#installation) · [Quick start](#quick-start) ·
[Worker](#worker) · [Static typing](#static-typing) · [Dead letters](#dead-letters) · [Event bus](#event-bus) ·
[Benchmarks](#benchmarks) ·
[Guarantees](#delivery-guarantees) ·
[Backpressure](#bounded-backlog-and-backpressure) ·
[Diagnostics](#runtime-diagnostics) · [API](#api-overview) ·
[Storage compatibility](docs/storage-compatibility.md) ·
[Migrating to 1.3](docs/migrating-to-1.3.md) ·
[Operational envelope](docs/operational-envelope.md) ·
[Changelog](CHANGELOG.md) ·
[Development](#development)

## Installation

Using [uv](https://docs.astral.sh/uv/):

```bash
uv add localqueue
```

Using pip:

```bash
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](CHANGELOG.md#migrating-from-050).

## Quick start

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

```python
# 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.

```python
# 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](#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.

```python
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](docs/typing.md) 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](docs/dead-letters.md) for at-least-once replay safety.

## Event bus

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

```bash
uv add "localqueue[bus]"
```

```python
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](https://github.com/brunoportis/localqueue/blob/main/docs/event-bus.md).

## Delivery guarantees

> [!IMPORTANT]
> `localqueue` provides **at-least-once** delivery, not exactly-once delivery.
> A worker can finish an external side effect and crash before `ack()`, causing
> the job to be delivered again. Make handlers idempotent when duplicate effects
> matter.

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

| Behavior | Guarantee |
| --- | --- |
| Successful `put()` | The job was committed to SQLite and has an internal ID. |
| Worker crash | An unacknowledged job becomes available after its lease expires. |
| Stale worker | `ack()`, `nack()`, `fail()`, and lease extension require the current receipt. |
| Retries | `max_retries` allows that many retries after the initial delivery; exhaustion moves the job to `failed`. |
| Deduplication | A `job_id` is unique within a named queue while its record exists. |
| Ordering | Ready jobs are claimed by insertion ID, but completion order is best effort under concurrency. |

`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](https://github.com/brunoportis/localqueue/blob/main/docs/guarantees.md)
for the complete durability, lease, retry, fencing, and deduplication contract.

## Configuration

```python
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,
)
```

| Option | Default | Description |
| --- | ---: | --- |
| `path` | required | Directory containing the shared `localqueue.db` file. |
| `name` | `"default"` | Logical queue name; several queues can share one database. |
| `delivery` | `DeliveryPolicy()` | Immutable lease duration and retry policy. |
| `durability` | `DurabilityMode.RELAXED` | Throughput-oriented `RELAXED` or stronger `DURABLE` synchronization intent. |
| `serializer` | JSON | Object implementing `dumps(obj) -> bytes` and `loads(bytes) -> obj`. |
| `max_pending_jobs` | `None` | Optional positive limit for ready + processing jobs in this logical queue. |

`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:

```python
# 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:

```python
# 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:

```python
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](https://github.com/brunoportis/localqueue/blob/main/docs/backpressure.md)
for polling, timeout, transition, multiprocess, and error details.

## API overview

| Method | Purpose |
| --- | --- |
| `put(data, job_id=None, *, block=True, timeout=None)` | Enqueue with optional deduplication and capacity waiting. |
| `put_many(items, *, block=True, timeout=None)` | Atomically enqueue a complete batch, optionally using `EnqueueItem` for per-item deduplication. |
| `get(block=True, timeout=None)` | Claim a job and start its lease. |
| `get_nowait()` | Claim immediately or raise `Empty`. |
| `ack(job)` | Confirm successful processing. |
| `nack(job, delay=0, last_error=None)` | Return a transient failure to the queue, optionally with a non-negative delay. |
| `fail(job, last_error=None)` | Move a permanent failure to the dead-letter state. |
| `extend_lease(job, seconds)` | Renew the current delivery lease by a positive duration. |
| `reclaim_expired_leases()` | Reclaim expired leases explicitly; `get()` also does this automatically. |
| `stats()` | Return `ready`, `processing`, `acked`, and `failed` counts. |
| `diagnostics()` | Return a typed, immutable, read-only operational snapshot. |
| `check_integrity(*, mode="full", max_errors=100)` | Run a bounded, typed full or quick SQLite integrity check. |
| `backup(destination_directory)` | Create a verified online backup in a new exclusive directory. |
| `list_failed(limit=100, offset=0)` | Inspect dead-letter jobs. |
| `retry_failed(message_id)` | Non-blockingly move a dead-letter job back to `ready`, or raise `Full`. |
| `purge(older_than, include_failed=False)` | Delete old terminal records. |
| `vacuum()` | Compact the shared SQLite database. Run during a maintenance window. |

## Runtime diagnostics

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

```python
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](https://github.com/brunoportis/localqueue/blob/main/docs/diagnostics.md)
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:

```python
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](https://github.com/brunoportis/localqueue/blob/main/docs/maintenance.md)
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](https://pyo3.rs/) extension. Rust owns the transactional state machine,
while SQLite WAL mode provides local persistence and coordinates competing
writers.

```text
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:

```bash
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](docs/benchmarks.md) 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:

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

Run the configured strict type check for the production package:

```bash
uv run pyrefly check --progress-bar no
```

Rust quality checks:

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

Build a release wheel:

```bash
uv run maturin build --release --locked
```

## License

Licensed under the [Apache License 2.0](https://github.com/brunoportis/localqueue/blob/main/LICENSE).

