Python API¶

Everything below is importable from cereyan unless a module path is shown. This page is rendered from the docstrings in python/cereyan/ by mkdocstrings; edit the docstrings, not this page.

Decorators and the App¶

flow ¶

flow(fn: Callable | None = None, *, name: str | None = None, description: str | None = None, tags: Iterable[str] = (), group: str | None = None, run_name: str | Callable[..., str] | None = None, isolated: bool = False, log_prints: bool = False, app=None, **options)

Register a function as a flow on app or on the default App.

Use as @flow or @flow(...). Calling the returned Flow runs it: offline, the run is recorded in the local store; while a server is up, the run is handed to the server and its logs stream back. Parameters come from the function's type hints and are coerced before the body runs.

Parameters:

Name	Type	Description	Default
`fn`	`Callable`	The function to wrap; supplied by the decorator syntax.	`None`
`name`	`str \| None`	Flow name; defaults to the function name. A flow's identity is `(project, name)`.	`None`
`description`	`str \| None`	Shown in the UI; defaults to the function's docstring.	`None`
`tags`	`Iterable[str]`	Tags copied onto every run.	`()`
`group`	`str \| None`	The group this flow is listed under, inside its project, in the UI and in the `group` filter of the API, the CLI, and MCP. The same name declared in two projects is two groups. Defaults to the project, which lists the flow as one of the project's own flows.	`None`
`run_name`	`str \| Callable[..., str] \| None`	A `str.format` template over the parameters, such as `"etl-{day}"`, or a callable taking the parameters as keyword arguments and returning the name.	`None`
`isolated`	`bool`	Run every run in a fresh engine process instead of the warm pool.	`False`
`log_prints`	`bool`	Tee `print` output into the run log at INFO.	`False`
`app`	`App \| None`	The `App` to register on; defaults to the default App of the defining module.	`None`
`schedule`	`Cron \| Interval \| RRule \| Continuous \| list \| None`	A `Cron`, `Interval`, `RRule`, or `Continuous`, or a list of them; the server materialises runs from it.	required
`schedules`	`Cron \| Interval \| RRule \| Continuous \| list \| None`	Same as `schedule`; both are combined.	required
`retries`	`int`	How many times a failed run is retried.	required
`retry_delay`	`float \| list[float] \| exponential`	Wait before each retry: seconds, a list of per-attempt seconds, or `exponential`.	required
`retry_on`	`tuple[type[BaseException], ...] \| None`	Retry only these exception types; any other failure ends the run on its first attempt. Combined with `retry_when`, both must allow a retry.	required
`retry_when`	`Callable[[BaseException, int], bool] \| None`	Called with the exception and the attempt that just failed; return `False` to stop retrying. A predicate that raises stops the retry and leaves the original failure recorded.	required
`timeout_seconds`	`float \| None`	End the run as Failed with sub-state `TimedOut` after this many seconds.	required
`crash_retries`	`int \| None`	How many times a run whose engine died is rerun before it is marked Failed; defaults to `[defaults] crash_retries` in `cereyan.toml`, then 5.	required
`priority`	`int`	Dispatch order among queued runs, higher first; never preempts a running run. A negative value also raises the engine's OS niceness on Linux and macOS.	required
`max_concurrent`	`int \| None`	Cap on runs of this flow in Pending or Running at once, implemented as a resource named after the flow; unlimited by default.	required
`on_overlap`	`str`	What a new run does when `max_concurrent` is reached: `"enqueue"` (wait as AwaitingResource, the default), `"skip"` (end Skipped), `"cancel_new"` (end Cancelled), `"cancel_old"` (cancel the flow's other runs and take their slot), or `"buffer_one"` (wait, unless a run is already waiting, in which case end Skipped).	required
`start_deadline`	`float \| None`	Seconds a run may wait to start before it is skipped with reason `missed_start_deadline`; a schedule's own value wins.	required
`checkpoint`	`bool \| None`	Whether completed tasks' results are checkpointed and replayed by a later attempt. `None` (the default) replays crash reruns and resumes after `wait_for_input`; `True` also replays in-process retries; `False` never checkpoints. Replay stops at the first task whose inputs differ.	required
`checkpoint_max_bytes`	`int`	Largest encoded result that is checkpointed (default 50 MB); a larger or unpicklable result executes again on replay.	required
`unique`	`Unique \| None`	At most one run per key at a time; see `Unique`. A second submission answers with the run that holds the key (or, with `on_conflict="replace"`, cancels it), wherever the run is created.	required
`fresh_within`	`float \| timedelta \| None`	The flow's health is FAIL when its last Completed run is older than this, WARN past three quarters of it.	required
`expect_by`	`str \| None`	A cron by which a run must have completed (`"0 9 * * *"`, in `expect_by_tz`); health is FAIL when the latest deadline passed without a completed run since the previous one.	required
`expect_by_tz`	`str \| None`	IANA timezone for `expect_by`.	required
`expected_duration`	`float \| timedelta \| None`	A Running run longer than this makes the flow WARN and records `run.overdue` once.	required
`overdue_factor`	`float \| None`	As `expected_duration`, against this many times the median duration of the last 20 Completed runs.	required
`resources`	`dict[str, float] \| None`	Named resources and the amount each run holds, for example `{"db": 1}`; a run waits as AwaitingResource until they are free.	required
`after`	`str \| tuple \| list[str] \| None`	Upstream dependency: a flow name, `(name, {param: template})` to map the upstream run's parameters onto this flow's, or a list of names together with `batch_key` for fan-in.	required
`batch_key`	`str \| None`	With a list in `after`, the parameter whose value identifies a batch; this flow runs once per value after every upstream has a completed run for it.	required
`disable_after`	`tuple[int, int \| None, int] \| None`	`(count, window_seconds, persist_seconds)`: after `count` failures within `window_seconds` the flow's schedules pause for `persist_seconds` and resume automatically. A window of `None` counts failures in a row instead: a Completed run resets the count.	required
`runner`	`ThreadRunner \| ProcessRunner \| None`	A `ThreadRunner` or `ProcessRunner` executing tasks submitted with `submit` and `map`; a thread runner by default.	required
`bulk_complete`	`Callable \| None`	`bulk_complete(values) -> set` called once by a backfill; runs for the returned values are recorded as Skipped.	required
`on_completion`	`Iterable[Callable]`	Hooks `hook(flow, run, state)` called after a run completes.	required
`on_failure`	`Iterable[Callable]`	Hooks called after a run fails.	required
`on_crashed`	`Iterable[Callable]`	Hooks called after a run crashes.	required
`on_cancellation`	`Iterable[Callable]`	Hooks called after a run is cancelled.	required
`mcp_tool`	`bool`	Publish the flow as an MCP tool named `flow__<project>__<name>` whose arguments are its parameters, so an agent can start it directly.	required
`start_deadline`	`float \| None`	Seconds a run may wait to start before it is skipped.	required
`runs_on`	`str`	`"any"` (the default) lets a run execute on the server or on any remote worker whose code matches; `"server"` keeps every run on the server's machine, for flows that read or write files only the server has.	required

Returns:

Type	Description
`Flow`	The flow wrapping `fn`; call it like the original function.

Flow ¶

Flow(fn: Callable, *, name: str | None = None, description: str | None = None, tags: Iterable[str] = (), group: str | None = None, run_name: str | Callable[..., str] | None = None, isolated: bool = False, log_prints: bool = False, schedule: Any = None, schedules: Any = None, retries: int = 0, retry_delay: Any = 0, retry_on: tuple[type[BaseException], ...] | None = None, retry_when: Callable[[BaseException, int], bool] | None = None, timeout_seconds: float | None = None, crash_retries: int | None = None, priority: int = 0, max_concurrent: int | None = None, on_overlap: str = 'enqueue', resources: dict[str, float] | None = None, after: Any = None, batch_key: str | None = None, disable_after: tuple | None = None, runner: Any = None, bulk_complete: Callable | None = None, on_completion: Iterable[Callable] = (), on_failure: Iterable[Callable] = (), on_crashed: Iterable[Callable] = (), on_cancellation: Iterable[Callable] = (), mcp_tool: bool = False, start_deadline: float | None = None, checkpoint: bool | None = None, checkpoint_max_bytes: int = 50000000, unique: 'Unique | None' = None, fresh_within: 'float | timedelta | None' = None, expect_by: str | None = None, expect_by_tz: str | None = None, expected_duration: 'float | timedelta | None' = None, overdue_factor: float | None = None, runs_on: str = 'any')

A registered flow: the wrapped function plus its options, parameters, and identity.

Instances are created by flow; call one to start a run. Attributes mirror the decorator options, and parameters and schema describe the parameters derived from the function's type hints.

task ¶

task(fn: Callable | None = None, **options)

Register a function as a task: each call inside a flow becomes a task run.

Use as @task or @task(...). Outside a flow the function runs as-is. Inside a flow each call is recorded as a task run with a dynamic key such as load-0, and submit and map run calls concurrently on the flow's runner.

Parameters:

Name	Type	Description	Default
`fn`	`Callable`	The function to wrap; supplied by the decorator syntax.	`None`
`name`	`str \| None`	Task name; defaults to the function name.	required
`description`	`str \| None`	Shown in the UI; defaults to the function's docstring.	required
`tags`	`Iterable[str]`	Tags recorded on every task run.	required
`log_prints`	`bool`	Tee `print` output into the run log at INFO.	required
`retries`	`int`	How many times a failed task run is retried.	required
`retry_delay`	`float \| list[float] \| exponential`	Wait before each retry: seconds, a list of per-attempt seconds, or `exponential`.	required
`retry_on`	`tuple[type[BaseException], ...] \| None`	Retry only these exception types; any other failure ends the task run on its first attempt. Combined with `retry_when`, both must allow a retry.	required
`retry_when`	`Callable[[BaseException, int], bool] \| None`	Called with the exception and the attempt that just failed; return `False` to stop retrying. A predicate that raises stops the retry and leaves the original failure recorded.	required
`timeout_seconds`	`float \| None`	Fail the task run with sub-state `TimedOut` after this many seconds.	required
`output`	`Target \| Callable[..., Target] \| None`	A Target, or a callable over the task's arguments returning one; when the target exists the task run ends Skipped without executing.	required
`cache`	`CachePolicy \| None`	`INPUTS`, `SOURCE`, or `INPUTS + SOURCE`: reuse a persisted result when the arguments, the source, or both are unchanged. Requires `persist_result=True`.	required
`cache_expires`	`timedelta \| None`	A `timedelta` after which a cached result is stale.	required
`persist_result`	`bool`	Store the return value under `<home>/storage` so cached and replayed runs can read it.	required
`serializer`	`str`	`"pickle"` (default) or `"json"` for persisted results.	required
`resources`	`dict[str, float] \| None`	Named resources and the amount each task run holds, for example `{"gpu": 1}`.	required
`on_completion`	`Iterable[Callable]`	Hooks `hook(task, run, state)` called after a task run completes.	required
`on_failure`	`Iterable[Callable]`	Hooks called after a task run fails.	required
`on_cancellation`	`Iterable[Callable]`	Hooks called after a task run is cancelled.	required

Returns:

Type	Description
`Task`	The task wrapping `fn`.

Task ¶

Task(fn: Callable, *, name: str | None = None, description: str | None = None, tags: Iterable[str] = (), log_prints: bool = False, retries: int = 0, retry_delay: Any = 0, retry_on: tuple[type[BaseException], ...] | None = None, retry_when: Callable[[BaseException, int], bool] | None = None, timeout_seconds: float | None = None, output: Any = None, cache: CachePolicy | None = None, cache_expires: timedelta | None = None, persist_result: bool = False, serializer: str = 'pickle', resources: dict[str, float] | None = None, on_completion: Iterable[Callable] = (), on_failure: Iterable[Callable] = (), on_cancellation: Iterable[Callable] = ())

A registered task: the wrapped function plus its options.

Instances are created by task. Calling one inside a flow records a task run; submit and map return Future objects instead of waiting.

submit ¶

submit(*args, wait_for=None, **kwargs)

Run the task concurrently; returns a Future.

map ¶

map(iterable, *, wait_for=None, **static)

Submit one task run per element; returns a list of Futures.

App ¶

App(name: str | None = None, *, source_file: str | None = None)

Registry of flows and routes. Its name is the project every flow belongs to.

flow ¶

flow(fn: Callable | None = None, **options)

Decorator equivalent to @flow bound to this App.

task ¶

task(fn: Callable | None = None, **options)

Decorator equivalent to @task; tasks are not tied to an App, this exists for symmetry.

register ¶

register(flow: Flow) -> None

Add a flow to this App; raises FlowRegistrationError when another flow of the same name is registered.

route ¶

route(method: str, path: str)

Register a custom HTTP route handled by the decorated function.

Parameters:

Name	Type	Description	Default
`method`	`str`	HTTP method.	required
`path`	`str`	Path template with `{name}` placeholders, for example `"/api/ext/orders/{id}"`.	required

get ¶

get(path: str)

Register a GET route; see route.

post ¶

post(path: str)

Register a POST route; see route.

put ¶

put(path: str)

Register a PUT route; see route.

delete ¶

delete(path: str)

Register a DELETE route; see route.

patch ¶

patch(path: str)

Register a PATCH route; see route.

rule ¶

rule(on=None, flow=None, tags=None, states=None, unless=None, within=None, at=None, tz=None, **guards)

Register a code rule whose action calls the decorated fn(event, run).

A reactive rule fires when an event matching on happens. A proactive rule fires when the event named by unless does not happen: either within seconds of the on event (event-armed), or by each tick of the cron expression at in timezone tz (clock-armed, no on).

Event and state names are checked here, so a rule that could never fire raises at import rather than sitting silent. Use the constants in cereyan.events and cereyan.states to get them right the first time; a custom event name outside the reserved prefixes is always accepted. A value in states matches the run's state type or its sub-state name, so states=["Scheduled"] also covers Late and AwaitingRetry.

See cereyan.rules.register for the full argument list.

authenticator ¶

authenticator(fn: Callable) -> Callable

Register the function that validates credentials when auth is enabled.

fn(credential) receives the bearer token, or else the value of the cookie named by auth_cookie, and returns the user's name, or None to reject the credential; an exception rejects it too. It runs only when enable_auth is true, on a server thread, for each request that does not carry the static token, so cache anything slow such as a key set. One authenticator may be registered per process.

serve ¶

serve(host: str | None = None, port: int | None = None, **options) -> int

Serve the flows and routes registered so far, blocking until stopped.

Host, port, token, socket, base_path, enable_auth, auth_cookie, auth_scope, login_url, allowed_hosts (a list of host names), allow_unauthenticated, mcp_read_only, and metrics_public given here rank below the CLI flags and environment and above cereyan.toml.

ready is called with the running server once it accepts requests. Called from a thread other than the main thread, serve installs no signal handlers: call stop() on the server ready received to shut it down, and serve returns once it has.

get_default_app ¶

get_default_app(source_file: str | None = None) -> App

The default App, created on first use and named after the directory of the module that first needed it.

The current run¶

cereyan.runtime is a live view of the run executing in this thread. Each attribute is None outside a run, so a task called directly from a test reads None rather than raising.

Attribute	Type	Carries
`cereyan.runtime.run`	`RunContext`	`id`, `external_id`, `name`, `flow`, `parameters`, and `flow_name` and `project`
`cereyan.runtime.task_run`	`TaskRunContext`	`id` (the external UUID), `name`, `task_key`, `dynamic_key`
`cereyan.runtime.flow`	`Flow`	The flow of the current run

from cereyan import flow, task, runtime

@task
def record() -> str:
    return f"{runtime.run.name}/{runtime.task_run.dynamic_key}"

@flow
def report() -> str:
    return f"{runtime.flow.name}#{runtime.run.id}: {record()}"

assert runtime.run is None
assert report().endswith("/record-0")   # "report#1: <run name>/record-0"

Schedules and retry delays¶

Cron `dataclass` ¶

Cron(cron: str, timezone: str | None = None, day_or: bool = True, catchup: str = 'skip', catchup_max: int = 100, catchup_window: int | None = None, jitter: int = 0, start_deadline: int | None = None, key: str | None = None)

A cron schedule evaluated by wall clock in timezone.

to_json ¶

to_json() -> dict[str, Any]

The schedule as the server stores it.

Interval `dataclass` ¶

Interval(interval: float | timedelta, anchor: datetime | None = None, timezone: str | None = None, catchup: str = 'skip', catchup_max: int = 100, catchup_window: int | None = None, jitter: int = 0, start_deadline: int | None = None, key: str | None = None)

Fire every interval (seconds or timedelta) from anchor.

Intervals under a day are elapsed time; longer intervals keep their local wall-clock time across DST changes.

to_json ¶

to_json() -> dict[str, Any]

The schedule as the server stores it; raises ValueError for a non-positive interval.

RRule `dataclass` ¶

RRule(rrule: str, timezone: str | None = None, catchup: str = 'skip', catchup_max: int = 100, catchup_window: int | None = None, jitter: int = 0, start_deadline: int | None = None, key: str | None = None)

An iCalendar recurrence rule set; must include DTSTART.

to_json ¶

to_json() -> dict[str, Any]

The schedule as the server stores it.

Continuous `dataclass` ¶

Continuous(delay: float | int | timedelta | str = 0, jitter: int = 0, start_deadline: int | None = None, key: str | None = None)

Run again delay after the last run ends, once a processor is free.

delay is seconds, a timedelta, or a string such as "30m". At most one run of the schedule is ever waiting, in line, or running; the wait holds no processor. A continuous schedule has no fire times, so it takes no catch-up options. Pause the schedule to stop the loop; disable_after on the flow stops it after repeated failures and resumes it later.

to_json ¶

to_json() -> dict[str, Any]

The schedule as the server stores it; raises ValueError for a negative or unreadable delay.

exponential `dataclass` ¶

exponential(base: float = 1.0, jitter: float = 0.0, maximum: float = 3600.0)

Retry delay growing as base * 2**attempt with optional jitter.

delay ¶

delay(attempt: int) -> float

Seconds to wait before retry number attempt (0-based), capped at maximum.

Targets, caching and results¶

Target ¶

Bases: Protocol

Anything with exists(): the protocol a task's output= must satisfy.

exists ¶

exists() -> bool

True when the output this target stands for is already present.

LocalTarget ¶

LocalTarget(path: str | PathLike, *, mkdir: bool = True)

A file on the local filesystem written atomically.

exists ¶

exists() -> bool

True when the file exists.

open ¶

open(mode: str = 'r', **kwargs: Any)

Open the file; write modes write to a temporary path that replaces the target on close, so readers never see a partial file.

remove ¶

remove() -> None

Delete the file if it exists.

temporary_path ¶

temporary_path()

Yield a temporary path; on success it is renamed onto the target.

CachePolicy ¶

Bases: Flag

INPUTS and SOURCE are the members of CachePolicy, exported at the top level so cache=INPUTS + SOURCE reads naturally.

Concurrency¶

Future ¶

Future(task, external_id: str, dynamic_key: str)

Handle to a submitted task run.

done ¶

done() -> bool

True once the task run has finished, in any state.

wait ¶

wait(timeout: float | None = None) -> None

Block until the task run finishes or timeout seconds pass.

result ¶

result(timeout: float | None = None) -> Any

The task's return value, blocking until it is available; re-raises the task's exception on failure.

exception ¶

exception(timeout: float | None = None)

The exception the task raised, or None; blocks like result.

ThreadRunner ¶

ThreadRunner(max_workers: int | None = None)

Bases: BaseRunner

Runs submitted tasks on a bounded set of threads with the run context propagated. A task that waits on a child while every worker is busy gets a warning and a temporary extra worker so the wait cannot deadlock.

ProcessRunner ¶

ProcessRunner(max_workers: int | None = None)

Bases: BaseRunner

Runs each submitted task in a spawned process; arguments and results must be picklable and the task function importable. Timeouts terminate the worker.

Logging, events and artifacts¶

get_run_logger ¶

get_run_logger() -> Logger

A logger that carries the current run and task-run in its records.

emit_event ¶

emit_event(name: str, payload: dict | None = None, resource: dict | None = None) -> None

Record an event. Inside a run the resource defaults to that run; outside a run the event goes straight to the store or, when a server is up, to its API.

Any name is allowed except one under a reserved prefix (see cereyan.events.RESERVED_PREFIXES) that the engine does not emit, which is always a mistake rather than a custom event.

Raises:

Type	Description
`ValueError`	When the name is empty, not a string, or shadows the engine's own namespace.

EventName ¶

Bases: str

An engine event name; a str carrying the exact wire name.

Attributes:

Name	Type	Description
`resource`	`str`	The resource kind the event hangs off (`run`, `task_run`, `flow`, `schedule`, `resource`, `rule`).
`when`	`str`	One sentence on when the engine records it.
`payload_fields`	`tuple[str, ...]`	The payload keys the emit site sets.

EventGroup ¶

EventGroup(prefix: str, names: 'list[EventName]')

The events under one reserved prefix, as attributes.

events.run.failed is "run.failed" and events.run.any is "run.*", which matches every event under the prefix.

StateName ¶

Bases: str

A state type or named sub-state; a str carrying the wire name.

Attributes:

Name	Type	Description
`state_type`	`str`	For a sub-state, the type it belongs to; for a state type, itself.
`is_sub_state`	`bool`	`True` for a named sub-state such as `Late`.

create_markdown ¶

create_markdown(text: str, key: str | None = None) -> str

Attach a markdown artifact to the current run or task run.

Parameters:

Name	Type	Description	Default
`text`	`str`	Markdown source, rendered on the run page.	required
`key`	`str \| None`	Optional stable key; artifacts sharing a key form a history across runs.	`None`

Returns:

Type	Description
`str`	The artifact id.

Raises:

Type	Description
`CereyanError`	Outside a run, or when the artifact exceeds 1 MB.

create_table ¶

create_table(rows: list[dict] | list[list], key: str | None = None, columns: list[str] | None = None) -> str

Attach a table artifact.

Parameters:

Name	Type	Description	Default
`rows`	`list[dict] \| list[list]`	A list of dicts (columns inferred from the keys, in first-seen order) or a list of lists (then pass `columns`).	required
`key`	`str \| None`	Optional stable key.	`None`
`columns`	`list[str] \| None`	Column names; required for list-of-lists rows.	`None`

Returns:

Type	Description
`str`	The artifact id.

create_progress ¶

create_progress(percent: float, key: str | None = None, label: str | None = None) -> str

Attach a progress bar artifact.

Parameters:

Name	Type	Description	Default
`percent`	`float`	Completion from 0 to 100; values outside are clamped.	required
`key`	`str \| None`	Optional stable key; pass the same key to `update_progress` later.	`None`
`label`	`str \| None`	Text shown next to the bar.	`None`

Returns:

Type	Description
`str`	The artifact id.

update_progress ¶

update_progress(key: str, percent: float, label: str | None = None) -> str

Record a new value for the progress artifact with key.

Each update is a new artifact row under the same key, so the history is kept.

Parameters:

Name	Type	Description	Default
`key`	`str`	The key given when the progress artifact was created.	required
`percent`	`float`	Completion from 0 to 100.	required
`label`	`str \| None`	Text shown next to the bar.	`None`

Returns:

Type	Description
`str`	The artifact id.

create_link ¶

create_link(url: str, text: str | None = None, key: str | None = None) -> str

Attach a link artifact.

Parameters:

Name	Type	Description	Default
`url`	`str`	The link target.	required
`text`	`str \| None`	Link text; defaults to the URL.	`None`
`key`	`str \| None`	Optional stable key.	`None`

Returns:

Type	Description
`str`	The artifact id.

create_image ¶

create_image(url_or_bytes: str | bytes, key: str | None = None, media_type: str = 'image/png') -> str

Attach an image artifact from a URL or raw bytes.

Parameters:

Name	Type	Description	Default
`url_or_bytes`	`str \| bytes`	An image URL, or the image bytes to embed as a data URI.	required
`key`	`str \| None`	Optional stable key.	`None`
`media_type`	`str`	MIME type used when bytes are given.	`'image/png'`

Returns:

Type	Description
`str`	The artifact id.

Variables¶

Variable ¶

Named JSON values shared by every project on the machine, optionally encrypted.

Names match [a-z0-9][a-z0-9_./-]* and values are limited to 64 KB. Offline, values go straight to the store; while a server holds the store, they go through its API. Secrets are encrypted with the key kept at <home>/secret.key and are masked in the API and the UI.

get `staticmethod` ¶

get(name: str, default: Any = None) -> Any

Read a variable.

Parameters:

Name	Type	Description	Default
`name`	`str`	The variable name.	required
`default`	`Any`	Returned when the variable does not exist.	`None`

Returns:

Type	Description
`Any`	The stored value, decrypted for secrets.

set `staticmethod` ¶

set(name: str, value: Any, tags: list[str] | None = None, secret: bool = False, overwrite: bool = True) -> None

Create or update a variable.

Parameters:

Name	Type	Description	Default
`name`	`str`	The variable name.	required
`value`	`Any`	Any JSON-serialisable value (dates and dataclasses are converted).	required
`tags`	`list[str] \| None`	Tags shown on the Variables page.	`None`
`secret`	`bool`	Encrypt the value at rest and mask it in the API and the UI.	`False`
`overwrite`	`bool`	When `False`, raise if the variable already exists.	`True`

Raises:

Type	Description
`ValueError`	On an invalid name or a value over 64 KB.
`CereyanError`	When `overwrite` is `False` and the variable exists.

unset `staticmethod` ¶

unset(name: str) -> bool

Delete a variable.

Parameters:

Name	Type	Description	Default
`name`	`str`	The variable name.	required

Returns:

Type	Description
`bool`	`True` when a variable was deleted, `False` when none existed.

Messages and human input¶

receive ¶

receive(topic: str, *, prompt: str | None = None, schema: dict | None = None, timeout: float | None = None, default: Any = None) -> Any

Return the next message on topic, pausing the run when none is available.

timeout seconds: resume with default if no message arrives in time. The timer is server-side, so the run stays Paused and frees its engine.

Inside a served run the first call transitions the run to Paused with topic in its state details and ends the attempt; the engine is free while the run waits. POST /api/runs/{id}/messages/{topic} stores a message and resumes the run, which reruns the flow from the top and gets the message from this call.

Calls are numbered in the order the body reaches them, and a message belongs to the call it answered, so a flow can receive, resume, and receive again. A message is used only when the topic at that position still matches the one it was given for.

wait_for_input(prompt, schema) is receive('input', prompt=prompt, schema=schema).

Outside a served run topic='input' reads the answer from the terminal; other topics raise an error. Timeouts are not supported offline.

Raises:

Type	Description
`CereyanError`	When called outside the thread executing the flow body.

wait_for_input ¶

wait_for_input(prompt: str, schema: dict | None = None) -> Any

Return the answer given to this question, pausing the run when there is none.

This is receive('input', prompt=prompt, schema=schema); see :func:receive for full semantics.

publish_state ¶

publish_state(key: str, value: Any) -> None

Store value (any JSON, up to 64 KB) under key for this run, readable with GET /api/runs/{id}/state outside the flow.

This is flow-level state: it is not scoped to the current task, and it survives retries and resumes.

Durable waits¶

sleep ¶

sleep(seconds: float, *, min_durable: float = DEFAULT_MIN_DURABLE) -> None

Wait seconds. Inside a served run a wait of min_durable seconds or more pauses the run as Sleeping and frees the engine until the wake time; shorter waits, and every wait offline, block in process.

Raises:

Type	Description
`CereyanError`	From a task on another thread than the flow body.

sleep_until ¶

sleep_until(when: datetime, *, min_durable: float = DEFAULT_MIN_DURABLE) -> None

Wait until when (a naive datetime is UTC); as sleep, durable from min_durable seconds away.

wait_for_event ¶

wait_for_event(name: str, match: dict | None = None, within: float | None = None) -> dict

Wait for an event named name (exact, or prefix.*) whose payload carries every key of match with an equal value, and return it.

Inside a served run the run pauses as AwaitingEvent and the server wakes it when such an event is recorded. within seconds bound the wait; past them the call raises WaitTimeout. Offline the local store is polled for within seconds, which is required there.

Raises:

Type	Description
`WaitTimeout`	When `within` passes first.
`CereyanError`	Offline without `within`, or from a task thread.

wait_for_target ¶

wait_for_target(target, poke: float = 60.0, timeout: float | None = None)

Wait until target.exists() and return the target.

Inside a served run the run pauses as AwaitingTarget and is poked every poke seconds: each poke replays the body to this call, which checks again. timeout seconds bound the wait, after which the call raises WaitTimeout. Offline the target is polled in process.

Snooze ¶

Snooze(seconds: float, reason: str | None = None)

Bases: BaseException

Raise from a flow or a task to end the attempt without a failure and run the body again after seconds; the run shows Sleeping meanwhile and details.snoozes counts how often it did this.

WaitTimeout ¶

Bases: CereyanError

A durable wait ended without what it waited for: wait_for_event's within or wait_for_target's timeout passed.

Task state¶

get ¶

get(key: str, default: Any = None) -> Any

The value stored under key for this task in this run, or in the run it continues (a crash rerun or a retry from failure), else default.

set ¶

set(key: str, value: Any) -> None

Store value (any JSON, up to 64 KB) under key for this task in this run.

Raises:

Type	Description
`CereyanError`	Outside a run, for a bad key, or when the value is too large.

delete ¶

delete(key: str) -> bool

Remove key from this task's entries in this run; returns whether it existed.

items ¶

items() -> dict[str, Any]

Every entry this task stored in this run, as a dict.

Unique runs¶

Unique ¶

Unique(key: str | None = None, period: float | None = None, states: Iterable[str] = (), on_conflict: str = 'skip', debounce: float | None = None, max_wait: float | None = None)

At most one run of the flow per key at a time: @flow(unique=Unique(...)).

Parameters:

Name	Type	Description	Default
`key`	`str \| None`	Template over the flow's parameters, `"{day}"`; every parameter when `None`. Two submissions render the same key when they name the same run.	`None`
`period`	`float \| None`	Seconds; fixed windows bucket the key, so `period=3600` allows one run per key per clock hour.	`None`
`states`	`Iterable[str]`	State types in which an existing run counts, default every non-terminal one. Add `"Completed"` to keep a key taken after the run ends.	`()`
`on_conflict`	`str`	`"skip"` (the default) answers with the run that holds the key and creates nothing; `"replace"` cancels it and creates the new run; `"debounce"` creates the run `debounce` seconds ahead and, while it has not started, every further submission moves its start to now plus `debounce` (at most `max_wait` after its creation) and replaces its parameters; `"throttle"` keeps the first run in any sliding window of `period` seconds and answers the others with it.	`'skip'`
`debounce`	`float \| None`	Seconds a debounced run waits after the last submission.	`None`
`max_wait`	`float \| None`	The longest a debounced run may be pushed back from its creation.	`None`

spec ¶

spec() -> dict[str, Any]

The declaration as the server records it in the flow's options.

Client¶

The module-level functions use the server recorded in server.json of the runtime home; Client targets any server explicitly.

run ¶

run(flow: str, project: str | None = None, **parameters: Any) -> dict

Create a run of a registered flow on the live server and return it.

Parameters:

Name	Type	Description	Default
`flow`	`str`	Flow name.	required
`project`	`str \| None`	Project name; required when the name exists in several projects.	`None`
`**parameters`	`Any`	Flow parameters.	`{}`

Raises:

Type	Description
`ServerUnavailable`	When no server is running for the home.
`CereyanError`	When the flow is unknown or ambiguous.

get_run ¶

get_run(run_id: int) -> dict

GET /api/runs/{id} on the live server.

list_runs ¶

list_runs(**filters: Any) -> list[dict]

The items of GET /api/runs on the live server with the given filters.

list_flows ¶

list_flows(project: str | None = None) -> list[dict]

GET /api/flows on the live server, optionally filtered by project.

cancel ¶

cancel(run_id: int) -> dict

POST /api/runs/{id}/cancel on the live server.

Client ¶

Client(base_url: str = '', timeout: float = 30.0, token: str | None = None, socket_path: str | None = None)

HTTP client for a running server, on the standard library only.

Parameters:

Name	Type	Description	Default
`base_url`	`str`	The server URL, for example `http://127.0.0.1:4200`.	`''`
`timeout`	`float`	Seconds to wait for a response.	`30.0`
`token`	`str \| None`	API token; defaults to `CEREYAN_TOKEN`.	`None`
`socket_path`	`str \| None`	Connect over this Unix socket instead of TCP.	`None`

Most methods return the decoded JSON of the matching endpoint (see the HTTP API reference) and raise ApiError on an error status or ServerUnavailable when the server cannot be reached.

health ¶

health() -> bool

True when GET /api/health answers ok; never raises.

server ¶

server() -> dict

GET /api/server: version, home, engines, auth state, and listeners.

flows ¶

flows(project: str | None = None, group: str | None = None) -> list[dict]

GET /api/flows, optionally filtered by project and by group.

group matches the resolved group: the one a flow declared, else its project, so a project name selects that project's undeclared flows.

flow ¶

flow(flow_id: int) -> dict

GET /api/flows/{id}: one flow with its schedules and summary.

runs ¶

runs(**filters: Any) -> dict

GET /api/runs with the given filters (flow, project, group, state_type, tags, limit, cursor, ...); returns the page.

get_run ¶

get_run(run_id: int) -> dict

GET /api/runs/{id}: one run with its state and parameters.

task_runs ¶

task_runs(run_id: int) -> list[dict]

GET /api/runs/{id}/tasks: the run's task runs.

logs ¶

logs(run_id: int, after: int = 0, level: str | None = None, search: str | None = None, limit: int = 1000) -> dict

GET /api/runs/{id}/logs after sequence after, optionally filtered by level and search.

resume ¶

resume(run_id: int, input: Any) -> dict

POST /api/runs/{id}/resume: answer a Paused run with input and schedule its next attempt.

send_message ¶

send_message(run_id: int, topic: str, payload: Any) -> dict

POST /api/runs/{id}/messages/{topic}: send a message to a run on topic.

run_state ¶

run_state(run_id: int, scope: str | None = None, key: str | None = None) -> Any

GET /api/runs/{id}/state: the values a run stored with publish_state and task_state.

Without key, a list of {scope, key, value}; scope is "" for the flow body and a task's dynamic key otherwise. With key (and scope, default the flow body), {found, value} for that one entry.

cancel ¶

cancel(run_id: int) -> dict

POST /api/runs/{id}/cancel: cancel a queued run at once or ask a running one to stop.

delete_run ¶

delete_run(run_id: int) -> None

DELETE /api/runs/{id}: remove a finished run and its task runs, logs, and artifacts.

counts ¶

counts(project: str | None = None) -> dict

GET /api/counts: run counts by state for the dashboard, optionally per project.

backfill ¶

backfill(flow_id: int, parameter: str, start: str | None = None, end: str | None = None, *, interval: Any = '1d', concurrency: int = 1, extra_parameters: dict | None = None, reverse: bool = False, values: list[str] | None = None, missing_only: bool = False, force: bool = False) -> dict

POST /api/flows/{id}/backfill: create one run per step of parameter from start to end, or one per entry of values.

Parameters:

Name	Type	Description	Default
`flow_id`	`int`	The flow's id.	required
`parameter`	`str`	The date or datetime parameter to step.	required
`start`	`str \| None`	First value, ISO formatted; not needed with `values`.	`None`
`end`	`str \| None`	Last value, inclusive.	`None`
`interval`	`Any`	Step as seconds or a duration such as `"1d"` or `"12h"`.	`'1d'`
`concurrency`	`int`	How many of the backfill's runs may execute at once.	`1`
`extra_parameters`	`dict \| None`	Fixed values for other parameters.	`None`
`reverse`	`bool`	Create the newest value first.	`False`
`values`	`list[str] \| None`	Explicit parameter values instead of a range, in this order.	`None`
`missing_only`	`bool`	Leave out values whose latest run completed.	`False`
`force`	`bool`	A restatement: the runs ignore targets, caches, checkpoints and `bulk_complete`.	`False`

Returns:

Type	Description
`dict`	The backfill status with its id, tag, and counts.

backfill_status ¶

backfill_status(backfill_id: int) -> dict

GET /api/backfills/{id}: counts of the backfill's runs by state.

cancel_backfill ¶

cancel_backfill(backfill_id: int) -> dict

POST /api/backfills/{id}/cancel: cancel the backfill's remaining runs.

schedules ¶

schedules(flow_id: int) -> list[dict]

GET /api/flows/{id}/schedules: the flow's schedules with their next fire times.

upcoming ¶

upcoming(flow_id: int) -> list[dict]

GET /api/flows/{id}/upcoming: the runs materialised ahead for the flow.

settings ¶

settings() -> dict

GET /api/settings: resource totals, retention, and defaults.

events ¶

events(kind: str | None = None, after: int = 0, limit: int = 100, **filters: Any) -> list[dict]

GET /api/events newest first, filtered by name prefix kind and other filters; with after returns events following that sequence number in ascending order.

rules ¶

rules() -> list[dict]

GET /api/rules: every rule with its match clause, actions, guards, and fire counts.

create_rule ¶

create_rule(body: dict) -> dict

POST /api/rules: create a data rule from a rule body.

variables ¶

variables() -> list[dict]

GET /api/variables: every variable with secrets masked.

artifacts ¶

artifacts(run_id: int) -> list[dict]

GET /api/runs/{id}/artifacts: the run's artifacts.

submit ¶

submit(project: str, flow: str, parameters: dict | None = None, *, name: str | None = None, tags: list[str] | None = None, module: str | None = None, source_dir: str | None = None, description: str | None = None, parameter_schema: dict | None = None, options: dict | None = None, flow_tags: list[str] | None = None, flow_group: str | None = None, created_by: str = 'client', scheduled_time: int | None = None, delay: float | None = None, idempotency_key: str | None = None, idempotency_ttl: float | None = None) -> dict

POST /api/runs: create a run with explicit project and flow, registering the flow when needed.

This is the low-level call used by the offline handoff and by run; pass module and source_dir so the server can import a flow it has not seen. scheduled_time (microseconds) or delay (seconds) creates the run for later: it waits as Scheduled and starts at that time. idempotency_key makes a repeated call within idempotency_ttl seconds (default a day) answer with the run first created; the returned run then carries conflict: True, as it does when the flow's own unique key is held.

run ¶

run(flow: str, project: str | None = None, *, name: str | None = None, tags: list[str] | None = None, at: 'datetime | str | None' = None, delay: float | None = None, idempotency_key: str | None = None, idempotency_ttl: float | None = None, **parameters: Any) -> dict

Create a run of a registered flow by name. Returns the run.

at (a datetime, naive meaning UTC, or an ISO 8601 string) or delay (seconds) creates the run for later instead of now.

find_server ¶

find_server(home: str | None = None) -> Client | None

A client for the live server of home, or None when there is none; raises AuthRequired when a token is needed.

Custom routes¶

Request ¶

Request(method: str, path: str, path_params: dict[str, str], query: dict[str, list[str]], headers: dict[str, str], body: bytes)

The raw request handed to a handler parameter annotated Request.

json ¶

json() -> Any

The request body decoded as JSON, or None when empty.

text ¶

text() -> str

The request body decoded as UTF-8 text.

Response ¶

Response(body: bytes | str = b'', status: int = 200, headers: dict[str, str] | None = None, media_type: str | None = None)

An explicit response: body bytes or str, status, headers.

HTTPError ¶

HTTPError(status: int, message: str = '')

Bases: CereyanError

Raise inside a handler to send a status with a JSON error body.

Errors¶

CereyanError ¶

Bases: Exception

Base class for errors raised by cereyan.

ParameterError ¶

ParameterError(name: str, expected: str, value: object)

Bases: CereyanError, ValueError

A parameter value could not be coerced to its declared type.

StoreLocked ¶

Bases: RuntimeError

The store is locked by another process.

TransitionRejected ¶

Bases: RuntimeError

A state transition was rejected by the rules.

ServerUnavailable ¶

Bases: CereyanError

No live server was found for the home directory.

ApiError ¶

ApiError(status: int, body: Any)

Bases: CereyanError

The server answered with an error status; status and the decoded body are kept.

AuthRequired ¶

Bases: CereyanError

The server requires a token the client does not have (or a wrong one).

Python API¶

Decorators and the App¶

flow ¶

Flow ¶

task ¶

Task ¶

submit ¶

map ¶

App ¶

flow ¶

task ¶

register ¶

route ¶

get ¶

post ¶

put ¶

delete ¶

patch ¶

rule ¶

authenticator ¶

serve ¶

get_default_app ¶

The current run¶

Schedules and retry delays¶

Cron dataclass ¶

to_json ¶

Interval dataclass ¶

to_json ¶

RRule dataclass ¶

to_json ¶

Continuous dataclass ¶

to_json ¶

exponential dataclass ¶

delay ¶

Targets, caching and results¶

Target ¶

exists ¶

LocalTarget ¶

exists ¶

open ¶

remove ¶

temporary_path ¶

CachePolicy ¶

Concurrency¶

Future ¶

done ¶

wait ¶

result ¶

exception ¶

ThreadRunner ¶

ProcessRunner ¶

Logging, events and artifacts¶

get_run_logger ¶

emit_event ¶

EventName ¶

EventGroup ¶

StateName ¶

create_markdown ¶

create_table ¶

create_progress ¶

update_progress ¶

create_link ¶

create_image ¶

Variables¶

Variable ¶

get staticmethod ¶

set staticmethod ¶

unset staticmethod ¶

Messages and human input¶

receive ¶

wait_for_input ¶

publish_state ¶

Durable waits¶

sleep ¶

sleep_until ¶

wait_for_event ¶

wait_for_target ¶

Snooze ¶

WaitTimeout ¶

Task state¶

Cron `dataclass` ¶

Interval `dataclass` ¶

RRule `dataclass` ¶

Continuous `dataclass` ¶

exponential `dataclass` ¶

get `staticmethod` ¶

set `staticmethod` ¶

unset `staticmethod` ¶