Ogha — durable execution for Python

Write ordinary typed functions that survive crashes, restarts, deploys, and month-long waits. One App owns registration, connection, and worker lifecycle; Ogha records every durable boundary and reuses its result during recovery.

Breaking in 0.3.0. The explicit workflow Context, module-level decorators, and manual Worker authoring API were removed. Applications must use one App, @app.step, @app.workflow, and direct durable function calls. See Migrating to 0.3 before upgrading a running deployment.

from dataclasses import dataclass

import ogha

app = ogha.App("checkout")

@dataclass
class Order:
    id: str
    card: str
    amount: int

@dataclass
class Receipt:
    id: str
    approved: bool

@app.step(retry=ogha.RetryPolicy(max_attempts=5), pivot=True)
def charge(order: Order) -> Receipt:
    result = payments.charge(order.card, order.amount)
    return Receipt(result.id, result.approved)

@app.workflow
async def checkout(order: Order) -> Receipt:
    receipt = await charge(order)
    if order.amount > 10_000:
        await ogha.approval(
            "large-charge", evidence=receipt, timeout=24 * 60 * 60
        )
    ogha.event("order.charged", {"order": order.id})
    return receipt

if __name__ == "__main__":
    app.serve()

The original function signatures and annotations are retained. At a durable JSON boundary, Ogha restores supported resolved annotations—such as dataclasses, containers, enums, dates, UUIDs, and optional Pydantic models—so live execution and recovery see the same declared type. Unannotated and unresolved types remain plain decoded values rather than being guessed.

Install

pip install ogha          # Python 3.10+

Run the engine

Ogha ships as one container carrying the engine, storage, and dashboard. Pull it and open http://localhost:8080:

docker run -p 8080:8080 public.ecr.aws/h6s8i3h8/ogha:0.1.2-obf \
  --storage=postgres --dsn="postgres://…" --auth-disabled

A ready-to-use Compose stack is in the getting-started guide linked below.

Sync, Async, and Async Distributed

The three common experiences use the same durable Run:

Experience	Code	Caller	Worker placement
Sync	receipt = checkout.options(run_id=order.id).run(order)	waits for the typed result	default sticky, or the workflow's declared placement
Async	run = checkout.options(run_id=order.id).start(order)	receives a durable RunHandle immediately	default sticky placement
Async Distributed	run = distributed_checkout.options(run_id=order.id).start(order)	receives a durable RunHandle immediately	each step is independently dispatched and fenced

Declare distributed placement on the workflow:

@app.workflow(execution="async_distributed")
async def distributed_checkout(order: Order) -> Receipt:
    return await charge(order)

Sync versus Async is a caller completion choice. Async versus Async Distributed is a worker-placement choice. Waiting can be applied to either placement; there is no execution="sync". The default public execution value is "async", which uses Ogha's sticky placement.

For custom caller wait settings, start first and then read the result:

run = checkout.options(run_id="order-42", timeout=60 * 60).start(
    Order("order-42", "card-token", 1250),
)

receipt = run.result(timeout=30)       # caller wait only; the run keeps going
# receipt is a Receipt, recovered from its declared return annotation

await run is the non-blocking asyncio caller form. It moves status polling to a worker thread; ordinary Ogha workflow bodies accept only Ogha durable awaitables.

Eager handles and structured concurrency

Calling a step or child workflow inside a workflow creates it immediately and returns an awaitable Handle. Work does not wait for the Handle to be awaited.

@app.workflow
async def price_order(order: Order) -> list[Quote]:
    handles = [quote.options(name=carrier)(order, carrier) for carrier in CARRIERS]
    return await ogha.gather(*handles)

If a workflow returns while owned children are unfinished, Ogha automatically joins them in deterministic creation order. A child workflow can outlive its parent only when detach is explicit:

audit.detach(order)                       # an independent child Run
audit.options(detached=True)(order)       # equivalent immutable view

Ogha does not offer detached steps inside the parent's Run. On user-code failure, the facade requests cancellation of unfinished owned children and preserves the original failure even if cancellation cannot be delivered.

Immutable per-call options

Options never consume names from the function's keyword arguments and never mutate the registered function or another held view:

fast = charge.options(timeout=5, name="fast-charge")
slow = charge.options(timeout=30, name="slow-charge")

fast(order)       # eager Handle; charge itself is unchanged
slow(order)

Step views support name, timeout, and target. Workflow views support run_id, timeout, target, and detached.

Durable control helpers

The context-free helpers lower to Ogha's existing durable operations:

Helper	Meaning
await ogha.sleep(seconds)	durable timer; no worker held while sleeping
await ogha.signal(name, timeout=...)	externally settled webhook/event
await ogha.approval(action, evidence=..., timeout=...)	auditable, fail-closed approval
await ogha.gather(*handles)	wait for all owned work
await ogha.race(*handles)	first durable settlement
await ogha.quorum(n, *handles)	first durable quorum
ogha.event(name, payload)	replay-suppressed run fact
ogha.cancel(handle, reason)	cooperative child cancellation

approval is one externally settled Promise with evidence and a timeout whose expiry denies. It does not add a separate server state machine.

race and quorum select results but do not cancel their remaining Handles. Cancel every nonwinner idempotently on every replay. Do not guard that cleanup with if not handle.settled: a loser may already be terminal when the workflow replays, but explicit cancellation also releases it from structured child ownership.

Typed service boundaries

Declare a remote function stub to keep normal Python call signatures across an RPC boundary. The stub body is never executed by the caller.

@app.remote("billing", name="capture")
def capture(order: Order, *, idempotency_key: str) -> Receipt: ...

@app.workflow
async def checkout(order: Order) -> Receipt:
    return await capture(order, idempotency_key=order.id)

The billing service registers a local @billing.step(name="capture") with the same argument contract. The envelope is JSON and language-neutral; future SDKs can produce the same wire shape without copying Python's ambient execution mechanism.

Cron schedules

Engine-owned schedules keep firing and deduplicate occurrences while every worker is down. Stack the App-owned schedule declaration over a workflow:

@app.schedule("0 6 * * *", input={"report": "kpi"})
@app.workflow(name="reports.daily")
async def reports_daily(param: dict[str, str]) -> Report:
    occurrence = ogha.scheduled_time()
    return await build_report(param["report"], occurrence.isoformat())

To change a live declaration, change its fields and bump revision=. Worker startup converges to the highest revision.

@app.schedule is not the same as @app.step. A schedule asks the engine to create root workflow Runs on cron time, including while workers are offline. A Step creates child work only when an executing workflow calls it.

Migrating to 0.3

Version 0.3.0 intentionally removes the old public Context authoring surface and renames public execution="async_sticky" to execution="async". It also replaces the module-level schedule decorator's context= payload with @app.schedule(..., input=...).

The complete mechanical mapping is:

Before 0.3	Python 0.3
@ogha.step / @ogha.workflow	@app.step / @app.workflow
ctx.call(step, value)	step(value) and await its Handle
ctx.spawn(workflow, value)	workflow(value); use workflow.detach(value) only for independent work
ctx.join, ctx.sleep, ctx.wait, ctx.gate, ctx.emit, ctx.cancel	ogha.gather/race/quorum, ogha.sleep, ogha.signal, ogha.approval, ogha.event, ogha.cancel
Worker(...) and global registry	app.worker() or app.serve()
@ogha.scheduled(..., context=value)	@app.schedule(..., input=value)
Client.create_schedule/update_schedule(..., context=value)	Client.create_schedule/update_schedule(..., input=value)
scheduled_time(ctx) / ctx.info	scheduled_time() / ogha.info()
execution="async_sticky"	execution="async"

Do not roll new code onto the same target while old Runs are active. Current worker definition routing is by workflow name, not by (name, version). Deploy 0.3 workers to a new target or workflow name, send new Runs there, and keep the old worker fleet until its Runs drain. This is an application cutover rule; the server and wire protocol are unchanged.

Run metadata is available as an immutable value through ogha.info(). For example, ogha.info().run_id replaces reading execution identity from a Context object.

Documentation

• Overview: https://coding2fun.in/ogha
• Getting started: https://coding2fun.in/ogha/python
• Source: https://github.com/vedhlabs/sdk-python

License

Apache License 2.0. See LICENSE.