Metadata-Version: 2.5
Name: cairnq
Version: 0.10.0
Summary: SQLite-first, cross-language, storage-centered durable task runtime
Project-URL: Homepage, https://github.com/Jannchie/cairnq
Project-URL: Repository, https://github.com/Jannchie/cairnq
Project-URL: Issues, https://github.com/Jannchie/cairnq/issues
Author-email: Jannchie <jannchie@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: aiosqlite,cross-language,durable,job-queue,sqlite,task-queue,task-runtime,worker
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.10
Requires-Dist: aiosqlite>=0.19
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Provides-Extra: postgres
Requires-Dist: asyncpg>=0.29; extra == 'postgres'
Description-Content-Type: text/markdown

# cairnq (Python)

SQLite-first, cross-language, storage-centered durable task runtime. The Python
SDK. API and worker processes coordinate only through a shared SQLite file.

```python
from cairnq import CairnQ, Worker

# Worker side — a handler always receives (ctx, payload).
worker = Worker.sqlite("tasks.db")

@worker.task                          # registered under the function name, "create_summary"
async def create_summary(ctx, payload):
    await ctx.progress(0.2, "reading")
    return {"summary": await llm.summarize(payload["text"])}

worker.serve()                        # blocking entry point; Ctrl-C closes cleanly

# API side (in your server) — submit returns immediately.
tasks = CairnQ.sqlite("tasks.db")
task = await tasks.submit("create_summary", {"text": text}, key=f"summary:{aid}")
```

`@worker.task` defaults the task name to the function's name. Pass a string for a
dotted/namespaced name: `@worker.task("summary.create")`.

Synchronous call (submit + wait):

```python
from cairnq import TaskFailed, TaskTimeout

try:
    result = await tasks.call("create_summary", {"text": text}, wait_timeout_ms=10_000)
except TaskFailed as e:
    log(e.code, e.message, e.retryable)   # envelope fields, no e.error["code"] digging
except TaskTimeout as e:
    # The task keeps running — resume the wait instead of submitting again.
    result = await tasks.wait(e.task_id, timeout_ms=60_000)
    # …or tasks.wait_by_key(key), from a process that never held the id.
```

Inspect a task by id/key without memorizing status strings:

```python
task = await tasks.get_by_key(key)
if task and task.succeeded:        # also .failed / .canceled / .running / .queued / .is_terminal
    use(task.result)
```

Optionally define a task once and share the symbol across both ends — the name
lives in one place (no string drift), and `call()` is typed as the task's result:

```python
from cairnq import TaskDef

summarize = TaskDef[dict, dict]("summarize")

@worker.task(summarize)            # registered under summarize.name
async def handle(ctx, payload): ...

result = await tasks.call(summarize, {"text": text})
```

Opt-in: every API still accepts a plain name string (cross-language callers use it).

## Running it in production

```python
worker = Worker.sqlite(
    "tasks.db",
    concurrency=4,            # handler calls at once; use max_in_flight_bytes to bound memory
    retry_backoff_ms=1_000,   # window doubles per attempt, capped at retry_backoff_max_ms (30s),
                              # jittered over its upper half; 0 disables
    on_error=lambda exc, info: log.warning("worker survived %s: %s", info, exc),
)

# Nothing else deletes rows, so give the client a retention policy — it sweeps
# terminal tasks in bounded batches for as long as the handle is open. A
# per-status mapping keeps each status on its own clock (statuses left out are
# never swept): spent results go in minutes, failures stay for diagnosis.
tasks = CairnQ.sqlite(
    "tasks.db",
    retention=Retention(older_than_ms={"succeeded": 300_000, "failed": 7 * 24 * 3600_000}),
)
```

A **sync** handler (`def`, not `async def`) is dispatched to a thread, so the
usual shape around a blocking GPU or HTTP call keeps the worker's event loop —
and with it every lease this worker holds — alive:

```python
@worker.task("score")
def score(ctx, payload):
    return {"score": model.forward(payload["image"])}  # blocking, off the loop
```

A handler that does real side effects should bail out when it loses its lease —
the task is already running on another worker and nothing it writes is recorded:

```python
@worker.task("long.job")
async def long_job(ctx, payload):
    for chunk in chunks:
        if ctx.lost_lease or await ctx.canceled():
            return
        await process(chunk)
```

## Multi-host

Same code, Postgres instead of the file — `CairnQ.postgres(dsn)` /
`Worker.postgres(dsn)`. Install with `pip install cairnq[postgres]`.

`schema` puts cairnq's tables in a schema of their own:
`CairnQ.postgres(dsn, schema="cairnq")` creates it if absent and sets
`search_path` on every connection. **Every process in a deployment must agree on
it** — a queue whose API and worker resolve to different schemas is two empty
queues, and both sides come up healthy. cairnq refuses to connect where it can
see that about to happen; the TypeScript SDK takes the same option and applies
the same rule.

### Sharing the application's connection

Given a `PgExecutor` instead of a DSN, cairnq runs inside a session the
application already has — no second driver, no second pool:

```python
from cairnq import CairnQ, PgExecutor

executor: PgExecutor = ...   # ~40 lines over your driver
tasks = CairnQ.postgres(executor)
```

An adapter passes rows through as its driver produced them; cairnq normalizes
what the drivers disagree about. An executor cairnq was handed is never closed
by cairnq.

That shared session is also what lets a task's settlement commit together with
the rows the task produced:

```python
@worker.task("render.document")
async def render_document(ctx, payload):
    rendered = await render(payload)

    async def write(session):
        await session.query("insert into pages (doc, n) values ($1, $2)", [...])
        return {"pages": len(rendered)}   # becomes the task's result

    return await ctx.succeed_in(write)
```

Without it the two are separate transactions, and a crash between them leaves
work durable while the task still reads as running — on retry, recomputed. If
the lease turns out to be gone, the settlement matches no row and the caller's
writes roll back with it.

## Watching

`watch` calls back when the tasks on a queue may have changed — for a dashboard
that would otherwise poll:

```python
stop = tasks.watch(on_signal, queues=["render"])
```

It is notify-accelerated polling, not an event log. On Postgres an idle watch
costs nothing and a signal lands within milliseconds; where LISTEN is
unavailable — a transaction-mode pooler, or SQLite, which has no channel — the
timer alone still delivers `poll` signals. Treat a signal as "re-read now"; the
truth is in `stats()` / `list()` / `get()`.

The protocol (schema + canonical SQL) lives in `../cairnq-protocol` and is shared
verbatim with the TypeScript SDK. See `../cairnq-protocol/PROTOCOL.md`.
