resonate-http-service-design-python

Resonate HTTP Service Design — Python

Overview

Route handlers in a Python web framework (FastAPI, Flask, Django) are the entrypoint for durable workflows; they start or await invocations via the Resonate client. Actual business logic runs in worker processes that register durable functions. Downstream services (a DB worker, a search worker) expose their own durable functions and are invoked via RPC.

This skill covers the shape of that split: what runs where, how routes interact with workers, and how webhooks resolve durable promises.

Architecture model

 client → HTTP routes (FastAPI/Flask)
            → Resonate client (r.run / r.rpc / r.promises.resolve)
                → worker group A (business logic)
                    → ctx.rpc → worker group B (specialized, e.g. DB)

• HTTP server: FastAPI/Flask/Django app handling routes; maps HTTP requests to Resonate invocations
• Worker service: a process (same or different host) with r.register(...)-ed functions; connects to the Resonate server
• Resonate server: durable promise store + coordination hub (Rust v0.9.x, single port 8001)

Rules

• Route handlers are ephemeral. Use Client APIs (r.run, r.rpc, r.promises.*). Never use Context APIs in a route handler.
• Durable functions run in workers. async def with await ctx.run/rpc + r.register(fn).
• All external effects (DB, HTTP, filesystem) run inside ctx.run envelopes — never at the top of a durable function, never at module import time.
• Stable invocation IDs. Derive from request inputs (f"order:{order_id}") — not uuid.uuid4() at route-handler time unless you persist the ID in the response so the client can poll.

Route design patterns

1. Submit-and-poll (async HTTP)

Client submits a job, gets an ID, polls for status:

from __future__ import annotations
import os, time
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from resonate.resonate import Resonate

app = FastAPI()
r = Resonate(url=os.environ.get("RESONATE_URL", "http://localhost:8001"))


class JobSubmission(BaseModel):
    order_id: str
    items: list[str]


@app.post("/jobs")
async def submit_job(body: JobSubmission) -> dict:
    job_id = f"order:{body.order_id}"
    # Fire and forget — workflow runs durably in the background
    r.options(target="order-workers").rpc(job_id, "process_order", body.order_id, body.items)
    return {"job_id": job_id, "status": "started"}


@app.get("/jobs/{job_id}")
async def get_job(job_id: str) -> dict:
    try:
        record = await r.promises.get(job_id)
        state = record.state if hasattr(record, "state") else "pending"
        if state == "resolved":
            return {"job_id": job_id, "status": "done", "result": record.value}
        elif state == "rejected":
            return {"job_id": job_id, "status": "failed", "reason": record.value}
        else:
            return {"job_id": job_id, "status": "running"}
    except Exception:
        raise HTTPException(status_code=404, detail="job not found")