Workflow

blobhub.types.workflow.Workflow is the typed module for blobhub.compute.workflow blobs. It does one job: run a definition that is already deployed, follow the run until it ends, and read what it did. Writing and deploying definitions is blobhub-cli’s job.

import blobhub
from blobhub.types.workflow import Workflow

hub = blobhub.connect()
revision = hub.blob("acme", "support-bot").revision()     # a draft: runs live in its sessions

question = {
    "type": "message",
    "message": {"role": "user", "content": [{"type": "text", "text": "Where is my order?"}]},
}
run = Workflow(revision).execute(
    "answer-question",                     # the definition: an alias or id on this revision
    input_data={"question": question},     # data component alias -> value
    session="sdk-demo",                    # a session alias, created when absent
)
print(run.url)                             # the run in blobhub.io's debugger
status = run.wait(timeout=900, on_event=lambda event: print(event["type"], event.get("message", "")))

Workflow(revision) raises TypeMismatch when the revision’s blob is not a workflow blob.

Execute

workflow.execute(reference, *, input_data=None, session=None) creates one execution and returns an Execution as soon as it exists, with status creating. It does not wait; wait() does. Where it runs and what it runs are separate. The run lives in a session of the revision you built Workflow over, so that revision must be a draft you can write to; a committed one raises Conflict (revision_not_writable). The definition it runs is the reference, which may name a definition on any revision your credential can read, committed ones included. See Workflow Reference.

The reference

Form	Example	Runs
`<definition>`	`answer-question`	a definition of this revision, by alias or id
`<blob id>/<revision>/<definition>`	`5f2c…/default/answer-question`	a definition of the blob with that id
`<org>/<blob>/<revision>/<definition>`	`acme/faq-bot/default/answer-question`	a fully qualified definition

<revision> is default or a revision id, and a blob named by alias needs its organization. Each part is at most 64 characters. A mapping works too, keyed like the API’s workflow object: definition alone; blob, revision and definition; or org, blob, revision and definition. A malformed reference raises ValueError before anything is sent. The platform refuses a definition it cannot resolve with PermissionDenied, and a playground definition named by id with CommandError (invalid_definition_category).

Input data

input_data maps a component alias to a value. Each value reaches the workflow’s data component whose alias port names it and whose persistence is execution. An alias that matches no such component is ignored without an error. A value is a data type value of the component’s own type: a message value for a data.message component, a messages value for data.messages, an integer value for data.integer. The platform requires only an object with a type and never checks it against the component, and the SDK sends values exactly as given, so a mismatch reaches the workflow unnoticed. For the usual data.message input:

input_data = {
    "question": {
        "type": "message",
        "message": {"role": "user", "content": [{"type": "text", "text": "Where is my order?"}]},
    },
}

Sessions

Every execution belongs to a session. session says which:

`session`	What happens
`None`, the default	A new session is created for this run.
an open session’s id or alias	That session is used.
an alias no session has	A session is created under that alias. If another caller creates it first, theirs is used.
an id no session has	`PermissionDenied`. An id is never created; absent and unreachable answer alike.
a session that is not open	`Conflict` (`session_not_open`), before anything is created.

A session alias is 6 to 42 characters of a–z, 0–9, _ and -, and must not look like an id. A session execute created stays even when creating the execution then fails.

Following a run

Attribute	Holds
`id`, `session_id`	The execution and its session.
`status`	`creating`, `running`, `stopping`, `stopped`, `completed` or `failed`, as last read.
`url`	The run in blobhub.io’s debugger.
`record`	The whole execution record, as Get Execution returns it.

run.refresh() reads the record again.

Wait

run.wait(*, timeout=900.0, on_event=None, interval=2.0) polls every interval seconds until the run is completed, failed or stopped, and returns that status.

A failed or stopped run is returned, not raised. Check the status.
timeout raises WaitTimeout and leaves the run going. Call wait() again to keep following it.
on_event receives each new execution event, oldest first, as a dict with its type, its created_at and usually a message. No event is passed twice, across calls to wait() too. An exception from the callback propagates, and the event it was given counts as delivered.
The last events can arrive after wait() returns. The platform indexes events a moment after it writes them, and a stopped run writes its status before its last events. run.events(), read afterwards, is the complete log.
Without on_event, wait() reads no events at all.

A session closed while the run is going stops it, and wait() returns stopped.

Events

run.events() returns the run’s whole event log, oldest first, following the listing to its end.

Stop

run.stop() asks the platform to stop the run, which turns stopping, then stopped. Only a running execution can be stopped: straight after execute() the run is still creating, and stop() raises CommandError (invalid_execution_status). Wait for running first:

import time

while run.refresh().status == "creating":
    time.sleep(1)
run.stop()

When creating a run fails

Creating a session and creating an execution are writes, and the SDK never repeats a write after a failure that may have reached the platform (see Errors and retries). The platform takes no idempotency key, so a repeated create_execution would start a second run. execute() makes up to three requests: it reads the session, creates it when it is absent, then creates the execution. Only the last can start a run. A failure in either session step leaves at most an empty session, which the next call with the same alias finds and reuses. But execute() raises the same classes from every step, so when it raises ServerError, or a NetworkError that is not RequestNotSent, a run may have started. Look before you run it again. With a session alias, that is two reads:

session = revision.query("get_session", session_id="sdk-demo")["session"]
runs = revision.query("list_executions", session_id=session["id"])["executions"]

Both read indexes that catch up a moment after a write, so wait a few seconds first. Then an empty list, or a PermissionDenied from get_session for a session that was never created, means nothing ran. A RateLimited or RequestNotSent error means nothing ran, and running it again is safe. Name the session with an alias whenever a second run would matter: a session execute() created for you has an id you never saw.

Errors

Error	When
`ValueError`	The reference is malformed, or `session` is blank.
`TypeError`	`input_data` is not a mapping of alias strings to value dicts.
`PermissionDenied`	The definition or a session id does not resolve, or the credential cannot write here.
`Conflict` `session_not_open`	The session is closed, closing or being deleted.
`Conflict` `revision_not_writable`	The revision is committed.
`CommandError` `invalid_definition_category`	The reference names a playground definition by id.
`CommandError` `limit_exceeded`	The revision is at its sessions limit, or the session at its executions limit.
`CommandError` `invalid_execution_status`	`stop()` on a run that is not `running`.
`WaitTimeout`	`wait()` ran out of time; the run continues.

Python SDK

Execute

The reference

Input data

Sessions

Following a run

Wait

Events

Stop

When creating a run fails

Errors

See also

​Execute

​The reference

​Input data

​Sessions

​Following a run

​Wait

​Events

​Stop

​When creating a run fails

​Errors

​See also

Execute

The reference

Input data

Sessions

Following a run

Wait

Events

Stop

When creating a run fails

Errors

See also