Runs started by events

Webhooks are delivered at least once: the sender retries until it sees a 2xx, so the same event can arrive twice. The route that receives one does three things: check it came from the sender, submit a run with an idempotency key derived from the event’s id, and return quickly. submit only queues the run, so the response does not wait for the work.

The route below sits beside the one pyrula init --fastapi writes (Start from a scaffold); in that file, call worker.client.submit instead of building a second client. This version builds its own client so it runs on its own.

import hashlib
import hmac
import json
import tempfile

from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.testclient import TestClient
from pyrula.workflows import WorkflowClient, open_store, workflow

SECRET = b"example-signing-secret"  # from your secret manager, never the source


@workflow(name="docs_recipe_process_document")
async def process_document(ctx, document_id: str) -> dict:
    return await ctx.step("extract", lambda: {"document_id": document_id})


# A fresh file per run of this example; an app keeps one store path for good.
store = open_store(default=f"sqlite:///{tempfile.mkdtemp()}/events.db")
client = WorkflowClient(store)
app = FastAPI()


def verify(body: bytes, signature: str) -> None:
    # Most senders sign the raw body with a shared secret; follow your sender's scheme exactly.
    expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="Unauthorized")


@app.post("/webhooks/documents", status_code=202)
async def document_received(request: Request, x_signature: str = Header(default="")) -> dict:
    body = await request.body()
    verify(body, x_signature)
    try:
        event = json.loads(body)
        event_id, document_id = event["id"], event["document"]["id"]
    except (ValueError, KeyError, TypeError) as err:
        raise HTTPException(status_code=400, detail="Bad event") from err
    run = await client.submit(
        "docs_recipe_process_document",
        {"document_id": document_id},
        idempotency_key=f"document-event:{event_id}",
    )
    return {"run_id": run.run_id}


def post(http: TestClient, event: dict, secret: bytes = SECRET):
    body = json.dumps(event).encode()
    signature = hmac.new(secret, body, hashlib.sha256).hexdigest()
    return http.post("/webhooks/documents", content=body, headers={"X-Signature": signature})


with TestClient(app) as http:
    event = {"id": "evt_1", "document": {"id": "doc_9"}}
    first = post(http, event).json()
    assert post(http, event).json() == first  # the sender retried: same run
    assert post(http, {**event, "id": "evt_2"}).json() != first  # a new event: a new run
    assert post(http, event, secret=b"wrong").status_code == 401
    assert http.post("/webhooks/documents", content=b"{}", headers={"X-Signature": hmac.new(
        SECRET, b"{}", hashlib.sha256).hexdigest()}).status_code == 400

A key lives 24 hours. A redelivery after that starts a second run, and within the window a repeat returns the first run even if that run failed (Starting and talking to runs). If a sender retries for longer than that, record handled event ids yourself.

What to use as the key

The key must be the same for every delivery of one event and different for different events.

Source	Key from	Check
A SaaS webhook	The event id in the payload or headers	The sender’s docs say whether the id is stable across retries
Inbound email through a mail-to-webhook service	The message’s Message-ID header, or the service’s own event id	Redelivery by the service keeps the same one
S3 notification	Bucket, object key and version id (sequencer on an unversioned bucket)	S3 notifications reach HTTP through SNS or EventBridge; the record has no single event id

This table is a starting point, not a tested integration: Pyrula has no S3 or email code, and the claims in the last column are the senders’ to confirm. A key made of fields that identify the work, such as invoice:{invoice_id}:{version}, is as good as an event id and survives a sender that mints a new event id per retry.

Django, Flask and other sync code

A sync view uses the blocking client in pyrula.workflows.sync. Create it once per process and share it between threads (From sync code). The function that starts the run is the same whatever the framework:

import tempfile

from pyrula.workflows import open_store
from pyrula.workflows.sync import WorkflowClient

store = open_store(default=f"sqlite:///{tempfile.mkdtemp()}/sync-events.db")
client = WorkflowClient(store)


def start_from_event(event: dict) -> str:
    run = client.submit(
        "docs_recipe_process_document",
        {"document_id": event["document"]["id"]},
        idempotency_key=f"document-event:{event['id']}",
    )
    return run.run_id


event = {"id": "evt_1", "document": {"id": "doc_9"}}
assert start_from_event(event) == start_from_event(event)
client.close()

Called from a Flask view, with the signature check done first:

from flask import Flask, abort, jsonify, request

app = Flask(__name__)


@app.post("/webhooks/documents")
def document_received():
    if not signature_is_valid(request.get_data(), request.headers.get("X-Signature", "")):
        abort(401)
    return jsonify(run_id=start_from_event(request.get_json())), 202

signature_is_valid is your check, like verify above. A Django view is the same call. The sync client cannot be used from a thread running an event loop: use the async client there.

What this does not do

• No Pyrula code reads a bucket, a mailbox or a queue. A trigger that polls one belongs in a worker process or a scheduled run that calls submit. For Kafka, see Kafka sources.
• The route is yours, so its authentication is too. The run endpoints that RunRouter serves take their own AuthConfig (Authentication).