harnest.task

All Python modules · Import and compatibility guidance Use task to decorate application work. It returns a TaskCallable whose direct call runs inline and whose defer method submits queued work. The returned TaskHandle exposes status, cancellation, and result access. For discovery, storage setup, and durable waiting, see Create and run Tasks, Task and cron storage, and Durable execution.

Task parameters

Parameter	Contract
`task(function)`	Supports bare `@task` and configured `@task(...)`. The decorated function must have a docstring.
`queue="default"`	1–64 characters. Starts with an ASCII letter; remaining characters may include ASCII letters, digits, `.`, `_`, `~`, and `-`. Set on the decorator, not on `.defer()`.
`max_retries=3`	Integer from 0 to 100. Set on the decorator.
`defer(args, *kwargs)`	Bound against the task’s original signature and serialized as strict JSON. Positional-only parameters and `*args` are unsupported in the decorated task.
`idempotency_key=None`	Optional nonblank text of at most 512 characters. Durable tools derive a replay-stable key when omitted.
`schedule_in=None`	Optional finite, non-negative delay in seconds.

The names idempotency_key and schedule_in are options on .defer(); avoid using them as task input names. Arguments and results must be strict JSON values: no credentials, arbitrary Python objects, non-string object keys, or non-finite numbers. Workers may retry, so task side effects must be idempotent.

Handle behavior

Member	Contract
`id`	Opaque queue job identifier.
`task_name`	Compiler-owned registered task name.
`await status()`	Returns the owning runtime’s job state.
`await cancel()`	Returns whether cancellation succeeded.
`await result()`	Returns the JSON-safe completed result. Unfinished work requires an async `@tool(durable=True)` and Harnest-owned checkpointer so the invocation can suspend and resume.

Handles belong to the compiled runtime that returned them. There is no public handle reconstruction or job-listing API. Calling .defer() without an active compiled runtime raises TaskUnavailableError; calling the decorated function directly does not require queue storage. TaskStore and TaskRecord are storage adapter contracts. TaskDefinition and CompiledTask describe compiler/runtime registrations. Use Task and cron storage to choose a production provider; MemoryTaskStore does not make work survive a process restart.

Storage protocols

Contract	Required operations
`TaskStore`	`start`, `close`, `enqueue_task`, `get_task`, `claim_tasks`, `renew_task_lease`, `finish_task`, `cancel_task`
`CronStore`	`start`, `close`, `create_cron`, `get_cron`, `list_crons`, `update_cron`, `delete_cron`, `list_due_crons`, `commit_cron_occurrence`

Exports

Name	Kind
`CompiledTask`	Class
`MemoryTaskStore`	Class
`PostgresTaskStore`	Class
`TaskCallable`	Class
`TaskDefinition`	Class
`TaskHandle`	Class
`TaskRecord`	Class
`TaskStore`	Class
`TaskStoreConflictError`	Class
`TaskUnavailableError`	Class
`safe_task_result`	Function
`task`	Function

CompiledTask

One compiler-named task executable in the compiled application runtime.

CompiledTask(name: str, source: str, definition: TaskDefinition, authored: TaskCallable[Any]) -> None

Fields (— means no declared default):

Field	Type	Default
`name`	`str`	`—`
`source`	`str`	`—`
`definition`	`TaskDefinition`	`—`
`authored`	`TaskCallable[Any]`	`—`

Parameters and example

Parameter	Type	Default
`name`	`str`	Required
`source`	`str`	Required
`definition`	`TaskDefinition`	Required
`authored`	`TaskCallable[Any]`	Required

from harnest.task import CompiledTask

task = CompiledTask(name=..., source=..., definition=..., authored=...)

CompiledTask.function

Return the original callable only at the trusted worker boundary.

Example

from harnest.task import CompiledTask

task: CompiledTask
value = task.function

CompiledTask.queue

Return the authored queue without exposing the definition object.

Example

from harnest.task import CompiledTask

task: CompiledTask
value = task.queue

CompiledTask.max_retries

Return the bounded retry count used during native registration.

Example

from harnest.task import CompiledTask

task: CompiledTask
value = task.max_retries

MemoryTaskStore

Implement atomic contracts in one process; records do not survive restart.

MemoryTaskStore() -> None

Example

from harnest.task import MemoryTaskStore

store = MemoryTaskStore()

MemoryTaskStore.start

async MemoryTaskStore.start() -> None

Satisfy lifecycle ownership without opening external resources.

Example

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
await store.start()

MemoryTaskStore.close

async MemoryTaskStore.close() -> None

Release no resources; in-memory data lives only with this instance.

Example

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
await store.close()

MemoryTaskStore.enqueue_task

async MemoryTaskStore.enqueue_task(record: TaskRecord) -> TaskRecord

Commit an isolated task snapshot and its deduplication identity.

Parameters and example

Parameter	Type	Default
`record`	`TaskRecord`	Required

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
record = await store.enqueue_task(record=...)

MemoryTaskStore.get_task

async MemoryTaskStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None

Read one detached snapshot through the caller’s ownership scope.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`user_id`	`str \| None`	`None`

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
record = await store.get_task(application_id=..., job_id=...)

MemoryTaskStore.list_task_metadata

async MemoryTaskStore.list_task_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Return a bounded owner-scoped metadata page without copying private payloads.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.list_task_metadata(application_id=..., user_id=...)

MemoryTaskStore.claim_tasks

async MemoryTaskStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]

Claim due work, reclaim expired leases and retire exhausted attempts.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`queues`	`tuple[str, ...]`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required
`limit`	`int`	`1`

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.claim_tasks(
    application_id=...,
    queues=...,
    now=...,
    lease_seconds=...,
)

MemoryTaskStore.renew_task_lease

async MemoryTaskStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool

Extend only a lease that still grants execution ownership.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.renew_task_lease(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    lease_seconds=...,
)

MemoryTaskStore.finish_task

async MemoryTaskStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool

Commit an outcome only while the exact execution lease remains valid.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`status`	`str`	Required
`result`	`Any`	`None`
`failure_code`	`str \| None`	`None`
`retry_at`	`float \| None`	`None`

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.finish_task(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    status=...,
)

MemoryTaskStore.cancel_task

async MemoryTaskStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool

Cancel one owner’s active work and revoke any in-flight attempt.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`job_id`	`str`	Required
`now`	`float`	Required

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)

MemoryTaskStore.create_cron

async MemoryTaskStore.create_cron(record: CronRecord) -> CronRecord

Create a schedule or return an exact keyed retry without reactivation.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
record = await store.create_cron(record=...)

MemoryTaskStore.get_cron

async MemoryTaskStore.get_cron(*, application_id: str, user_id: str, schedule_id: str) -> CronRecord | None

Read a detached schedule through both ownership predicates.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
record = await store.get_cron(application_id=..., user_id=..., schedule_id=...)

MemoryTaskStore.list_crons

async MemoryTaskStore.list_crons(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Return one bounded page from this in-memory reference collection.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.list_crons(application_id=..., user_id=...)

MemoryTaskStore.list_cron_metadata

async MemoryTaskStore.list_cron_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Project an owner-scoped schedule page without copying task arguments.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.list_cron_metadata(application_id=..., user_id=...)

MemoryTaskStore.update_cron

async MemoryTaskStore.update_cron(record: CronRecord, *, expected_revision: int) -> CronRecord

Replace allowed fields only at the current scoped revision.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required
`expected_revision`	`int`	Required

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
record = await store.update_cron(record=..., expected_revision=...)

MemoryTaskStore.delete_cron

async MemoryTaskStore.delete_cron(*, application_id: str, user_id: str, schedule_id: str) -> bool

Remove one owner’s schedule and its reusable authored key.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.delete_cron(application_id=..., user_id=..., schedule_id=...)

MemoryTaskStore.list_due_crons

async MemoryTaskStore.list_due_crons(
    *,
    application_id: str,
    now: float,
    after: tuple[float, str] | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Read active due schedules with a stable timestamp-and-ID cursor.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`now`	`float`	Required
`after`	`tuple[float, str] \| None`	`None`
`limit`	`int`	`100`

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
result = await store.list_due_crons(application_id=..., now=...)

MemoryTaskStore.commit_cron_occurrence

async MemoryTaskStore.commit_cron_occurrence(
    *,
    application_id: str,
    user_id: str,
    schedule_id: str,
    expected_revision: int,
    due_at: float,
    next_run_at: float,
    task: TaskRecord,
) -> TaskRecord | None

Use one lock for revision check, task insert and schedule advancement.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required
`expected_revision`	`int`	Required
`due_at`	`float`	Required
`next_run_at`	`float`	Required
`task`	`TaskRecord`	Required

from harnest.task import MemoryTaskStore

store: MemoryTaskStore
record = await store.commit_cron_occurrence(
    application_id=...,
    user_id=...,
    schedule_id=...,
    expected_revision=...,
    due_at=...,
    next_run_at=...,
    task=...,
)

PostgresTaskStore

Persist jobs and schedules with transactional claims and fenced leases. _pool supports composing this provider with the combined session store; externally supplied pools remain owned by their caller.

PostgresTaskStore(
    dsn: str,
    *,
    pool_options: Mapping[str, Any] | None = None,
    setup_schema: bool = True,
) -> None

Parameters and example

Parameter	Type	Default
`dsn`	`str`	Required
`pool_options`	`Mapping[str, Any] \| None`	`None`
`setup_schema`	`bool`	`True`

from harnest.task import PostgresTaskStore

store = PostgresTaskStore(dsn=...)

PostgresTaskStore.start

async PostgresTaskStore.start() -> None

Open one pool and install or validate the durable schema.

Example

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
await store.start()

PostgresTaskStore.close

async PostgresTaskStore.close() -> None

Close only the pool opened by this instance.

Example

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
await store.close()

PostgresTaskStore.create_cron

async PostgresTaskStore.create_cron(record: CronRecord) -> CronRecord

Create a schedule once without reactivating an existing cancelled key.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
record = await store.create_cron(record=...)

PostgresTaskStore.get_cron

async PostgresTaskStore.get_cron(*, application_id: str, user_id: str, schedule_id: str) -> CronRecord | None

Read a schedule only within its application and owner scope.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
record = await store.get_cron(application_id=..., user_id=..., schedule_id=...)

PostgresTaskStore.list_crons

async PostgresTaskStore.list_crons(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Apply owner filtering and keyset pagination in the database.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.list_crons(application_id=..., user_id=...)

PostgresTaskStore.update_cron

async PostgresTaskStore.update_cron(record: CronRecord, *, expected_revision: int) -> CronRecord

Replace one schedule conditionally, preserving its immutable identity.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required
`expected_revision`	`int`	Required

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
record = await store.update_cron(record=..., expected_revision=...)

PostgresTaskStore.delete_cron

async PostgresTaskStore.delete_cron(*, application_id: str, user_id: str, schedule_id: str) -> bool

Delete a schedule using the same owner predicate as reads.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.delete_cron(application_id=..., user_id=..., schedule_id=...)

PostgresTaskStore.list_due_crons

async PostgresTaskStore.list_due_crons(
    *,
    application_id: str,
    now: float,
    after: tuple[float, str] | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Page due schedules in database order without loading inactive rows.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`now`	`float`	Required
`after`	`tuple[float, str] \| None`	`None`
`limit`	`int`	`100`

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.list_due_crons(application_id=..., now=...)

PostgresTaskStore.commit_cron_occurrence

async PostgresTaskStore.commit_cron_occurrence(
    *,
    application_id: str,
    user_id: str,
    schedule_id: str,
    expected_revision: int,
    due_at: float,
    next_run_at: float,
    task: TaskRecord,
) -> TaskRecord | None

Atomically enqueue an occurrence and advance its schedule revision.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required
`expected_revision`	`int`	Required
`due_at`	`float`	Required
`next_run_at`	`float`	Required
`task`	`TaskRecord`	Required

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
record = await store.commit_cron_occurrence(
    application_id=...,
    user_id=...,
    schedule_id=...,
    expected_revision=...,
    due_at=...,
    next_run_at=...,
    task=...,
)

PostgresTaskStore.enqueue_task

async PostgresTaskStore.enqueue_task(record: TaskRecord) -> TaskRecord

Persist an immutable task identity, returning exact idempotent retries.

Parameters and example

Parameter	Type	Default
`record`	`TaskRecord`	Required

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
record = await store.enqueue_task(record=...)

PostgresTaskStore.get_task

async PostgresTaskStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None

Read a scoped job; omitted user scope is reserved for trusted workers.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`user_id`	`str \| None`	`None`

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
record = await store.get_task(application_id=..., job_id=...)

PostgresTaskStore.claim_tasks

async PostgresTaskStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]

Claim disjoint batches and retire exhausted attempts after crashes.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`queues`	`tuple[str, ...]`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required
`limit`	`int`	`1`

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.claim_tasks(
    application_id=...,
    queues=...,
    now=...,
    lease_seconds=...,
)

PostgresTaskStore.renew_task_lease

async PostgresTaskStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool

Renew only an unexpired attempt still owned by the calling worker.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.renew_task_lease(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    lease_seconds=...,
)

PostgresTaskStore.finish_task

async PostgresTaskStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool

Fence completion and retries; terminal transitions erase private input.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`status`	`str`	Required
`result`	`Any`	`None`
`failure_code`	`str \| None`	`None`
`retry_at`	`float \| None`	`None`

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.finish_task(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    status=...,
)

PostgresTaskStore.cancel_task

async PostgresTaskStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool

Cancel an owned nonterminal job and revoke any running attempt.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`job_id`	`str`	Required
`now`	`float`	Required

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)

PostgresTaskStore.list_task_metadata

async PostgresTaskStore.list_task_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Push owner, cursor, projection, ordering, and limit into one database query.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.list_task_metadata(application_id=..., user_id=...)

PostgresTaskStore.list_cron_metadata

async PostgresTaskStore.list_cron_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Read a bounded metadata projection with owner filtering in one SQL query.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.task import PostgresTaskStore

store: PostgresTaskStore
result = await store.list_cron_metadata(application_id=..., user_id=...)

TaskCallable

Preserve direct invocation while adding an asynchronous defer boundary.

TaskCallable(definition: TaskDefinition) -> None

Parameters and example

Parameter	Type	Default
`definition`	`TaskDefinition`	Required

from harnest.task import TaskCallable

task_callable = TaskCallable(definition=...)

TaskCallable.call

TaskCallable.__call__(*args: Any, **kwargs: Any) -> Any

Execute inline when the author deliberately calls the task itself.

Parameters and example

Parameter	Type	Default	Description
`*args`	`Any`	Optional	Positional arguments for the task function.
`**kwargs`	`Any`	Optional	Keyword arguments for the task function.

# build_report is your @task function
report = await build_report(account_id="acct_123")  # runs inline, no queue

TaskCallable.defer

async TaskCallable.defer(
    *args: Any,
    idempotency_key: str | None = None,
    schedule_in: float | None = None,
    **kwargs: Any,
) -> TaskHandle

Commit safe arguments to the active compiled task runtime.

Parameters and example

Parameter	Type	Default	Description
`*args`	`Any`	Optional	Positional arguments for the task function.
`idempotency_key`	`str \| None`	`None`	Optional key, at most 512 characters, that deduplicates submissions.
`schedule_in`	`float \| None`	`None`	Optional delay in seconds before the task can run.
`**kwargs`	`Any`	Optional	Keyword arguments for the task function, stored as strict JSON.

# build_report is your @task function
handle = await build_report.defer(account_id="acct_123")

TaskDefinition

Authoring metadata retained on a decorated function.

TaskDefinition(function: Callable[..., Any], queue: str, max_retries: int) -> None

Fields (— means no declared default):

Field	Type	Default
`function`	`Callable[..., Any]`	`—`
`queue`	`str`	`—`
`max_retries`	`int`	`—`

Parameters and example

Parameter	Type	Default
`function`	`Callable[..., Any]`	Required
`queue`	`str`	Required
`max_retries`	`int`	Required

from harnest.task import TaskDefinition

definition = TaskDefinition(function=..., queue=..., max_retries=...)

TaskHandle

Opaque queue handle returned after a job is committed. Obtain this object from the owning Harnest API; its constructor includes runtime bindings. Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`task_name`	`str`	`—`

Example

from harnest.task import TaskHandle

handle: TaskHandle = await build_report.defer(account_id="acct_123")

TaskHandle.status

async TaskHandle.status() -> str

Read the native job state through the owning runtime.

Example

from harnest.task import TaskHandle

handle: TaskHandle
result = await handle.status()

TaskHandle.cancel

async TaskHandle.cancel() -> bool

Cancel queued work without exposing the backend application.

Example

from harnest.task import TaskHandle

handle: TaskHandle
result = await handle.cancel()

TaskHandle.result

async TaskHandle.result() -> Any

Return completed work or suspend through a durable managed tool.

Example

from harnest.task import TaskHandle

handle: TaskHandle
result = await handle.result()

TaskRecord

Private persisted job, including reconstructable invocation capabilities. Epoch timestamps are UTC seconds. Attempts count claims, starting at one; max_retries permits that many attempts after the first. Providers retain an immutable enqueue fingerprint after clearing terminal private payloads.

TaskRecord(
    job_id: str,
    application_id: str,
    user_id: str,
    task_name: str,
    queue: str,
    arguments: Mapping[str, Any],
    invocation: Mapping[str, Any] | None = None,
    agent_permissions: tuple[str, ...] | None = None,
    trigger: str = 'agent',
    status: str = 'pending',
    scheduled_at: float = 0.0,
    max_retries: int = 3,
    attempt: int = 0,
    lease_token: str | None = None,
    lease_expires_at: float | None = None,
    result: Any = None,
    failure_code: str | None = None,
    idempotency_key: str | None = None,
    created_at: float = 0.0,
    updated_at: float = 0.0,
    cron_schedule_id: str | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`job_id`	`str`	`—`
`application_id`	`str`	`—`
`user_id`	`str`	`—`
`task_name`	`str`	`—`
`queue`	`str`	`—`
`arguments`	`Mapping[str, Any]`	`—`
`invocation`	`Mapping[str, Any] \| None`	`None`
`agent_permissions`	`tuple[str, ...] \| None`	`None`
`trigger`	`str`	`'agent'`
`status`	`str`	`'pending'`
`scheduled_at`	`float`	`0.0`
`max_retries`	`int`	`3`
`attempt`	`int`	`0`
`lease_token`	`str \| None`	`None`
`lease_expires_at`	`float \| None`	`None`
`result`	`Any`	`None`
`failure_code`	`str \| None`	`None`
`idempotency_key`	`str \| None`	`None`
`created_at`	`float`	`0.0`
`updated_at`	`float`	`0.0`
`cron_schedule_id`	`str \| None`	`None`

Parameters and example

Parameter	Type	Default
`job_id`	`str`	Required
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_name`	`str`	Required
`queue`	`str`	Required
`arguments`	`Mapping[str, Any]`	Required
`invocation`	`Mapping[str, Any] \| None`	`None`
`agent_permissions`	`tuple[str, ...] \| None`	`None`
`trigger`	`str`	`'agent'`
`status`	`str`	`'pending'`
`scheduled_at`	`float`	`0.0`
`max_retries`	`int`	`3`
`attempt`	`int`	`0`
`lease_token`	`str \| None`	`None`
`lease_expires_at`	`float \| None`	`None`
`result`	`Any`	`None`
`failure_code`	`str \| None`	`None`
`idempotency_key`	`str \| None`	`None`
`created_at`	`float`	`0.0`
`updated_at`	`float`	`0.0`
`cron_schedule_id`	`str \| None`	`None`

from harnest.task import TaskRecord

record = TaskRecord(
    job_id=...,
    application_id=...,
    user_id=...,
    task_name=...,
    queue=...,
    arguments=...,
)

TaskStore

Atomic operations required for at-least-once task execution. All selection, scope predicates, ordering, and limits execute inside the datastore. A provider must durably commit before returning success. Expired lease holders cannot renew or finish work. Terminal transitions clear arguments, invocation and permissions while retaining the result/failure.

TaskStore(*args, **kwargs)

Example

from harnest.task import TaskStore

class MyTaskStore(TaskStore):
    ...

TaskStore.start

async TaskStore.start() -> None

Open owned resources and initialize the provider’s durable schema.

Example

from harnest.task import TaskStore

store: TaskStore
await store.start()

TaskStore.close

async TaskStore.close() -> None

Release owned resources after workers and dispatchers have stopped.

Example

from harnest.task import TaskStore

store: TaskStore
await store.close()

TaskStore.enqueue_task

async TaskStore.enqueue_task(record: TaskRecord) -> TaskRecord

Atomically persist payload and job, or return an identical submission. Idempotency is scoped by application, user, task name and optional key. Conflicting definitions raise TaskStoreConflictError. Persist a separate fingerprint so terminal payload cleanup cannot defeat deduplication.

Parameters and example

Parameter	Type	Default
`record`	`TaskRecord`	Required

from harnest.task import TaskStore

store: TaskStore
record = await store.enqueue_task(record=...)

TaskStore.get_task

async TaskStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None

Read one scoped job; omitted user is a trusted worker-only lookup.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`user_id`	`str \| None`	`None`

from harnest.task import TaskStore

store: TaskStore
record = await store.get_task(application_id=..., job_id=...)

TaskStore.claim_tasks

async TaskStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]

Atomically claim due pending or expired running jobs in queue scope. Return a bounded eligible batch with best-effort earliest-due ordering; concurrent workers and lease recovery do not guarantee strict FIFO. Each claim increments attempt and issues a fresh lease token. Expired final attempts become failed with task_failed instead of remaining stranded or retrying indefinitely. Empty queues selects no work. Limits are integers from 1 to 1000 and apply to returned claims.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`queues`	`tuple[str, ...]`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required
`limit`	`int`	`1`

from harnest.task import TaskStore

store: TaskStore
result = await store.claim_tasks(
    application_id=...,
    queues=...,
    now=...,
    lease_seconds=...,
)

TaskStore.renew_task_lease

async TaskStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool

Extend a running, unexpired matching lease; stale owners return false.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required

from harnest.task import TaskStore

store: TaskStore
result = await store.renew_task_lease(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    lease_seconds=...,
)

TaskStore.finish_task

async TaskStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool

Fence a completion, failure or retry with a matching unexpired lease. Status is completed, failed, or pending. Pending requires retry_at and remaining retries; it preserves payload. All transitions clear leases. Stale owners return false and must never overwrite another attempt.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`status`	`str`	Required
`result`	`Any`	`None`
`failure_code`	`str \| None`	`None`
`retry_at`	`float \| None`	`None`

from harnest.task import TaskStore

store: TaskStore
result = await store.finish_task(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    status=...,
)

TaskStore.cancel_task

async TaskStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool

Atomically cancel pending/running work, invalidate leases and scrub. Unknown, foreign and terminal jobs return false. A running effect may already have happened; cancellation prevents subsequent state commits.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`job_id`	`str`	Required
`now`	`float`	Required

from harnest.task import TaskStore

store: TaskStore
result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)

TaskStoreConflictError

A stable job identity or idempotency key has a different definition.

Example

from harnest.task import TaskStoreConflictError

try:
    ...
except TaskStoreConflictError as error:
    ...

TaskUnavailableError

A task was deferred without its compiled runtime being active.

Example

from harnest.task import TaskUnavailableError

try:
    ...
except TaskUnavailableError as error:
    ...

safe_task_result

Normalize a task result to the strict JSON values safe for persistence.

safe_task_result(value: Any) -> Any

Parameters and example

Parameter	Type	Default
`value`	`Any`	Required

from harnest.task import safe_task_result

result = safe_task_result(value=...)

task

Mark a callable for queue, scheduling, and retry execution. Deferred execution uses Harnest workers and the application’s configured lifecycle.storage.tasks provider. Durable storage makes queued work recoverable across restarts; importing this API requires no queue library. Direct calls remain ordinary Python calls; defer is queue execution and never promises restoration of a suspended Python frame.

task(function: F | None = None, *, queue: str = 'default', max_retries: int = 3) -> Any

Parameters and example

Parameter	Type	Default	Description
`function`	`F \| None`	`None`	The decorated function. Filled in by bare `@task`; omit it when passing options. Needs a docstring and JSON-safe arguments.
`queue`	`str`	`'default'`	Queue name, 1–64 characters, starting with a letter. Set it here, not on `.defer()`.
`max_retries`	`int`	`3`	Retries after a failed run, from 0 to 100.

from harnest.task import task

@task(queue="reports", max_retries=3)
async def build_report(account_id: str) -> dict:
    """Build one account report."""
    return {"account_id": account_id}

​Task parameters

​Handle behavior

​Storage protocols

​Exports

​CompiledTask

​CompiledTask.function

​CompiledTask.queue

​CompiledTask.max_retries

​MemoryTaskStore

​MemoryTaskStore.start

​MemoryTaskStore.close

​MemoryTaskStore.enqueue_task

​MemoryTaskStore.get_task

​MemoryTaskStore.list_task_metadata

​MemoryTaskStore.claim_tasks

​MemoryTaskStore.renew_task_lease

​MemoryTaskStore.finish_task

​MemoryTaskStore.cancel_task

​MemoryTaskStore.create_cron

​MemoryTaskStore.get_cron

​MemoryTaskStore.list_crons

​MemoryTaskStore.list_cron_metadata

​MemoryTaskStore.update_cron

​MemoryTaskStore.delete_cron

​MemoryTaskStore.list_due_crons

​MemoryTaskStore.commit_cron_occurrence

​PostgresTaskStore

​PostgresTaskStore.start

​PostgresTaskStore.close

​PostgresTaskStore.create_cron

​PostgresTaskStore.get_cron

​PostgresTaskStore.list_crons

​PostgresTaskStore.update_cron

​PostgresTaskStore.delete_cron

​PostgresTaskStore.list_due_crons

​PostgresTaskStore.commit_cron_occurrence

​PostgresTaskStore.enqueue_task

​PostgresTaskStore.get_task

​PostgresTaskStore.claim_tasks

​PostgresTaskStore.renew_task_lease

​PostgresTaskStore.finish_task

​PostgresTaskStore.cancel_task

​PostgresTaskStore.list_task_metadata

​PostgresTaskStore.list_cron_metadata

​TaskCallable

​TaskCallable.call

​TaskCallable.defer

​TaskDefinition

​TaskHandle

​TaskHandle.status

​TaskHandle.cancel

​TaskHandle.result

​TaskRecord

​TaskStore

​TaskStore.start

​TaskStore.close

​TaskStore.enqueue_task

​TaskStore.get_task

​TaskStore.claim_tasks

​TaskStore.renew_task_lease

​TaskStore.finish_task

​TaskStore.cancel_task

​TaskStoreConflictError

​TaskUnavailableError

​safe_task_result

​task

Task parameters

Handle behavior

Storage protocols

Exports

CompiledTask

CompiledTask.function

CompiledTask.queue

CompiledTask.max_retries

MemoryTaskStore

MemoryTaskStore.start

MemoryTaskStore.close

MemoryTaskStore.enqueue_task

MemoryTaskStore.get_task

MemoryTaskStore.list_task_metadata

MemoryTaskStore.claim_tasks

MemoryTaskStore.renew_task_lease

MemoryTaskStore.finish_task

MemoryTaskStore.cancel_task

MemoryTaskStore.create_cron

MemoryTaskStore.get_cron

MemoryTaskStore.list_crons

MemoryTaskStore.list_cron_metadata

MemoryTaskStore.update_cron

MemoryTaskStore.delete_cron

MemoryTaskStore.list_due_crons

MemoryTaskStore.commit_cron_occurrence

PostgresTaskStore

PostgresTaskStore.start

PostgresTaskStore.close

PostgresTaskStore.create_cron

PostgresTaskStore.get_cron

PostgresTaskStore.list_crons

PostgresTaskStore.update_cron

PostgresTaskStore.delete_cron

PostgresTaskStore.list_due_crons

PostgresTaskStore.commit_cron_occurrence

PostgresTaskStore.enqueue_task

PostgresTaskStore.get_task

PostgresTaskStore.claim_tasks

PostgresTaskStore.renew_task_lease

PostgresTaskStore.finish_task

PostgresTaskStore.cancel_task

PostgresTaskStore.list_task_metadata

PostgresTaskStore.list_cron_metadata

TaskCallable

TaskCallable.call

TaskCallable.defer

TaskDefinition

TaskHandle

TaskHandle.status

TaskHandle.cancel

TaskHandle.result

TaskRecord

TaskStore

TaskStore.start

TaskStore.close

TaskStore.enqueue_task

TaskStore.get_task

TaskStore.claim_tasks

TaskStore.renew_task_lease

TaskStore.finish_task

TaskStore.cancel_task

TaskStoreConflictError

TaskUnavailableError

safe_task_result

task