Coming from Celery

A Celery task becomes a Pyrula @workflow, and each side effect inside it becomes a @step. If a worker dies mid-task, another worker claims the run, replays the finished steps from the journal and runs the rest. Callers stay plain functions: on pyrula.workflows.sync.WorkflowClient, submit() queues a run like task.delay() and result() blocks like AsyncResult.get(). One store is queue and result backend, and the worker fires schedules itself, so there is no beat process.

Scaffold

pyrula init --celery
python app.py

The scaffold doesn’t import Celery (Start from a scaffold).

Concept map

Celery	Pyrula	Change
@app.task	@workflow, with @step around each side effect	Small rewrite
Celery(broker=, backend=)	open_store(...): one store is queue and results	Direct swap
celery -A proj worker	pyrula worker --workflow app, or a WorkflowWorker	Operational change
worker_concurrency	WorkerOptions(max_concurrent_runs=), 50 by default	Operational change
task.delay(), apply_async()	client.submit(name, params) on the blocking WorkflowClient	Direct swap
countdown=, eta=	submit(delay=), submit(start_at=)	Direct swap
task_id=	submit(idempotency_key=); a repeat returns the first run	Small rewrite
AsyncResult(id), .state	client.handle(name, run_id), handle.status()	Direct swap
AsyncResult.get(timeout=)	handle.result(timeout=); a failed run raises DurableInvocationError	Small rewrite
autoretry_for, max_retries=N	@step(retry=RetryPolicy(retryable=(E,), max_attempts=N + 1))	Small rewrite
retry_backoff, retry_jitter	RetryPolicy initial_backoff, max_backoff, jitter	Direct swap
self.retry(exc=...)	Raise; Transient retries, Poison and Fatal never do	Small rewrite
Re-running a failed task	handle.retry() or pyrula runs retry	Direct swap
revoke(id)	handle.cancel() at the next step boundary	Direct swap
revoke(id, terminate=True)	handle.terminate(); a running step may finish its effects	Direct swap
chain(a.s(), b.s())	Await one step or ctx.run_child after the other	Conceptual redesign
group(...), chord(...)	await ctx.gather(...) in one run; ctx.run_child across workers	Conceptual redesign
celery beat, crontab	store.schedule_create(Schedule.create(..., schedule_id="digest")); keep the id fixed	Operational change
rate_limit="10/s"	@workflow(rate_limit=Rate(key=..., per_second=10)), shared by every worker	Small rewrite
time_limit, soft_time_limit	@step(timeout=) cancels async code; nothing hard-kills a running step	Not supported
acks_late, visibility_timeout	Durable by default; a dead worker’s run is reclaimed when its lease expires	Operational change
Queues and routing	None; separate fleets use separate store namespaces	Not supported
priority=, expires=	None; submit(owner=) gives tenants fair turns	Not supported
task_always_eager	WorkflowTestEnv from pyrula.workflows.testing	Direct swap
Flower	pyrula dev app.py --open, pyrula runs list, Pyrula Cloud	Operational change

Example

The task fails once and its RetryPolicy runs it again. The caller is plain sync code; the worker runs in a thread only to keep the example in one file.

import asyncio
import tempfile
import threading
from datetime import timedelta

from pyrula.workflows import (
    RetryPolicy,
    StepContext,
    WorkerOptions,
    WorkflowContext,
    WorkflowWorker,
    open_store,
    step,
    workflow,
)
from pyrula.workflows.sync import WorkflowClient


# @app.task(autoretry_for=(ConnectionError,), max_retries=2) becomes a retried step.
@step(retry=RetryPolicy(max_attempts=3, initial_backoff=0.05, retryable=(ConnectionError,)))
def send_email(ctx: StepContext, address: str) -> dict:
    if ctx.attempt == 1:
        raise ConnectionError("provider unreachable")
    return {"sent_to": address, "attempts": ctx.attempt}


# The task body; its registered name is what callers submit.
@workflow(name="docs_from_celery_welcome")
async def welcome(ctx: WorkflowContext, user_id: int) -> dict:
    address = await ctx.step("load_address", lambda: f"user-{user_id}@example.com")
    return await send_email(address)


# Broker and result backend in one. A fresh file per run of this example; an app keeps one path.
store = open_store(default=f"sqlite:///{tempfile.mkdtemp()}/welcome.db")
client = WorkflowClient(store)  # one per process, shared between threads

# `celery worker`: in production, `pyrula worker --workflow app` in its own process.
worker = WorkflowWorker(
    workflows=[welcome],
    store=store,
    options=WorkerOptions(claim_poll_interval=0.05, timer_poll_interval_s=0.1),
)
thread = threading.Thread(target=lambda: asyncio.run(worker.run()), daemon=True)
thread.start()


def enqueue_welcome(user_id: int):
    # send_welcome.apply_async((user_id,), countdown=1, task_id=f"welcome:{user_id}")
    return client.submit(
        "docs_from_celery_welcome",
        {"user_id": user_id},
        delay=timedelta(seconds=1),
        idempotency_key=f"welcome:{user_id}",
    )


try:
    run = enqueue_welcome(42)
    assert enqueue_welcome(42).run_id == run.run_id  # a repeated key returns the first run
    assert run.result(timeout=30) == {"sent_to": "[email protected]", "attempts": 2}
finally:
    worker.stop()
    thread.join(timeout=10)
    client.close()

Under Celery prefork or gunicorn --preload, create the client at import but don’t call it before the fork, and open a Valkey store after it (From sync code).