Python adapter reference

All exports are available from the coflux package (import coflux as cf).

Decorators

Both @workflow and @task accept async def functions in addition to regular functions. The coroutine is run to completion by the executor, and the target's return type is the coroutine's resolved value.

@workflow

Defines a workflow — the entry point for a run.

@cf.workflow(
    name: str | None = None,
    wait: bool | Iterable[str] | str = False,
    cache: bool | float | timedelta | Cache = False,
    retries: int | bool | Retries = 0,
    recurrent: bool = False,
    defer: bool | Defer = False,
    delay: float | timedelta = 0,
    memo: bool = False,
    requires: dict[str, str | bool | list[str]] | None = None,
    timeout: float | timedelta = 0,
    streams: Streams | None = None,
    concurrency: int | Concurrency = 0,
)

Parameter	Type	Default	Description
name	str | None	None	Custom target name (defaults to function name)
wait	bool | Iterable[str] | str	False	Wait for Execution arguments to resolve before starting
cache	bool | float | timedelta | Cache	False	Caching configuration
retries	int | bool | Retries	0	Retry configuration
recurrent	bool	False	Re-execute after completion (recurrent)
defer	bool | Defer	False	Deferring configuration
delay	float | timedelta	0	Delay before execution (seconds)
memo	bool	False	Enable memoisation as default for all tasks in the run
requires	dict | None	None	Tag requirements for worker routing (applied to entire run)
timeout	float | timedelta	0	Execution timeout in seconds (0 = no timeout)
streams	Streams | None	None	Default stream configuration (buffer, timeout) for streams the target produces: a generator body's stream, or any registered with cf.stream()
concurrency	int | Concurrency	0	Concurrency limit — how many executions may run at once (0 = no limit)

@task

Defines a task — an operation that can be called from a workflow or another task. Tasks are the building blocks of a workflow and each invocation becomes a step in the run.

@cf.task(
    name: str | None = None,
    wait: bool | Iterable[str] | str = False,
    cache: bool | float | timedelta | Cache = False,
    retries: int | bool | Retries = 0,
    recurrent: bool = False,
    defer: bool | Defer = False,
    delay: float | timedelta = 0,
    memo: bool | Iterable[str] = False,
    requires: dict[str, str | bool | list[str]] | None = None,
    timeout: float | timedelta = 0,
    streams: Streams | None = None,
    concurrency: int | Concurrency = 0,
)

Parameters are the same as @workflow, except:

Parameter	Difference
memo	Also accepts an iterable of parameter names to memo on (e.g., memo=["user_id"])

@stub

References a target defined in another module, without importing it. This allows modules with different dependencies to be hosted on different workers. The stub's function body is used as a fallback when called outside of a Coflux context (e.g., in tests).

@cf.stub(
    module: str,
    *,
    name: str | None = None,
    type: Literal["workflow", "task"] = "task",
    wait: bool | Iterable[str] | str = False,
    cache: bool | float | timedelta | Cache = False,
    retries: int | bool | Retries = 0,
    recurrent: bool = False,
    defer: bool | Defer = False,
    delay: float | timedelta = 0,
    memo: bool | Iterable[str] = False,
    concurrency: int | Concurrency = 0,
)

Parameter	Type	Default	Description
module	str	required	Module where the target is defined
type	Literal["workflow", "task"]	"task"	Whether referencing a workflow or task

Other parameters behave the same as @task. requires and timeout are not available on stubs.

Target

Decorating a function with @workflow, @task, or @stub returns a Target object. The target can be called directly (blocking) or submitted for asynchronous execution.

target(…) -> T

Calls the target synchronously — submits for execution and blocks until the result is available. Equivalent to target.submit(…).result().

target.submit(…) -> Execution[T]

Submits the target for asynchronous execution and returns an Execution handle. The caller can continue other work and retrieve the result later.

Per-call-site overrides

Each with_* method returns a new Target with the corresponding decorator-level option overridden, leaving the original target unchanged. This is useful for one-off variations without re-decorating, and the methods can be chained:

my_task.with_retries(3).with_timeout(30).submit(x)
my_task.with_cache(False).submit(x)  # disable caching for this call

cached = my_task.with_cache(60)  # stash a configured variant
cached.submit(a)
cached.submit(b)

Method	Description
with_cache(cache)	Override caching. Pass False to disable.
with_retries(retries)	Override retries. Pass 0 or False to disable.
with_defer(defer)	Override defer configuration.
with_concurrency(concurrency)	Override the concurrency limit. Pass 0 to disable.
with_memo(memo)	Override memoisation configuration.
with_delay(delay)	Override submission delay (seconds or timedelta).
with_timeout(timeout)	Override execution timeout.
with_requires(requires)	Override worker routing tags.
with_streams(streams)	Override the default stream configuration for streams the target produces.

Execution

Returned by target.submit(). Represents a running or completed execution, and acts as a future for its result. Can be passed as an argument to other tasks, or returned from a task/workflow.

Properties

Property	Type	Description
id	str	Execution ID
module	str	Module name
target	str	Target name

execution.result() -> T

Blocks (suspends) until the execution completes and returns the result. If the execution failed, raises the corresponding exception.

Raises: ExecutionError if the execution failed, or an ExecutionTerminated subclass (ExecutionCancelled, ExecutionTimeout, ExecutionAbandoned, ExecutionCrashed) if it ended without completing. If called inside a cf.suspense(timeout=...) scope and the timeout expires before the handle resolves, raises TimeoutError.

execution.poll(timeout=None, *, default=None) -> T | D

Checks whether a result is ready without suspending the caller. Returns the result if available, otherwise returns default. Useful for coordination patterns where you want to check multiple executions or do other work while waiting.

Parameter	Type	Default	Description
timeout	float | None	None	Seconds to wait before returning default
default	D	None	Value to return if result isn't ready

execution.cancel() -> None

Cancels the execution and its descendants.

Input

Returned by prompt.submit(...). Represents a requested input, and acts as a future for the response. Like Execution, it can be passed to other tasks and only carries an ID across the wire.

Properties

Property	Type	Description
id	str	Input ID

input.result() -> T

Blocks (suspends) until the input is responded to and returns the value. If the prompt was created with a model, the value is parsed into that type.

Raises: InputDismissed if the prompt was dismissed; ExecutionCancelled if the input was cancelled. Raises TimeoutError if called inside a cf.suspense(timeout=...) scope and the timeout expires before a response arrives.

input.poll(timeout=None, *, default=None) -> T | D

Non-suspending check for a response. Returns the value if available, or default otherwise.

input.cancel() -> None

Cancels the input, transitioning it to a terminal cancelled state (distinct from dismissed).

Prompt

Defines a prompt for requesting input from a user. Submit it to get an Input handle.

cf.Prompt(
    template: str,
    model: type[T] | None = None,
    *,
    title: str | None = None,
    actions: tuple[str, str] | None = None,
    schema: str | dict | None = None,
    requires: dict[str, str | bool | list[str]] | None = None,
)

Parameter	Type	Default	Description
template	str	required	Message template. May contain {placeholders} substituted at submission.
model	type | None	None	Pydantic model (or primitive type) used to render a form and validate the response. Without a model, the prompt is approval-only.
title	str | None	None	Short title shown above the prompt.
actions	tuple[str, str] | None	None	Custom labels for the respond/dismiss buttons, as (respond, dismiss).
schema	str | dict | None	None	Raw JSON schema (alternative to model).
requires	dict | None	None	Routing tags for matching to responders.

prompt(**placeholders) -> T

Submits the prompt and blocks until a response is available. Equivalent to prompt.submit(...).result().

prompt.submit(**placeholders) -> Input[T]

Submits the prompt and returns an Input handle without blocking.

Per-submission overrides

Each with_* method returns a new Prompt with overrides applied:

Method	Description
with_key(key)	Use an explicit memoisation key (per-run).
with_initial(value)	Pre-populate the form with an initial value (e.g. a Pydantic instance).
with_actions(respond, dismiss)	Override button labels.
with_requires(requires)	Override routing tags.

Configuration classes

These are used as values for decorator parameters when more control is needed than the shorthand forms allow.

Cache

Advanced caching configuration. See caching.

cf.Cache(
    max_age: float | timedelta | None = None,
    params: Iterable[str] | str | None = None,
    namespace: str | None = None,
    version: str | None = None,
)

Parameter	Type	Default	Description
max_age	float | timedelta | None	None	Maximum age of cached result
params	Iterable[str] | str | None	None	Parameters to include in cache key (None = all)
namespace	str | None	None	Cache namespace
version	str | None	None	Cache version (change to invalidate)

Defer

Advanced deferring configuration. See deferring.

cf.Defer(
    params: Iterable[str] | str | None = None,
)

Parameter	Type	Default	Description
params	Iterable[str] | str | None	None	Parameters to defer on (None = all)

Concurrency

Advanced concurrency limit configuration. See concurrency limits.

cf.Concurrency(
    limit: int,
    params: bool | Iterable[str] | str = False,
    namespace: str | None = None,
)

Parameter	Type	Default	Description
limit	int	required	Executions allowed at once (must be >= 1)
params	bool | Iterable[str] | str	False	Parameters the limit is keyed on (False = the target as a whole, True = all)
namespace	str | None	None	Pool to draw the limit from (defaults to "{module}:{target}")

Retries

Advanced retry configuration. See retries.

cf.Retries(
    limit: int | None = None,
    backoff: tuple[float | timedelta, float | timedelta] = (1, 60),
    when: type[BaseException] | tuple[type[BaseException], ...] | Callable[[BaseException], bool] | None = None,
)

Parameter	Type	Default	Description
limit	int | None	None	Maximum retries (None = unlimited)
backoff	tuple	(1, 60)	Backoff range (min, max) in seconds
when	type, tuple, callable, or None	None	Exception filter (None = retry on any error)

Streams

Default stream configuration for a target, passed as streams= to @task / @workflow. Only applies to targets that produce streams.

cf.Streams(
    buffer: int | None = 0,
    timeout: float | timedelta | None = None,
)

Parameter	Type	Default	Description
buffer	int | None	0	How many items the producer may run ahead of its slowest consumer (0 = lockstep, None = no backpressure)
timeout	float | timedelta | None	None	Idle timeout: the stream is closed if no item is appended within this window

Streams

Ordered sequences of values produced by one execution and consumed by others as they grow. See streams.

stream(generator, *, buffer=..., timeout=...)

Registers a generator as a stream and returns a Stream handle to embed in a return value or pass to another task. A task whose body is itself a generator is registered automatically, with its result being the handle. Unspecified options inherit from the target's Streams configuration. Must be called inside a task or workflow body.

Stream

A handle to a stream, typed by its items (Stream[T]).

Method	Description
for item in stream / async for	Iterate from the first item, blocking until each arrives, ending when the stream closes
stream.slice(start, stop=None)	A view of positions [start, stop)
stream.partition(n, i)	A view of every n-th item starting at i, for parallel consumers
stream.stride(start=0, stop=None, step=1)	The general form of the above
stream.id	An opaque identifier for the stream, as shown in Studio

Views compose, and can be passed to other tasks. If the producer raised, iterating raises the same error; if it was cancelled, abandoned, crashed or timed out, iterating raises the corresponding ExecutionTerminated subclass; if it was a recurrent target that finished its iteration, StreamSuperseded.

iter(stream) returns a StreamIterator and aiter(stream) an AsyncStreamIterator. Both can be released early with close() / aclose(), or by using them as a context manager (with / async with). An open subscription holds a bounded producer's backpressure, so release one you stop reading before the stream ends; dropping the last reference releases it too.

Checkpoints

State that survives across executions of a step — retries, suspensions, recurrences and re-runs. See checkpoints.

Checkpoint

cf.Checkpoint(
    name: str,
    *,
    default: T,
)
cf.Checkpoint[T | None](
    name: str,
)

Parameter	Type	Default	Description
name	str	required	Checkpoint name, unique within the step. Can't start with _, which is reserved for adapter-managed state
default	T	—	Returned when the checkpoint is unset or has been reset. Omit it and get() may return None. With a declared type, validated as one here

T is the type get() returns. It's inferred from default when one is given, so cf.Checkpoint("cursor", default=0) is a Checkpoint[int]. Without a default the checkpoint can read as None, so spell the type out: cf.Checkpoint[int | None]("cursor"). With Pydantic installed, a spelled-out type — any type it can validate, from a model to a plain annotation like dict[str, int] — is validated on write and on read; without it, T only informs type checkers.

checkpoint.get() -> T

The current value, or the declared default if it isn't set. A checkpoint explicitly set to None reads back as None. With Pydantic, the stored value is validated as a T and returned as one (a model as an instance).

checkpoint.set(value) -> None

Sets the value, replacing anything already there. With Pydantic, the value is validated as a T first and stored as plain data (a model as its fields).

checkpoint.update(fn: Callable[[T], T]) -> T

Sets the value to fn(current) and returns what was stored. fn receives the declared default when the checkpoint isn't set. A read followed by a write, not an atomic swap.

checkpoint.reset() -> None

Clears the checkpoint, so get() returns the declared default again. Distinct from set(None), which stores None.

checkpoint.is_set() -> bool

Whether the checkpoint has a value (including an explicit None).

Metrics

Record numeric values from executions, streamed in real-time and rendered as charts in Studio. See metrics.

Metric

Defines a metric and provides a record() method for emitting data points. Metrics can be defined at module level or within a function.

cf.Metric(
    key: str,
    *,
    group: str | MetricGroup | None = None,
    scale: str | MetricScale | None = None,
    throttle: float | None = 10,
)

Parameter	Type	Default	Description
key	str	required	Metric key name
group	str | MetricGroup | None	None	Group (metrics in the same group share a chart)
scale	str | MetricScale | None	None	Y-axis scale (metrics with the same scale name share a y-axis)
throttle	float | None	10	Max data points per second (None = no limit)

metric.record(value, *, at=None)

Records a data point. at specifies an explicit x-value (defaults to time since execution start).

MetricGroup

Configures the x-axis for a group of metrics displayed on a shared chart.

cf.MetricGroup(
    name: str,
    *,
    units: str | None = None,
    lower: float | None = None,
    upper: float | None = None,
)

MetricScale

Configures the y-axis. Metrics sharing a scale name within a group share a y-axis.

cf.MetricScale(
    name: str | None = None,
    *,
    units: str | None = None,
    progress: bool = False,
    lower: float | None = None,
    upper: float | None = None,
)

progress(iterable, key="progress", *, group=None)

Wraps a sized iterable and automatically records a progress metric as items are consumed. Renders as a progress bar in Studio.

Context functions

group(name=None, *, concurrency=0)

Context manager for grouping child executions in the graph view, and optionally capping how many of them run at once. See groups.

with cf.group("batch processing", concurrency=4):
    for item in items:
        process.submit(item)

Parameter	Type	Default	Description
name	str | None	None	Label for the group in Studio.
concurrency	int	0	At most this many of the group's children run at once. 0 means no limit. Direct children only; scoped to this execution of the caller.

suspense(timeout=None)

Context manager that sets a timeout on .result() calls within its scope. If the result isn't ready within the timeout, the execution is suspended and automatically re-run later. See suspense.

suspend(delay=None)

Explicitly suspends the current execution. It will be re-run after the specified delay. delay can be float (seconds), timedelta, or datetime.

flush()

Blocks until buffered state (checkpoints, metrics, logs) has reached the server. Not needed before suspending, returning or raising — those flush automatically. See checkpoints.

select(handles, *, cancel_remaining=False)

Wait for the first of one or more handles (Execution and/or Input) to resolve. Returns (winner, remaining) — call winner.result() to get the value (or to raise the exception that resolved it). Picks up its timeout from any enclosing cf.suspense(timeout=...) scope; raises TimeoutError if the wait expires. See select.

winner, remaining = cf.select([a.submit(), b.submit()])

Parameter	Type	Default	Description
handles	Sequence[Execution | Input]	required	Handles to wait on (must be non-empty).
cancel_remaining	bool	False	Atomically cancel non-winner Execution handles when one resolves. Input handles are left pending.

cancel(handles)

Atomically cancel one or more handles. Execution handles are cancelled recursively (descendants too); Input handles transition to a terminal cancelled state (distinct from dismissed). Already-resolved handles are silently skipped.

cf.cancel([execution_a, execution_b, input_c])

Logging

Structured logging functions that associate key-value pairs with a template string. Values are stored separately from the template, enabling filtering and search in Studio. See logging.

cf.log_debug(template=None, **kwargs)
cf.log_info(template=None, **kwargs)
cf.log_warning(template=None, **kwargs)
cf.log_error(template=None, **kwargs)

asset(entries=None, *, at=None, match=None, name=None)

Creates and persists a collection of files as an asset, which can be inspected and downloaded from Studio or the CLI. See assets.

Catalog

Catalog[T](path)

A handle to a path in the catalog, declared once (usually at module level) and used from any target. A {placeholder} in the path is filled in with at(). Nothing round-trips until the handle is used. An invalid path raises ValueError here rather than on use. T is the type of the values at the path. With Pydantic installed, any type it can validate — a model, a dataclass, a plain annotation like dict[str, int] — is validated on publish and on read; without it, T only informs type checkers. Also a handle for cf.select, resolving when the path has a version this execution hasn't seen, which on an empty path is the first. Not a value: pass the path or the value it holds to a task, not the handle.

handle.template is the path as declared, placeholders included; handle.path is the path once every placeholder is bound, and a ValueError before.

handle.at(**placeholders) -> Catalog[T]

A handle to the path with the given placeholders filled in, of the same type. Values are substituted as strings and the result has to be a valid path. Placeholders left out stay unbound; a handle with any unbound refuses publish(), current() and next() with ValueError.

handle.publish(value) -> int

Publishes a value at the path and returns the version's number. value is anything that can be passed to a task: an asset, a data structure holding assets, a reference to something external. Facts about the publish, such as a metric, go in the value alongside the thing itself. Publishing what is already the latest version (the same value) returns the existing version's number without writing. With Pydantic, the value is validated as a T first and stored as plain data (a model as its fields).

handle.current() -> T

The value at the path as of the execution's snapshot. Waits for a first publish if there is none — blocking outside a cf.suspense scope, suspending inside one. With Pydantic, validated as a T and returned as one (a model as an instance).

handle.next() -> NoReturn

Suspends the execution until the path has a version newer than the execution can see, whether or not inside a cf.suspense scope. The execution that resumes the step sees it with current(). Never returns.

Exceptions

Exception	Description
ExecutionError	Child execution failed. When the original exception type can be resolved, the raised exception subclasses both ExecutionError and the original type, so you can catch either.
ExecutionTerminated	Base class for the reasons below: the execution (or stream producer) ended without completing. Catch it to handle them all.
ExecutionCancelled	Child execution (or input) was cancelled.
ExecutionTimeout	Child execution exceeded its configured timeout.
ExecutionAbandoned	The worker running the child execution went away without reporting a result.
ExecutionCrashed	The child execution's process ended without reporting a result.
StreamSuperseded	A stream ended because its recurrent producer finished an iteration. Not a failure: the next iteration produces its own stream.
InputDismissed	A requested input was dismissed by the responder.
RequestError	The server refused a request the execution made — reading a catalog version it can't see, say. The refusal's code is .code.
TimeoutError (built-in)	A cf.suspense(timeout=...) wait expired before a handle resolved.