Metadata-Version: 2.4
Name: continuum-task-server-sdk
Version: 0.0.9
Summary: Python SDK for the Continuum Task Server
Project-URL: Homepage, https://github.com/ContinuumWorkflow/continuum-task-server-sdk-python
Project-URL: Issues, https://github.com/ContinuumWorkflow/continuum-task-server-sdk-python/issues
Author: Continuum
License: MIT
Keywords: continuum,queue,sdk,task,worker
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Requires-Dist: httpx<1.0,>=0.27
Requires-Dist: pydantic<3.0,>=2.6
Provides-Extra: dev
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: respx>=0.21; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Description-Content-Type: text/markdown

# Continuum Task Server SDK for Python

Python client for the [Continuum](https://github.com/ContinuumWorkflow) task server. Designed so a 20-line script can stand up a worker that claims queue items, runs your code, and reports results.

- **`TaskServer`** — decorator-based worker loop: claim, heartbeat **while the handler runs**, status updates, backoff, graceful shutdown. Long-running claims (`auto_complete=False`) need **your** heartbeat loop (documented below).
- **`ContinuumClient`** — thin pythonic client over the management + queue REST APIs (task types, task items, versions, queue, content store).

Requires Python 3.10+.

## Installation

Tagged releases (`v*`) are published to [PyPI](https://pypi.org/project/continuum-task-server-sdk/):

```bash
pip install continuum-task-server-sdk
```

Dev and PR builds are still attached to private GitHub Releases. Install those with a token:

```bash
pip install \
  "https://${GITHUB_TOKEN}@github.com/ContinuumWorkflow/continuum-task-server-sdk-python/releases/download/v0.1.0/continuum_task_server_sdk-0.1.0-py3-none-any.whl"
```

`GITHUB_TOKEN` must be a PAT with repo read access.

## Quickstart: build a task server in 20 lines

```python
import os
from continuum_task_server import TaskServer

server = TaskServer(
    base_url=os.environ["CONTINUUM_URL"],
    api_key=os.environ["CONTINUUM_API_KEY"],
)

@server.task("echo")
def echo(item):
    return {"echoed": item.input_data_json}

@server.task("greet")
def greet(item):
    name = (item.input_data_json or {}).get("name", "world")
    return {"message": f"hello, {name}"}

if __name__ == "__main__":
    server.run()
```

Run it. The server polls `/api/queue/claim` for `echo` and `greet`, claims items as they become available, runs your handler in a thread, heartbeats while it runs, and:

- **Handler returns** a value → queue item marked `ENDED`, return value JSON-encoded as `outputData` (default `auto_complete=True`).
- **Handler raises** → queue item marked `KILLED`, error info written to `outputData`.

Press `Ctrl+C` (or send `SIGTERM`) and the server stops polling and waits up to `shutdown_timeout` seconds for in-flight handlers to finish.

### Deferred completion (`auto_complete=False`)

The handler runs under the normal **in-handler** heartbeat (same as any task). When it returns, the server **does not** send `ENDED` and **does not** keep heartbeating: you own the claim until you finish or lose it to timeouts.

Typical pattern: persist `item.id` (and anything else you need), then on each pass of **your** poller (cron, loop, worker restart), call **`server.client.queue.heartbeat(queue_item_id)`** so the Continuum claim stays alive, and when the real-world condition is met call **`server.complete_queue_item(...)`** or **`server.fail_queue_item(...)`**. Use the **same worker API key** as the process that claimed the item (often the same `TaskServer` / `ContinuumClient` config loaded from env).

```python
@server.task("wait-for-mail", auto_complete=False)
def wait_for_mail(item):
    db.insert_outstanding(queue_item_id=str(item.id), payload=item.input_data_json)
    # Returns without ENDED — no background heartbeat from TaskServer

# Elsewhere: each time you poll your DB for outstanding work (including after restart):
for row in db.outstanding_rows():
    server.client.queue.heartbeat(row.queue_item_id)
    if mail_arrived(row):
        server.complete_queue_item(row.queue_item_id, output_data={"received": True})
```

Standalone process (no `TaskServer`): build a `ContinuumClient` with the worker key and call `client.queue.heartbeat` / `client.queue.update_status` the same way.

### TaskServer options

```python
TaskServer(
    base_url="http://localhost:8080",
    api_key="...",
    max_workers=4,            # thread pool size across all tasks
    poll_interval=1.0,        # initial poll delay (backs off when idle)
    max_poll_interval=5.0,    # max idle poll delay
    heartbeat_interval=15.0,  # how often to call /heartbeat per running task
    shutdown_timeout=30.0,    # how long to wait for handlers during shutdown
)
```

Per-task concurrency limit:

```python
@server.task("docker-run", concurrency=2)
def run_docker(item):
    ...
```

Handler signature: `def handler(item: QueueItem) -> dict | list | str | None`. Inside, you have:

- `item.input_data` — raw JSON string from the queue item (or `None`).
- `item.input_data_json` — parsed value (`dict` / `list` / `str` / `None`).
- `server.client` — full `ContinuumClient` if you need to chain management calls, fetch content, enqueue child tasks, etc.

## Using `ContinuumClient` directly

```python
from continuum_task_server import ContinuumClient, TaskStatus

with ContinuumClient(base_url="http://localhost:8080", api_key="...") as client:
    # Task types
    types = client.task_types.list()
    echo_type = client.task_types.get_by_name("echo")

    # Task items + versions
    item = client.task_items.get_by_name("my-task")
    versions = client.task_items.versions.list(item.id)

    # Enqueue work
    queued = client.queue.add(task_name="echo", input_data={"hello": "world"})

    # Worker-side primitives (normally handled by TaskServer)
    claimed = client.queue.claim("echo")
    if claimed is not None:
        client.queue.heartbeat(claimed.id)
        client.queue.update_status(claimed.id, TaskStatus.ENDED, output_data={"ok": True})

    # Content store
    content = client.content_store.get_by_url("db://...")
    if content is not None:
        print(content.as_text())
```

`input_data` / `output_data` accept `dict` / `list` / `str` / `None`; non-string values are JSON-encoded for you.

### Errors

All API failures raise `ContinuumError` or a specific subclass: `BadRequestError`, `UnauthorizedError`, `ForbiddenError`, `NotFoundError`, `ConflictError`, `ServerError`. Each carries `status_code` and `body`.

```python
from continuum_task_server import ContinuumClient, NotFoundError

with ContinuumClient(...) as client:
    try:
        client.task_types.get_by_name("does-not-exist")
    except NotFoundError as e:
        print(e.status_code, e.body)
```

## Endpoints covered

| Group | Method | Path |
| --- | --- | --- |
| Task Types | `GET`/`POST` | `/api/management/task-types[/{id}\|/by-name]` |
| Task Items | `GET`/`POST` | `/api/management/task-items[/{id}\|/by-name\|/{id}/publish]` |
| Task Item Versions | `GET`/`POST`/`PATCH` | `/api/management/task-items/{id}/versions[...]` |
| Queue (management) | `GET`/`POST` | `/api/management/queue-items[/{id}]` |
| Queue (worker) | `POST` | `/api/queue/claim`, `/api/queue/queue-items/{id}/heartbeat`, `/api/queue/queue-items/{id}/status` |
| Queue content | `GET` | `/api/queue/queue-items/{id}/content` |
| Content store | `GET` | `/api/management/content-store[/{id}\|?url=db://...]` |

All requests send `Api-Key: <your-key>`. `204 No Content` responses are normalized to `None` (e.g. `client.queue.claim()` returns `None` when nothing's available).

## Development

```bash
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

ruff check .
ruff format --check .
pytest
```

Tests use [`respx`](https://lundberg.github.io/respx/) to mock the httpx transport — no live server required.

## Versioning

- **Tag `v1.2.3`** → release wheel `1.2.3` attached to a `v1.2.3` GitHub Release.
- **Push to `main`** → prerelease wheel `0.1.0.dev{run}+{shortsha}` attached to a `dev-{shortsha}` Release.
- **Pull request** → prerelease wheel `0.1.0b{pr}.{run}` attached to a `pr-{pr}` Release; install command posted as a PR comment.

## License

MIT.
