# Coming from Celery

> Move a Celery task module to Pyrula workflows and call it from sync code with the blocking client: a concept map, the scaffold and an executed example.

Source: https://docs.pyrula.io/getting-started/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

```bash
pyrula init --celery
python app.py
```

The scaffold doesn't import Celery ([Start from a scaffold](https://docs.pyrula.io/getting-started/existing-app.md#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`](https://docs.pyrula.io/reference/workflows.md#workflowworker) | Operational change |
| `worker_concurrency` | [`WorkerOptions(max_concurrent_runs=)`](https://docs.pyrula.io/reference/workflows.md#workeroptions), 50 by default | Operational change |
| `task.delay()`, `apply_async()` | `client.submit(name, params)` on the blocking [`WorkflowClient`](https://docs.pyrula.io/reference/workflows-sync.md#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`](https://docs.pyrula.io/reference/workflows.md#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.

```python
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": "user-42@example.com", "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](https://docs.pyrula.io/workflows/guides/client.md#from-sync-code)).