Python

Install

pip install honker

The wheel bundles the native extension on supported platforms, so normal Python apps do not need a separate libhonker_ext file.

Open

import honker

db = honker.open("app.db")

The default is watcher_backend="polling", the stable backend. "kernel" (filesystem wake hints) and "shm" (the SQLite WAL-index fast path) are experimental opt-ins that raise unless the package was built with the matching Cargo features — published wheels are polling-only, so build from source with maturin develop --release --features kernel-watcher,shm-fast-path to use them.

Queue

q = db.queue("emails")
q.enqueue({"to": "alice@example.com", "order_id": 42})

job = q.claim_one("worker-1")
if job:
    send_email(job.payload)
    job.ack()

Function Tasks

Python provides the highest-level task convenience API: @q.task and @q.periodic_task decorators, a process-global handler registry, in-process workers, and a python -m honker worker CLI.

@q.task(name="send-email", retries=3)
def send_email(to):
    return {"sent": True, "to": to}

result = send_email("alice@example.com")
# In another process: python -m honker worker myapp.tasks:db --queue=emails
print(result.get(timeout=10))

See Function tasks for worker startup, results, stable names, periodic tasks, and the Java/Kotlin registry equivalent.

Transactional Outbox

Use the outbox when the durable database commit should trigger an external side effect such as email, webhooks, cache invalidation, or search indexing.

outbox = db.outbox("email", lambda payload: send_email(payload))

with db.transaction() as tx:
    tx.execute("INSERT INTO orders (id, total) VALUES (?, ?)", (42, 9900))
    outbox.enqueue({"order_id": 42}, tx=tx)

await outbox.run_worker("email-worker")

The outbox is just a reserved Honker queue named _outbox:<name>, so another Python process can deliver jobs after the writer exits.

ORM Section

For SQLAlchemy, SQLModel, or Django, keep the ORM transaction as the owner of your business write. Load the Honker extension on the same SQLite connection, then call honker_enqueue, notify, or stream SQL functions inside that transaction.

from sqlalchemy import text

with Session(engine) as session, session.begin():
    session.execute(text("INSERT INTO orders (id, total) VALUES (:id, :total)"),
                    {"id": 42, "total": 9900})
    session.execute(
        text("SELECT honker_enqueue(:queue, :payload, NULL, NULL, 0, 3, NULL)"),
        {"queue": "emails", "payload": '{"order_id":42}'},
    )

See the full Python ORM recipe for extension loading, typed payloads, and Django connection hooks.

Testing

Use tmp_path / "test.db" rather than :memory:. Cross-process workers and watchers need file locking, WAL, and reopen semantics that in-memory SQLite cannot model.