Workflows (sync)

Generated from the type stubs and docstrings. Do not edit by hand.

Types

RunHandle

One run: what WorkflowClient.submit returns and WorkflowClient.handle gets.

Each method is the client’s method of the same name, bound to this run; see pyrula.workflows.RunHandle for the async form.

RunHandle(client: WorkflowClient, name: str, run_id: str)

Fields:

• name: str
• run_id: str

signal

signal(self, signal: str, payload: Optional[dict[str, Any]] = None, *, signal_id: Optional[str] = None) -> str

query

query(self, query: str, *args: Any, **kwargs: Any) -> Any

update

update(self, update: str, payload: Optional[dict[str, Any]] = None, *, update_id: Optional[str] = None, timeout: float = 30.0) -> Any

cancel

cancel(self, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> None

terminate

terminate(self, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> TerminateResult

retry

retry(self) -> RunHandle

resume

resume(self, action: str, value: Any = None, *, wait_id: Optional[str] = None) -> None

status

status(self) -> RunStatus

result

result(self, *, timeout: Optional[float] = None, on_interrupt: Any = None) -> Any

events

events(self) -> list[RunEvent]

---

WorkflowClient

Start and talk to workflow runs from synchronous code.

Wraps pyrula.workflows.WorkflowClient on an event loop owned by this client. A daemon thread runs that loop and starts on the first call; each method hands its coroutine to the loop and blocks for the answer, so one client can be shared by every thread of a threaded server. Exceptions are the async client’s own.

A method called from a thread that is running an event loop raises RuntimeError: it would block that loop. Use the async client there. After a fork the child builds its own loop on first use and never touches the parent’s. close (or with) stops the thread; an open client is also closed at interpreter exit.

WorkflowClient(store: Store, *, poll_interval: float = 0.1)

Fields:

• store: Store

close

close(self) -> None

Stop the loop thread. Idempotent. A call blocked in another thread raises RuntimeError; later calls raise it too. Calls cut short this way may still have taken effect; see the guide.

submit

submit(self, name: str, params: Optional[dict[str, Any]] = None, *, idempotency_key: Optional[str] = None, owner: Optional[str] = None, metadata: Optional[dict[str, Any]] = None, allow_unregistered: bool = True, start_at: Optional[datetime] = None, delay: float | timedelta | None = None) -> RunHandle

Queue a new run of name and return a handle to it.

See pyrula.workflows.WorkflowClient.submit for every argument.

handle

handle(self, name: str, run_id: str) -> RunHandle

A handle to an existing run.

trigger

trigger(self, name: str, event: dict[str, Any]) -> Optional[RunHandle]

Send event to name’s Collect or Debounce buffer; returns a handle to the run a completed batch started, else None. See pyrula.workflows.WorkflowClient.trigger.

plan_retry

plan_retry(self, name: str, run_id: str) -> RetryPlan

What retrying a failed run would carry over and re-run, without starting anything. See pyrula.workflows.WorkflowClient.plan_retry.

find_retry

find_retry(self, name: str, run_id: str) -> Optional[ExistingRetry]

The retry already started for a failed run, or None. See pyrula.workflows.WorkflowClient.find_retry.

retry

retry(self, name: str, run_id: str) -> RunHandle

Start a new run that carries on from where the failed run stopped; see pyrula.workflows.WorkflowClient.retry.

signal

signal(self, name: str, run_id: str, signal: str, payload: Optional[dict[str, Any]] = None, *, signal_id: Optional[str] = None) -> str

Send a signal; returns "delivered" or "buffered". See pyrula.workflows.WorkflowClient.signal.

query

query(self, name: str, run_id: str, query: str, *args: Any, **kwargs: Any) -> Any

Call a ctx.on_query handler against the run’s replayed state. See pyrula.workflows.WorkflowClient.query.

update

update(self, name: str, run_id: str, update: str, payload: Optional[dict[str, Any]] = None, *, update_id: Optional[str] = None, timeout: float = 30.0) -> Any

Deliver an update, wait for a worker to apply it, and return its value. See pyrula.workflows.WorkflowClient.update.

cancel

cancel(self, name: str, run_id: str, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> None

Request cancellation. See pyrula.workflows.WorkflowClient.cancel.

terminate

terminate(self, name: str, run_id: str, *, reason: Optional[str] = None, requested_by: Optional[str] = None) -> TerminateResult

End the run now, with no worker involved. See pyrula.workflows.WorkflowClient.terminate.

resume

resume(self, name: str, run_id: str, action: str, value: Any = None, *, wait_id: Optional[str] = None) -> None

Answer the run’s ctx.interrupt. See pyrula.workflows.WorkflowClient.resume.

status

status(self, name: str, run_id: str) -> RunStatus

The run’s current status.

result

result(self, name: str, run_id: str, *, timeout: Optional[float] = None, on_interrupt: Any = None) -> Any

Wait for the run to finish and return its result.

timeout is enforced by the async client, which raises DurableInvocationTimeout; the other failures are the async client’s too. See pyrula.workflows.WorkflowClient.result.

events

events(self, name: str, run_id: str) -> list[RunEvent]

The run’s own journal, oldest first.

list_runs

list_runs(self, name: str, *, limit: int = 50, cursor: Optional[str] = None, owner: Optional[str] = None) -> tuple[list[dict[str, Any]], Optional[str]]

Runs of name, newest first, and the cursor for the next page. With owner, only that owner’s. The store’s own list_runs, called on this thread.