fusee
Fusee for Python: durable execution at the speed of the disk.
Definitions are plain values, apart from any engine::
import fusee
review = fusee.signal("review", bool)
@fusee.workflow("checkout")
async def checkout(ctx: fusee.Ctx, order: Order) -> str:
quote = await ctx.step("quote", lambda: pricing.quote(order))
charge = await ctx.step("charge", lambda s: payments.charge(quote, key=s.key),
retry=fusee.Retry(attempts=5))
approved = await ctx.wait_for(review, timeout="1d")
return charge["id"] if approved else await ctx.step("refund", lambda: payments.refund(charge))
engine = await fusee.Engine.open("./data", workflows=[checkout])
run = await engine.start(checkout, "order-123", order)
await run.signal(review, True)
print(await run.result())
Workflow code runs under Fusee's deterministic scheduler, so it may only
await ctx operations (or ctx.all / ctx.race of them and of other
workflow coroutines). Step functions run on the asyncio loop (or a thread,
with thread=True) and may do anything.
Runs at once: at most limit runs of the type run at a time, or of
each key (key(input), a tenant say); the others wait, in the order
they were started, as fusee.queued/<id> runs, and result() waits
through the queue. The turns are durable: a run keeps its turn while it
sleeps and across crashes. Starts from outside a workflow take turns; a
child started by a workflow does not.
Names who asks for the signals, cancellations, terminations and
resets made inside the block: their journal records carry it as by
(audit). The HTTP API names its token or the proxy's signed-in user
the same way.
Defines a workflow type::
@fusee.workflow("checkout")
async def checkout(ctx: fusee.Ctx, order: Order) -> str: ...
@fusee.workflow alone names it after the function. The input's and
result's annotations (a dataclass, pydantic model, msgspec struct,
TypedDict, datetime, ...) decide what the code receives and what
run.result() returns; input= and output= override them.
timeout is the workflow's deadline from its start.
Defines a signal. A plain string works wherever a signal does.
Defines a query: status = fusee.query("status", result=Status).
Defines an update: approve = fusee.update("approve", arg=Approval, result=bool).
Defines a keyed object type (see ObjectType).
A workflow type. Calling it runs the function directly (for tests).
A signal's name and payload type: review = fusee.signal("review", bool).
Abstract base class for generic types.
On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::
class Mapping[KT, VT]:
def __getitem__(self, key: KT) -> VT:
...
# Etc.
On older versions of Python, however, generic classes have to explicitly inherit from Generic.
After a class has been declared to be generic, it can then be used as follows::
def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
try:
return mapping[key]
except KeyError:
return default
Abstract base class for generic types.
On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class's name::
class Mapping[KT, VT]:
def __getitem__(self, key: KT) -> VT:
...
# Etc.
On older versions of Python, however, generic classes have to explicitly inherit from Generic.
After a class has been declared to be generic, it can then be used as follows::
def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT:
try:
return mapping[key]
except KeyError:
return default
A keyed object type. Handlers are exclusive (one call at a time per key, in arrival order; they may change state) unless marked shared::
cart = fusee.object("cart", {"add": add, "items": fusee.shared(items)})
or with decorators::
cart = fusee.object("cart")
@cart.exclusive
async def add(ctx, item): ...
How a failing step is retried. attempts includes the first.
A failure marked final (fusee.Fatal, or any exception with
retryable = False) is never retried. non_retryable names
exception types that are final; retry_if(failure, attempt) decides
each time. Each delay is drawn between half of it and all of it unless
jitter=False; no retry starts later than max_elapsed after the
first failure. on_exhausted="pause": when the retries run out on a
failure that could be retried, the run pauses (its parent and callers
keep waiting) until run.resume() runs the step again with a fresh
set of attempts; "fail" fails the step; None takes the engine's
on_exhausted.
When a schedule fires: a cron expression or an interval.
An embedded engine: it records runs in a directory (or memory) and runs the workflows it was opened with.
Opens an engine that runs workflows (and objects).
With no directory: FUSEE_DIR; else FUSEE_URL (a worker of a
Fusee server); else ./fusee-data. durability is local
(the default with a directory), memory, server or
replicated.
profile="dev" (or FUSEE_PROFILE=dev) is for development: it
says so on stderr, keeps every run's history, flushes with fsync
rather than F_FULLFSYNC on macOS (it survives a process crash,
not power loss), warns when a synchronous step holds the event loop,
replays each run against the code when it parks or ends and logs any
difference, makes result() of a stuck run raise
DeterminismError, shows ctx.log records on stderr when logging
is not configured, and serves the UI.
http="127.0.0.1:7601" serves the HTTP API and UI from the engine
("off": never; under the dev profile the default is
FUSEE_HTTP, else 7601, else a free port).
record_stacks=False keeps stack traces out of recorded failures.
on_exhausted="pause" pauses a run whose step runs out of retries
(unless the step's Retry says otherwise) until run.resume().
sync_steps="thread" runs synchronous step functions on threads.
codec=aes_gcm({...}) encrypts payloads; telemetry=
open_telemetry() reports spans and metrics.
Serves the HTTP API and the web UI on a thread of their own (they
call this engine through its event loop; nothing rides the step
path). Off loopback a token is required. Returns the URL::
print(engine.serve()) # http://127.0.0.1:7601
Adds workflows or object types after the engine opened (prefer
Engine.open(workflows=[...])).
Calls handler on an engine event: "stuck" (workflow_id,
message) when a run stops on a determinism error, "paused"
(workflow_id, step, failure) when a step of this process runs out
of retries and pauses its run, and "waiting_for_code"
(workflow_id, type) when a run has waited five seconds for code
no process registered.
A handle to a run by workflow ID; workflow types its result.
Starts a run of workflow (a definition or a type name) under
workflow ID id; returns once it is durable. id_policy (an
IdPolicy or {"running": ..., "finished": ...}) says
what happens when the ID is taken. delay="10m" or at=datetime
starts it later: a durable timer (the run fusee.delay/<id>), then
the start; result() waits through the delay.
Signals the run id, starting it first with input if it is
not running: the signal is recorded before the new run's code first
runs. Returns the run::
await engine.signal_with_start(cart, "cart-7", Cart(), add, item)
Starts the workflow if it is not running and delivers an update in
the same record. Returns (run, result).
Creates a schedule (or updates its spec) that starts workflow
with input on spec: a cron string, a dict
({"every": "1h"}) or a ScheduleSpec.
Runs, newest first per shard, filtered by type, status, start time
and search attributes ({"amount": {"gte": 100}})::
async for r in engine.list(status="failed"):
print(r.id, r.error)
Finishes an async completion with a value; returns once recorded durably.
Fails an async completion (a message, an exception or a Failure).
Debounces a start under key: starts inside the window collapse
into one run, which begins with the latest start's input when the
window closes (window after the latest start, or max_wait
after the first). Returns once this start is durable, with the
handle of that run (await run.started()). event_id makes a
retried call join once.
Adds an event to the batch under key: once it holds
max_events events (default 100) or max_wait (default 1s) has
passed since its first, one run starts with all of them as its
input (a list).
Defines (or resizes) a concurrency and rate limit for steps.
Every limit this engine defined, with its permits now, by name
(fusee limits, the UI's Limits page).
Counters summed over shards (as GET /v1/metrics's total),
with the runs stuck and waiting for code.
Per-shard counters in shard-number order, durations in microseconds.
A run's journal as an export file (.fjnl) for Replayer.
Starts, signals and reads runs through a Fusee server. The same
interface as Engine::
client = await fusee.Client.connect("db1:7600", token_file="/etc/fusee/token")
run = await client.start(checkout, "order-123", order)
Connects to a server: address (default FUSEE_URL) and, as
options, tls={"ca": ..., "cert": ..., "key": ...} and
token_file.
Sends one protocol call without blocking the event loop.
See Engine.start (delay=/at= start it later).
A server address with its options in the form the native layer reads.
A handle to a run by workflow ID, from an engine or a client.
Sends a signal: await run.signal(review, True).
Calls a query handler; the answer waits until the state it read is durable.
Sends an update and waits for its handler's durable result. A
retried call with the same id gets the recorded result.
The run's result (following continue-as-new and reset). A failed
run raises its error: a step's as StepError (e.cause is the
step's failure), anything else as WorkflowError.
Waits until a run with this ID exists (a debounced or batched run starts only when its window closes).
Delivers cancellation to the run's waits; its code may compensate.
Ends the run at once; children follow their parent close policy.
Resumes a run that paused because a step ran out of retries
(Retry(on_exhausted="pause")): the step runs again at once, with
a fresh set of attempts, on the code this process has now. Raises
RejectedError if nothing is paused.
Starts a new run whose journal is this one's cut before before_step.
Runs a failed run again from its failed step: a reset before it (the steps before keep their results).
One run, as listings and run.status() give it.
A run and its journal: each record a dict with its kind (start,
step, attempt, timer, signal, wait, child, child_done, send, update,
update_done, completed, heartbeat, attributes, chunk, stream_end, done,
...), times and run IDs left out.
A limit and its permits right now (engine.limits()).
A key of an object type, called from outside workflow code.
Calls a handler and waits for its durable result. Calls with the
same id run once, and each gets the recorded result.
Calls a handler without waiting for its result: returns once the call is accepted.
Edits the state from outside its handlers: an exclusive call, so
it waits for the key and commits as a handler's writes do. edit
has clear_all/clearAll, clear (fields), set (fields
and values; None removes one) and patch (a JSON Patch over
the state as one object), applied in that order. if_version
refuses the edit unless the state is still at that version (the
version in describe()'s object). Returns {"version",
"changed"}.
Sets one field of the state.
Applies a JSON Patch (RFC 6902) to the state as one JSON object.
Clears one field, or (none named) the whole state; the key remains.
The key's run: its journal, and object with its state and
pending calls.
A schedule, addressed by name.
engine.schedules.get(name), engine.schedules.list().
A reader of a run's stream: async for chunk in run.stream("tokens")
yields each chunk once it is durable, in order, and ends once the stream
is complete. offset is where a reader that reconnects continues;
status says how the stream ended (ok, a failed step's
failed, or the run's status if it ended first).
Observe a replica, including while its worker waits to become primary.
The workflow context. Code between ctx calls must be deterministic: it runs again against the journal after a crash.
Registers a step that undoes what the workflow did so far: if the
workflow fails or is cancelled, every registered compensation runs
as a step named name, newest first, then the failure is raised.
Register it after the step it undoes succeeded::
hold = await ctx.step("reserve", lambda: inventory.reserve(sku))
ctx.compensate("release", lambda: inventory.release(hold))
await ctx.step("charge", lambda: payments.charge(cents)) # fails: "release" runs
A logger tagged with the workflow ID and run (the fusee
logger). Silent while the run replays its journal, so a line is
written once however often the code runs; not journaled.
Runs fn (sync or async) as a durable step and returns its
recorded result (as JSON decodes it, or as output's type).
fn may take one argument, the step context s: s.key (an
idempotency key), s.attempt, s.heartbeat(p), s.stream(n).
It runs once everything the run recorded before is durable; with
read_only=True at once (use it only for steps that change
nothing outside). thread=True runs a synchronous fn on a
thread so it does not hold the event loop.
With limit, each attempt first takes a permit of that limit
(engine.limit); a higher priority goes first, and steps of
one priority take turns by fairness_key.
A step completed from outside: send(cb) starts work elsewhere
with cb.token (or cb.url); the workflow waits until a process
calls engine.complete(token, value) or engine.fail(token,
error). send is retried under the retry policy, each attempt
with a new token.
Opens the run's stream name for appending (readers follow it
with run.stream(name)). Its appends are recorded once (a replay
appends nothing), and close() or the run's end completes it. A
step opens a stream of its own with s.stream(name).
Sleeps durably: the wake time is recorded and survives restarts.
Waits for the next signal; None on timeout.
Starts a child workflow and waits for its result. The child's ID makes the start idempotent.
Starts a child without waiting. By default it is terminated when
this workflow closes; parent_close="abandon" lets it outlive it.
Signals another workflow, once this point is durable.
Asks another workflow to cancel, once this point is durable.
Key key of object type type, called from this workflow:
each call is a journal operation, run exactly once.
Registers a read-only query handler (run.query(query, arg)).
Registers an update. validate runs first (raise, or return a
string, to refuse; a refusal records nothing); handler (sync or
async, may run steps) runs once per accepted update.
Ends this run and starts a new run of the same ID with input,
in one record. Await it; it never returns.
True for new runs, and for running ones the first time they get here after the deploy; false while replaying an older journal.
The step function's optional argument: ctx.step("charge", lambda s:
pay(key=s.key)). Built only for a function that declares it.
A key stable across retries and recovery for this step (distinct for each run of a workflow ID, including one that replaced it by reset). Pass it to services that accept idempotency keys.
True once the run was cancelled, terminated or closed. An async
step function is cancelled at its next await; a function on a
thread (thread=True) checks this to stop early.
Reports progress. Coalesced; never waits for the disk.
Opens the step's stream name: the step's result completes it,
and an attempt that runs again finds what earlier attempts recorded
in recorded().
The context of an object's handler: the workflow context of the key's
run, its state and its object_key.
A key's state as a handler sees it: JSON values under string keys.
The value, or default when the key has none. An exclusive
handler sees its own writes.
A key called from workflow code (ctx.object(type, key)).
A writer of one of the run's streams (ctx.stream / s.stream).
What ctx.await_completion's send function receives: the token to
complete with (engine.complete(token, value)), and the callback URL
when the engine serves its HTTP API with a public address.
A failure as recorded: what a failed step, update, object call or
workflow carries, in every SDK. type is the error's class name
(absent for a plain Exception); code and details are the
application's; retryable=False marks it final.
Raise (or subclass) to fail a step without retries::
class CardDeclined(fusee.Fatal):
code = "card_declined"
raise CardDeclined("card declined", details={"last4": "4242"})
Any exception with retryable = False is final as well.
A step failed after its last attempt (or with a final failure). The
step's own failure is cause: e.cause.type, e.cause.code,
e.cause.details.
A workflow, update or object handler failed; its failure is cause.
A request was refused and nothing was recorded.
Runs workflow code against journals exported with fusee export
(or engine.export_journal), without running step functions or
timers, and reports the first difference by step name::
report = await fusee.Replayer(workflows=[checkout]).replay_file("order-1.fjnl")
assert report.ok, report.error
OpenTelemetry instruments for an engine: a fusee.step span per
step attempt, current while the step's function runs, and the metrics
fusee.steps.executed, fusee.step.duration (ms),
fusee.workflows.started and fusee.workflows.finished. Replayed
steps record nothing. Build one with open_telemetry(), or pass any
tracer and meter with the OpenTelemetry API's methods.
Instruments from the application's opentelemetry package (the
API; the application configures providers and exporters).
Payload encryption with AES-256-GCM: aes_gcm({"k2": new_key, "k1":
old_key}). Keys are 32 bytes (or hex / base64 text). The first key
encrypts; every key decrypts.
A duration in milliseconds: a number (ms), "500ms", "10s",
"5m", "2h", "1d", or a timedelta.
The worker side of invocations over HTTP (suspend to zero).
When a Fusee server started with --invoke URL has work for shards no
worker holds, it POSTs to that URL; each request opens a fresh engine
for the invocation's shards, runs them until they are idle or the
budget ends, and closes it. The handler is an ASGI app (uvicorn
module:app); lambda_handler is an AWS Lambda handler; invoke
serves any other framework. An engine belongs to the thread that opened
it: run invocations on one thread (Lambda and ASGI servers do)::
app = invocation_handler("fusee.internal:7600?token-file=/run/token",
workflows=[order],
secret=os.environ["FUSEE_INVOKE_SECRET"])
register(engine), if given, runs after the engine opens (for setup
beyond the definitions).
See InvocationHandler.
Checks a fusee-signature value (t=<unix ms>,v1=<hex HMAC-SHA256>)
within five minutes. Returns why it fails, or None.