harnest.cron

All Python modules · Import and compatibility guidance Use Cron for a fixed application-owned declaration. Use create to obtain a CronJob for a user-owned schedule. The module functions and CronJob methods manage dynamic schedules; Cron itself has no pause, resume, or delete methods. For setup and usage, see Cron schedules, Dynamic schedules, and Task and cron storage.

Declare a cron function

@cron(...) creates a queued task for a same-named function in a root cron/<name>.py file. Give the function a docstring. A fixed expression uses the supplied arguments on each occurrence. Omit the expression for dynamic scheduling only; in that form, do not supply arguments or a non-UTC timezone. Do not stack @cron with @task because the decorator creates its own task. Direct calls run inline. queue="default" and max_retries=3 use the Task validation rules.

Schedule parameters

Parameter	Contract
`Cron.schedule` / `expression`	Exactly five numeric cron fields: minute, hour, day of month, month, day of week. Supports `*`, comma lists, ascending ranges, and positive `/step`; rejects outer whitespace.
`timezone`	`"UTC"` only. Dynamic schedules always use UTC.
`Cron.task`	A Harnest task callable discovered in the application.
`create(task=...)`	A discovered `@task` or `@cron` callable, or the name of a deployed `@cron` function. A string does not import or register code.
`arguments`	Strict JSON-safe keyword arguments matching the target signature. An update replaces the entire mapping. Omit it to preserve current arguments; `None` is not an update mapping.
`key`	1–128 characters. Starts with an ASCII letter or digit; remaining characters may also include `.`, `_`, `:`, `~`, and `-`. Scoped to application and user.
`create(max_runs=...)`	Optional positive integer, at most 1,000,000. Cancels the schedule after that many occurrences are queued.
`create(max_consecutive_failures=...)`	Optional positive integer, at most 1,000,000. Cancels after that many terminally failed occurrences in a row; success resets the count. The decorator’s `max_retries` applies within one occurrence.
`schedule_id` / `CronJob.id`	Opaque `cron_` ID returned by Harnest. Keep it for later owner-scoped operations.
`list(after=..., limit=50)`	Returns a tuple ordered by ID. Pass the final job ID as the exclusive next-page cursor; `limit` must be an integer from 1 to 100.

Dynamic schedule behavior

Calls require an active managed invocation and a cron-enabled runtime. Register the same durable provider for tasks and cron. Each operation checks the active application and user; retaining a CronJob does not grant access outside that scope.

Operation	Return and behavior
`create(...)`	A `CronJob`. Exact retries with the same key and definition return the existing record, including paused or cancelled jobs. A different definition conflicts.
`get(id)`	A `CronJob`, or `None` for a missing or unowned ID.
`update(id, ...)` / `job.update(...)`	A new `CronJob` snapshot with supplied expression or arguments replaced. Preserves ID, key, owner, and target.
`pause(id)` / `job.pause()`	A new snapshot in `paused` state; stops future occurrences.
`resume(id)` / `job.resume()`	A new snapshot in `active` state; a paused schedule resumes at its next matching UTC minute.
`cancel(id)` / `job.cancel()`	A new snapshot in terminal `cancelled` state. Retains the record.
`delete(id)` / `job.delete()`	`True` if an owned record was removed, otherwise `False`.

CronJob is immutable. Assign the returned snapshot, for example job = await job.pause(), or fetch it again to read updated fields. Cancelling or deleting a schedule does not cancel tasks already queued or running. Cancelled schedules cannot be updated, paused, or resumed. CronJob.run_count counts queued occurrences; CronJob.consecutive_failures counts final failed outcomes since the last success. Reaching either configured limit cancels future scheduling.

Errors

Error	When it occurs
`CronUnavailableError`	No active managed invocation, no cron-enabled runtime, or a job used with a different runtime.
`CronNotFoundError`	Update, pause, resume, or cancel targets a missing or unowned job.
`CronConflictError`	Conflicting key reuse, an invalid cancelled-state change, an unavailable target during update, or a conflicting revision.
`CronRuntimeError`	Durable persistence or dispatch fails.
`ValueError` / `TypeError`	Invalid expression, ID, key, limit, target, or task arguments.
`CronStoreConflictError`	Provider-level conflict for custom storage implementations.

CronRecord, CronStore, and CompiledCron are integration contracts. Application tools normally work with CronJob and the module functions.

Exports

Name	Kind
`CompiledCron`	Class
`Cron`	Class
`CronConflictError`	Class
`CronJob`	Class
`CronNotFoundError`	Class
`CronRecord`	Class
`CronRuntimeError`	Class
`CronStore`	Class
`CronStoreConflictError`	Class
`CronUnavailableError`	Class
`cancel`	Function
`create`	Function
`cron`	Function
`delete`	Function
`get`	Function
`list`	Function
`pause`	Function
`resume`	Function
`update`	Function

CompiledCron

One compiler-resolved schedule with a stable application-owned identity.

CompiledCron(
    name: str,
    source: str,
    schedule: str,
    timezone: str,
    task: CompiledTask,
    arguments: Mapping[str, Any],
) -> None

Fields (— means no declared default):

Field	Type	Default
`name`	`str`	`—`
`source`	`str`	`—`
`schedule`	`str`	`—`
`timezone`	`str`	`—`
`task`	`CompiledTask`	`—`
`arguments`	`Mapping[str, Any]`	`—`

Parameters and example

Parameter	Type	Default
`name`	`str`	Required
`source`	`str`	Required
`schedule`	`str`	Required
`timezone`	`str`	Required
`task`	`CompiledTask`	Required
`arguments`	`Mapping[str, Any]`	Required

from harnest.cron import CompiledCron

cron = CompiledCron(
    name=...,
    source=...,
    schedule=...,
    timezone=...,
    task=...,
    arguments=...,
)

CompiledCron.task_name

Return the compiler-owned native task name targeted by the schedule.

Example

from harnest.cron import CompiledCron

cron: CompiledCron
value = cron.task_name

Cron

Schedule one discovered task using a strict five-column UTC expression.

Cron(
    schedule: str,
    task: TaskCallable[Any],
    arguments: Mapping[str, Any] = ...,
    timezone: str = 'UTC',
) -> None

Fields (— means no declared default):

Field	Type	Default
`schedule`	`str`	`—`
`task`	`TaskCallable[Any]`	`—`
`arguments`	`Mapping[str, Any]`	`dict()`
`timezone`	`str`	`'UTC'`

Parameters and example

Parameter	Type	Default	Description
`schedule`	`str`	Required	Five-field UTC cron expression.
`task`	`TaskCallable[Any]`	Required	A discovered Harnest task callable.
`arguments`	`Mapping[str, Any]`	`...`	JSON-safe keyword arguments for each occurrence.
`timezone`	`str`	`'UTC'`	Only `"UTC"` is supported.

from harnest.cron import Cron

daily_report = Cron("0 9 * * 1-5", task=build_report)

CronConflictError

Raised when a schedule key already owns a different definition.

Example

from harnest.cron import CronConflictError

try:
    ...
except CronConflictError as error:
    ...

CronJob

One user-owned recurring schedule retained by a running application. Obtain this object from the owning Harnest API; its constructor includes runtime bindings. Fields (— means no declared default):

Field	Type	Default
`id`	`str`	`—`
`key`	`str`	`—`
`expression`	`str`	`—`
`task_name`	`str`	`—`
`arguments`	`Mapping[str, Any]`	`—`
`status`	`str`	`—`
`timezone`	`str`	`'UTC'`
`max_runs`	`int \| None`	`None`
`max_consecutive_failures`	`int \| None`	`None`
`run_count`	`int`	`0`
`consecutive_failures`	`int`	`0`

Example

from harnest import cron

job = await cron.create(
    key="weekly-report",
    expression="0 9 * * 1",
    task="build_report",
)

CronJob.update

async CronJob.update(*, expression: str | None = None, arguments: Mapping[str, Any] | object = ...) -> CronJob

Replace supplied fields while preserving this schedule’s identity.

Parameters and example

Parameter	Type	Default	Description
`expression`	`str \| None`	`None`	New expression. Omit to keep the current one.
`arguments`	`Mapping[str, Any] \| object`	`...`	New arguments; replaces the whole mapping.

from harnest.cron import CronJob

job: CronJob
job = await job.update(expression="0 8 * * 1")

CronJob.pause

async CronJob.pause() -> CronJob

Stop future occurrences without deleting the schedule.

Example

from harnest.cron import CronJob

job: CronJob
job = await job.pause()

CronJob.resume

async CronJob.resume() -> CronJob

Resume future occurrences from the next matching UTC minute.

Example

from harnest.cron import CronJob

job: CronJob
job = await job.resume()

CronJob.cancel

async CronJob.cancel() -> CronJob

Permanently stop future occurrences while retaining this record.

Example

from harnest.cron import CronJob

job: CronJob
job = await job.cancel()

CronJob.delete

async CronJob.delete() -> bool

Delete this schedule within the currently active owner scope.

Example

from harnest.cron import CronJob

job: CronJob
result = await job.delete()

CronNotFoundError

Raised when a user-owned dynamic schedule no longer exists.

Example

from harnest.cron import CronNotFoundError

try:
    ...
except CronNotFoundError as error:
    ...

CronRecord

Private user-owned schedule with a durable next-occurrence cursor.

CronRecord(
    schedule_id: str,
    application_id: str,
    user_id: str,
    key: str,
    expression: str,
    task_name: str,
    arguments: Mapping[str, Any],
    next_run_at: float,
    status: str = 'active',
    revision: int = 0,
    timezone: str = 'UTC',
    created_at: float = 0.0,
    updated_at: float = 0.0,
    max_runs: int | None = None,
    max_consecutive_failures: int | None = None,
    run_count: int = 0,
    consecutive_failures: int = 0,
) -> None

Fields (— means no declared default):

Field	Type	Default
`schedule_id`	`str`	`—`
`application_id`	`str`	`—`
`user_id`	`str`	`—`
`key`	`str`	`—`
`expression`	`str`	`—`
`task_name`	`str`	`—`
`arguments`	`Mapping[str, Any]`	`—`
`next_run_at`	`float`	`—`
`status`	`str`	`'active'`
`revision`	`int`	`0`
`timezone`	`str`	`'UTC'`
`created_at`	`float`	`0.0`
`updated_at`	`float`	`0.0`
`max_runs`	`int \| None`	`None`
`max_consecutive_failures`	`int \| None`	`None`
`run_count`	`int`	`0`
`consecutive_failures`	`int`	`0`

Parameters and example

Parameter	Type	Default
`schedule_id`	`str`	Required
`application_id`	`str`	Required
`user_id`	`str`	Required
`key`	`str`	Required
`expression`	`str`	Required
`task_name`	`str`	Required
`arguments`	`Mapping[str, Any]`	Required
`next_run_at`	`float`	Required
`status`	`str`	`'active'`
`revision`	`int`	`0`
`timezone`	`str`	`'UTC'`
`created_at`	`float`	`0.0`
`updated_at`	`float`	`0.0`
`max_runs`	`int \| None`	`None`
`max_consecutive_failures`	`int \| None`	`None`
`run_count`	`int`	`0`
`consecutive_failures`	`int`	`0`

from harnest.cron import CronRecord

record = CronRecord(
    schedule_id=...,
    application_id=...,
    user_id=...,
    key=...,
    expression=...,
    task_name=...,
    arguments=...,
    next_run_at=...,
)

CronRuntimeError

Raised when durable cron persistence or dispatch fails safely.

Example

from harnest.cron import CronRuntimeError

try:
    ...
except CronRuntimeError as error:
    ...

CronStore

Durable CRUD and occurrence handoff on the same provider as TaskStore. User calls always include application and user predicates. Ordered queries and pagination run in the datastore. Cron parsing and next-run calculation belong to Harnest core. Providers enforce revisions and atomic handoff.

CronStore(*args, **kwargs)

Example

from harnest.cron import CronStore

class MyCronStore(CronStore):
    ...

CronStore.start

async CronStore.start() -> None

Open provider resources; a shared TaskStore instance starts once.

Example

from harnest.cron import CronStore

store: CronStore
await store.start()

CronStore.close

async CronStore.close() -> None

Release provider resources after the owning runtime stops dispatch.

Example

from harnest.cron import CronStore

store: CronStore
await store.close()

CronStore.create_cron

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

Create, or return exact definition at application/user/key identity. Exact retries return existing paused/cancelled schedules unchanged. Different definitions and conflicting schedule IDs raise conflict.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required

from harnest.cron import CronStore

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

CronStore.get_cron

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

Read one owner-scoped schedule without disclosing other owners.

Parameters and example

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

from harnest.cron import CronStore

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

CronStore.list_crons

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

Return schedules after the ID cursor; limit must be from 1 to 1000.

Parameters and example

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

from harnest.cron import CronStore

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

CronStore.update_cron

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

Replace mutable fields and increment revision with an atomic CAS. IDs, owner, key and task name are immutable. Cancelled is terminal. An identical cancelled retry returns the existing record unchanged. Missing records raise KeyError; state/revision conflicts raise conflict.

Parameters and example

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

from harnest.cron import CronStore

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

CronStore.delete_cron

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

Remove only an owner-scoped schedule; missing/foreign returns false.

Parameters and example

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

from harnest.cron import CronStore

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

CronStore.list_due_crons

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

Page active due schedules by time then ID; limit is from 1 to 1000.

Parameters and example

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

from harnest.cron import CronStore

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

CronStore.commit_cron_occurrence

async CronStore.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 one occurrence and advance the active schedule. Require matching owner, revision and next_run_at==due_at, and a task matching the schedule owner/target/arguments. next_run_at must advance. Stale or inactive returns None with no task insert. Duplicate deliveries cannot create another task. Advance revision within the same transaction.

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.cron import CronStore

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

CronStoreConflictError

A schedule definition, terminal state or expected revision conflicts.

Example

from harnest.cron import CronStoreConflictError

try:
    ...
except CronStoreConflictError as error:
    ...

CronUnavailableError

Raised when dynamic schedules are used outside a cron-enabled runtime.

Example

from harnest.cron import CronUnavailableError

try:
    ...
except CronUnavailableError as error:
    ...

cancel

Permanently stop future occurrences while retaining the job record.

async cancel(schedule_id: str) -> CronJob

Parameters and example

Parameter	Type	Default	Description
`schedule_id`	`str`	Required	The job’s `cron_` ID.

from harnest import cron

job = await cron.cancel("cron_...")

create

Schedule registered work with optional dispatch and failure limits.

async create(
    *,
    key: str,
    expression: str,
    task: TaskCallable[Any] | str,
    arguments: Mapping[str, Any] | None = None,
    max_runs: int | None = None,
    max_consecutive_failures: int | None = None,
) -> CronJob

Parameters and example

Parameter	Type	Default	Description
`key`	`str`	Required	Stable name for the user’s schedule, 1–128 characters.
`expression`	`str`	Required	Five-field UTC cron expression.
`task`	`TaskCallable[Any] \| str`	Required	A `@task`/`@cron` callable, or a deployed `@cron` function name.
`arguments`	`Mapping[str, Any] \| None`	`None`	JSON-safe keyword arguments for the target.
`max_runs`	`int \| None`	`None`	Cancel after this many queued occurrences.
`max_consecutive_failures`	`int \| None`	`None`	Cancel after this many final failures in a row.

from harnest import cron

job = await cron.create(
    key="weekly-report",
    expression="0 9 * * 1",
    task="build_report",
)

cron decorator

Declare a cron function in cron/ with an implicit queued task. Omit the schedule for dynamic scheduling only. The filename and function name must match. Direct calls remain inline; serving registers the task.

cron(
    schedule: str | None = None,
    *,
    arguments: Mapping[str, Any] | None = None,
    timezone: str = 'UTC',
    queue: str = 'default',
    max_retries: int = 3,
) -> Callable[[_F], TaskCallable[_F]]

Parameters and example

Parameter	Type	Default	Description
`schedule`	`str \| None`	`None`	Five-field UTC cron expression. Omit for user-created schedules.
`arguments`	`Mapping[str, Any] \| None`	`None`	JSON-safe keyword arguments for every occurrence.
`timezone`	`str`	`'UTC'`	Only `"UTC"` is supported.
`queue`	`str`	`'default'`	Queue name for the generated task.
`max_retries`	`int`	`3`	Retries for one occurrence, from 0 to 100.

from harnest.cron import cron

@cron("0 9 * * 1-5")
async def daily_report() -> None:
    """Build the weekday report."""
    ...

delete

Delete one user-owned schedule.

async delete(schedule_id: str) -> bool

Parameters and example

Parameter	Type	Default	Description
`schedule_id`	`str`	Required	The job’s `cron_` ID.

from harnest import cron

removed = await cron.delete("cron_...")

get

Return one schedule owned by the active invocation user.

async get(schedule_id: str) -> CronJob | None

Parameters and example

Parameter	Type	Default	Description
`schedule_id`	`str`	Required	The job’s `cron_` ID.

from harnest import cron

job = await cron.get("cron_...")

list

Return a bounded stable page of schedules for the active user.

async list(*, after: str | None = None, limit: int = 50) -> tuple[CronJob, ...]

Parameters and example

Parameter	Type	Default	Description
`after`	`str \| None`	`None`	Last job ID from the previous page.
`limit`	`int`	`50`	Page size, from 1 to 100.

from harnest import cron

jobs = await cron.list(limit=50)

pause

Pause one user-owned schedule.

async pause(schedule_id: str) -> CronJob

Parameters and example

Parameter	Type	Default	Description
`schedule_id`	`str`	Required	The job’s `cron_` ID.

from harnest import cron

job = await cron.pause("cron_...")

resume

Resume one user-owned schedule.

async resume(schedule_id: str) -> CronJob

Parameters and example

Parameter	Type	Default	Description
`schedule_id`	`str`	Required	The job’s `cron_` ID.

from harnest import cron

job = await cron.resume("cron_...")

update

Update one user-owned schedule through the active runtime.

async update(
    schedule_id: str,
    *,
    expression: str | None = None,
    arguments: Mapping[str, Any] | object = ...,
) -> CronJob

Parameters and example

Parameter	Type	Default	Description
`schedule_id`	`str`	Required	The job’s `cron_` ID.
`expression`	`str \| None`	`None`	New expression. Omit to keep the current one.
`arguments`	`Mapping[str, Any] \| object`	`...`	New arguments; replaces the whole mapping. Omit to keep them.

from harnest import cron

job = await cron.update("cron_...", expression="0 8 * * 1")

​Declare a cron function

​Schedule parameters

​Dynamic schedule behavior

​Errors

​Exports

​CompiledCron

​CompiledCron.task_name

​Cron

​CronConflictError

​CronJob

​CronJob.update

​CronJob.pause

​CronJob.resume

​CronJob.cancel

​CronJob.delete

​CronNotFoundError

​CronRecord

​CronRuntimeError

​CronStore

​CronStore.start

​CronStore.close

​CronStore.create_cron

​CronStore.get_cron

​CronStore.list_crons

​CronStore.update_cron

​CronStore.delete_cron

​CronStore.list_due_crons

​CronStore.commit_cron_occurrence

​CronStoreConflictError

​CronUnavailableError

​cancel

​create

​cron decorator

​delete

​get

​list

​pause

​resume

​update

Declare a cron function

Schedule parameters

Dynamic schedule behavior

Errors

Exports

CompiledCron

CompiledCron.task_name

Cron

CronConflictError

CronJob

CronJob.update

CronJob.pause

CronJob.resume

CronJob.cancel

CronJob.delete

CronNotFoundError

CronRecord

CronRuntimeError

CronStore

CronStore.start

CronStore.close

CronStore.create_cron

CronStore.get_cron

CronStore.list_crons

CronStore.update_cron

CronStore.delete_cron

CronStore.list_due_crons

CronStore.commit_cron_occurrence

CronStoreConflictError

CronUnavailableError

cancel

create

cron decorator

delete

get

list

pause

resume

update