liteworks 0.1.0

SQLite-backed task queue (stdlib only, no dependencies)

liteworks
=========

SQLite-backed task queue for Python 3.12+. Zero dependencies; uses Python's
stdlib HTTP server. Each queue is one SQLite file. Workers on multiple machines
connect over HTTP. No external service required.


ARCHITECTURE
------------

Uses a visibility timeout. The atomic claim is one SQL statement:

    UPDATE jobs SET claimed_by = ?, claimed_at = ?
    WHERE id = (
        SELECT id FROM jobs
        WHERE (claimed_by IS NULL OR claimed_at < ?) AND scheduled_at <= ?
        ORDER BY scheduled_at LIMIT 1
    ) RETURNING *

Ack uses a fencing token so a timed-out worker cannot clobber a fresh claim:

    DELETE FROM jobs WHERE id = ? AND claimed_by = ? RETURNING *

Ack + DLQ write are atomic (one SQL transaction via POST /{queue}/fail).
Snooze clears the claim and advances scheduled_at.
Retry schedules a future re-delivery by advancing scheduled_at and
incrementing delivery_count.

One server, multiple queues: each queue maps to {data_dir}/{queue}.db and
is created lazily on first use.

One handler per message type; fanout belongs in the handler.


INSTALL
-------

    pip install liteworks


QUICKSTART
----------

Start the server:

    liteworks serve --data-dir /var/liteworks --port 8000

In your application:

    from dataclasses import dataclass
    from liteworks import Liteworks, RetryPolicy, Snooze, Discard

    lw = Liteworks(base_url='http://localhost:8000', queue='emails')

    @dataclass
    class SendEmail:
        to: str
        subject: str

    @lw.handler
    def send_email(msg: SendEmail):
        # raise Snooze(60) to defer and retry in 60s (delivery_count unchanged)
        # raise Discard('bad_address') to move straight to DLQ
        pass

    lw.publish(SendEmail(to='[e-mail]', subject='Hello'))

    w = lw.worker()
    w.install_signal_handlers()
    w.run()


MULTIPLE QUEUES
---------------

One server hosts all queues. Each queue name maps to its own SQLite file.

    emails  = Liteworks(base_url='http://localhost:8000', queue='emails')
    reports = Liteworks(base_url='http://localhost:8000', queue='reports')

    emails.publish(SendEmail(...))
    reports.publish(GenerateReport(...))

Queue names: [a-zA-Z0-9_-], 1-64 characters.


DEMO
----

See demo.py for a runnable two-queue example with workers, stats, and a
browser dashboard. Run:

    python demo.py


STATS AND UI
------------

GET /{queue}/stats returns JSON:

    {
        "queue": "emails",
        "ready": 3,
        "claimed": 1,
        "scheduled": 0,
        "failed": 2,
        "completed": 47,
        "recent_failures": [
            {"id": "...", "reason": "SMTP timeout", "failed_at": 1720000000.0}
        ],
        "recent_runs": [
            {"id": "...", "worker_id": "w1", "completed_at": 1720000001.0}
        ]
    }

    lw.stats()   # via Transport; returns same dict

GET / returns a top-level dashboard (self-contained HTML, auto-refresh) showing
all queues and an enqueue form. Open http://localhost:8000/ in a browser.

GET /queues returns a JSON list of queue names.
GET /metrics returns Prometheus-format text metrics for all queues.


AUTH
----

Protect with a shared secret:

    liteworks serve --data-dir /var/liteworks --secret $LITEWORKS_SECRET

    lw = Liteworks(base_url='...', queue='emails', secret='...')

GET / is always accessible (browser loads the page before it can send auth).
GET /queues, GET /metrics, GET /{queue}/stats require Basic auth when secret set.
POST, DELETE, PATCH accept either HMAC-SHA256 x-signature (API clients) or
Basic auth (browser). Basic auth password is the secret; username is ignored.

Completed job retention (default: keep last 1000 per queue; 0 = disabled):

    liteworks serve --data-dir /var/liteworks --completed-retention 500

    lw = Liteworks(..., completed_retention=500)


RETRY POLICY
------------

Default: 3 retries, exponential backoff, 5s base, 300s cap, jitter on.

    from liteworks import RetryPolicy

    @lw.handler(policy=RetryPolicy(max_retries=5, base_delay=2.0, jitter=False))
    def handle(msg: MyMsg): ...


PUBLISH OPTIONS
---------------

    lw.publish(msg, msg_id='my-uuid')               # idempotency key
    lw.publish(msg, parent_id='parent-job-id')      # lineage
    lw.publish(msg, scheduled_at=time.time()+3600)  # deferred delivery


SCHEMA
------

Created automatically on first use of each queue:

    CREATE TABLE IF NOT EXISTS jobs (
        id TEXT PRIMARY KEY,
        payload TEXT NOT NULL,
        msg_id TEXT,
        parent_id TEXT,
        delivery_count INTEGER DEFAULT 0,
        scheduled_at REAL NOT NULL,
        claimed_by TEXT,
        claimed_at REAL,
        created_at REAL NOT NULL
    );

    CREATE TABLE IF NOT EXISTS failed_jobs (
        id TEXT PRIMARY KEY,
        payload TEXT NOT NULL,
        msg_id TEXT,
        parent_id TEXT,
        reason TEXT,
        failed_at REAL NOT NULL
    );

    CREATE TABLE IF NOT EXISTS completed_jobs (
        id TEXT PRIMARY KEY,
        payload TEXT NOT NULL,
        msg_id TEXT,
        parent_id TEXT,
        delivery_count INTEGER DEFAULT 0,
        worker_id TEXT,
        completed_at REAL NOT NULL
    );

msg_id has a partial unique index (WHERE msg_id IS NOT NULL) enforcing
idempotent publish when the caller passes an explicit msg_id.
completed_jobs retains the last N entries per queue (trimmed on each ack),
where N is the completed_retention setting (default 1000; 0 disables tracking).


DURABILITY
----------

Use Litestream to replicate each SQLite file to S3:

    litestream replicate /var/liteworks/emails.db s3://bucket/emails.db


PUBLIC API
----------

    serve(data_dir, host='', port=8000, secret=None)   module-level, blocking

    Liteworks(data_dir=None, queue='default',
              base_url='http://localhost:8000',
              visibility_timeout=30.0, secret=None)
    Liteworks.handler(fn=None, *, policy=None)     decorator / decorator factory
    Liteworks.publish(msg, *, msg_id=None, parent_id=None, scheduled_at=None)
    Liteworks.stats()                              -> dict
    Liteworks.worker(*, poll_interval=0.5, backoff_interval=2.0, max_empty=4)
    Liteworks.serve(port=8000, host='')            blocking; requires data_dir

    Worker.run()
    Worker.stop()
    Worker.install_signal_handlers()

    Snooze(seconds)    raise from handler to defer without consuming a retry
    Discard(reason)    raise from handler to skip retries and go to DLQ

    RetryPolicy(max_retries=3, backoff='exponential', base_delay=5.0,
                max_delay=300.0, jitter=True)