Ogha — durable execution for Python

Write ordinary Python. It survives crashes, restarts, deploys, and month-long waits. You split your code into steps (units of real work) and workflows (compositions of them); Ogha records every step and replays it faithfully after any failure, so a multi-step process finishes even if a worker dies halfway through. No retry loops, no idempotency tables, no state machines to hand-build.

Alpha. The API is stable in shape (nine methods); details may still change before 1.0.

import ogha

@ogha.step(retry=ogha.RetryPolicy(max_attempts=5), pivot=True)
def charge(order, amount):
    return payments.charge(order["card"], amount)   # runs once; recorded

@ogha.workflow
async def checkout(ctx, order):
    total   = await ctx.call(validate, order)
    receipt = await ctx.call(charge, order, total)  # survives a crash right here
    return receipt

Install

pip install ogha          # Python 3.10+

Run the engine

Ogha ships as one container that carries the engine, storage, and a live dashboard. Pull it (public, no login) 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 docker compose (engine + PostgreSQL) is in the getting-started guide below.

Upgrading an existing project is one script — it verifies PyPI, the docs' pinned compose, and the registry agree on the release, then swaps the engine (health-checked, auto-rollback) and optionally this SDK in the same move:

curl -fsSL -O https://coding2fun.in/ogha/update.sh && chmod +x update.sh
./update.sh --apply --sdk .venv/bin/python

The nine methods

Everything a workflow can do is nine methods on ctx:

Retries, timeouts, and saga compensation are declared on the step (retry=, compensate_with=, pivot=), not scattered through the workflow.

A worker

A worker is a long-running process that executes your steps; importing your module registers them.

import ogha, myapp   # importing registers @step / @workflow
client = ogha.connect("http://localhost:8080")
ogha.Worker(client, target="python://checkout", concurrency=8).run()

Submit work and wait for it:

run = client.submit("checkout", data, target="python://checkout")
result = client.result(run.run_id)      # blocks until terminal

Cron schedules

Run a workflow on a schedule the engine fires — occurrences keep firing and dedupe by their deterministic run id even while every worker is down:

@ogha.scheduled("0 6 * * *", context={"report": "kpi"})   # 5-field cron, UTC
@ogha.workflow(name="reports.daily")
async def reports_daily(ctx, param):
    at = ogha.scheduled_time(ctx)        # this occurrence, replay-safe
    return await ctx.call(build_report, param["report"], at.isoformat())

The worker serving the workflow creates the schedule at startup. To change a live schedule, edit the fields and bump revision= — the update is revision-guarded, so a rolling deploy converges instead of flapping:

@ogha.scheduled("0 7 * * *", context={"report": "kpi"}, revision=2)

Programmatic management mirrors it: client.create_schedule(...), client.update_schedule(..., revision=n), client.get_schedule(id), client.list_schedules(), client.delete_schedule(id).

Documentation

• Overview & the nine methods: https://coding2fun.in/ogha
• Getting started (full example): https://coding2fun.in/ogha/python

License

See the project for licensing.