Python API

The public labtasker package is synchronous and typed. Every resource operation is available both as a package-level function and as a method with the same name on Client. Methods return Labtasker domain models, not HTTP response wrappers.

Choose the Client form

Package-level functions share one lazily created process-wide Client:

import labtasker

task = labtasker.submit_task({"seed": 7}, routes=["sdxl"])
page = labtasker.list_tasks(status="pending")

Use an explicit Client for a large submission loop, deterministic cleanup, test isolation, or more than one Server:

from labtasker import Client

with Client(url="https://labtasker.example", token=token, queue="paper") as client:
    task = client.submit_task({"seed": 7}, routes=["sdxl"])

Client(url=None, token=None, queue=None) resolves each omitted field through environment, the current directory's .labtasker/config.toml, then defaults. Resolution happens once when the Client is constructed. Later chdir(), config, or environment changes do not retarget it. A non-null queue= on a Task method overrides only that operation's Queue; changing Servers requires a new Client.

close() is idempotent and never stops a local Server. Operations on a closed Client raise RuntimeError("Client is closed."). The package-level default Client has no close/reset API. There is no asynchronous Client.

Client.server_version is a read-only str | None: the normalized PEP 440 Server package version from the latest business response. It is None before a response or when that response has no usable version header. Reading it never makes a request. Older Servers may not advertise their version.

When a response reports a Server older than the Client, the Client writes an advisory warning to stderr recommending an upgrade. Each Client instance warns once per distinct older Server version, including patch and prerelease differences. Results, exceptions, and retries are unchanged; the warning does not mean the current operation is incompatible. There is no version preflight or automatic fallback. Applications needing feature-specific compatibility checks must account for server_version being unknown and still handle operation errors.

Task operations

submit_task(args=None, *, name=None, metadata=None, priority=0,
            max_attempts=3, routes=None, task_id=None, queue=None) -> Task
get_task(task_id, *, queue=None) -> Task
list_tasks(*, status=None, name=None, name_fuzzy=None, filter=None, order_by="created_at",
           descending=True, limit=100, cursor=None, queue=None) -> TaskPage
count_tasks(*, status=None, name=None, name_fuzzy=None, filter=None,
            group_by=None, limit=None, cursor=None, queue=None) -> int | GroupCountPage
update_task(task_id, changes, *, queue=None) -> Task
update_tasks(*, filter, changes, queue=None) -> BulkUpdateResult
cancel_task(task_id, *, queue=None) -> Task
requeue_task(task_id, *, queue=None) -> Task
delete_task(task_id, *, queue=None) -> None

Grouped counts

Use an ordered sequence of dimensions, not a comma-separated Python string:

page = client.count_tasks(status="pending", group_by=["routes", "status"])
for group in page.items:
    print(group.key, group.count)

Task dimensions are routes and status, singly or together in either order. GroupCountPage contains group_by, count, items: list[CountGroup], and next_cursor. Each CountGroup has key: dict[str, str] and count. The top-level count covers the complete selection. Multi-route Tasks appear in each compatible route group, so summing groups can exceed that total.

limit defaults to 100 groups and accepts 1–1000; cursor continues one page. Both require grouping. Groups sort by their keys in the requested dimension order. Follow cursors with the same selection and grouping; page size may change. Each page reads current data, so concurrent changes can affect later pages. Empty, repeated, whitespace-containing and unsupported dimensions are rejected. An old Server's scalar response to a grouped request raises TransportError.

Worker observations

list_workers(*, filter=None, limit=100, cursor=None, queue=None) -> WorkerPage
count_workers(*, filter=None, group_by=None, limit=None, cursor=None,
              queue=None) -> int | GroupCountPage

These methods also have package-level forms. WorkerPage contains items and next_cursor, sorted by Worker ID ascending with the same 1–1000 page limit. Each WorkerObservation exposes id, queue, route, status, nullable task_id, last_seen_at, and expires_at. Timestamps are UTC. All seven fields support the query language operators for their types.

workers = client.list_workers(filter='route == "sdxl" and status == "idle"')
counts = client.count_workers(group_by=["route", "status"])

Worker grouping supports route and status, singly or together. Plain counting returns an integer; grouped counting follows the Task page contract above. Only unexpired observations are returned. idle means awaiting work; busy includes execution, reporting and cleanup after finish(). Observations renew periodically and can be delayed. An advisory task_id may refer to a terminal or deleted Task. Use Task state for ownership and recovery decisions. Worker observation failures never block Task execution or consume the failure guard. There are no public Worker control methods.

Task submission details

Submission and idempotency

Task data must use strict JSON-compatible Python values. args, metadata, and result are objects; arrays are Python lists; object keys are strings. NaN, Infinity, cycles, arbitrary objects, and integers outside signed 64-bit range are rejected before transport.

When task_id is omitted, the Client generates one before its first network attempt. For retry safety across caller process restarts, persist a caller-chosen ID matching t_[A-Za-z0-9_-]{12} and replay the complete definition:

task = labtasker.submit_task(
    {"seed": 7},
    task_id="t_AbCdEf0123-_",
    name="baseline-seed-7",
    routes=["sdxl"],
)

The same ID and normalized definition return the Task's current representation, even if it is now running or terminal. A different definition at that ID raises APIError(code="task_id_conflict"); it never updates or overwrites the Task. Object-key order, route input order, and explicitly supplied default values do not change the normalized definition.

Selection and pagination

status, exact name, fuzzy name_fuzzy, and filter are combined with AND. limit must be from 1 through 1000. Ordering is stable and supports id, name, status, priority, attempt, max_attempts, last_route, and the public timestamps.

Name search uses list_tasks(name_fuzzy="tr ev") or count_tasks(name_fuzzy="tr ev"). It ignores case and outer whitespace; every whitespace-separated word must be a subsequence of the name, independently of word order. Empty searches do not restrict names. Punctuation is literal and ordering is unchanged. Use name="train_model_eval" or filter='name == "train_model_eval"' for strict equality.

Follow next_cursor with the same Queue, selectors, filter, order field, and direction:

page = labtasker.list_tasks(filter='status == "failed"', limit=100)
tasks = list(page.items)
while page.next_cursor is not None:
    page = labtasker.list_tasks(
        filter='status == "failed"',
        limit=100,
        cursor=page.next_cursor,
    )
    tasks.extend(page.items)

See Query language for the filter grammar. A cursor is opaque and is not a Task ID or offset.

Updates and lifecycle

changes is a TaskUpdate dictionary containing at least one of name, args, metadata, priority, max_attempts, routes, or result. Every supplied object or list is a complete replacement; there is no merge, dot-path patch, or add/remove operator. Unspecified fields remain unchanged.

Bulk update is one Server transaction. matched counts filter matches that are still non-running at execution time; updated counts rows whose stored value actually changed. A concurrent claim either sees the complete new values or wins first and excludes that running Task. If one matched non-running Task violates a state-dependent rule, the whole batch rolls back.

Lifecycle is changed only through these explicit actions; status is never a writable update field.

Queue operations

create_queue(name) -> Queue
list_queues() -> list[Queue]
delete_queue(name, *, cascade=False) -> None

create_queue is idempotent and returns the Queue. list_queues returns the complete, unpaginated list. Deleting an empty Queue succeeds without cascade; a non-empty Queue requires cascade=True. Deletion is rejected while any Task in the Queue is running, even with cascade. A successful cascade atomically deletes the Queue and every Task in it; it does not delete Worker journals or external artifacts.

The default Queue is created only with a fresh database. If explicitly deleted, it must be explicitly recreated.

Response models

Response models are frozen, strict Pydantic models. Known fields keep stable types; unknown response fields are ignored so a newer v2 Server may add optional fields. Use model_dump(mode="json") for a JSON-ready representation. Mutating a dict inside a returned model changes only that local object and never updates the Server.

Task states are exactly pending, running, succeeded, failed, and cancelled. Timestamps are timezone-aware UTC datetime values.

Worker API

@loop(route="default", queue=None, idle_timeout=300,
      force_stop_timeout=None, max_consecutive_failures=5)
def worker(...): ...

TaskArg(default=..., path=None, resolver=None)
task_info() -> TaskInfo
finish(result=None, *, skip_if_no_labtasker=False) -> None
cancellation_requested() -> bool
set_force_stop_timeout(seconds: float | None) -> None

Only parameters marked by TaskArg(...) bind from Task args. Binding is strict and happens after claim; ordinary return succeeds with {}. finish() accepts one JSON object and completes the Task before local cleanup continues. It is stable once accepted and may be called only once. The context helpers require an active Worker execution; cancellation and force-stop helpers require a Python Worker execution.

See Python Workers for binding, cancellation, failure, and Worker-lifetime semantics.

Errors and retries

Invalid Python arguments raise ValueError before transport. Using an active-run helper outside its valid context raises RuntimeError. Operational failures use:

Reads, list/count, and idempotent Task creation use bounded transport retries. Ordinary lifecycle, update, and deletion mutations are not automatically retried after an uncertain response; inspect current state and decide explicitly.

TransientError, TaskError, and FatalWorkerError are Worker outcome signals, not Client-operation errors and not subclasses of LabtaskerError.