Migrating from Celery

If you already run Celery in production, you have a working solution to a real problem. Celery has shipped a decade of background work in the Python ecosystem, and the question on this page is not whether Celery is good. The question is what Flux actually changes for you, what it does not, and how to port code from one to the other without breaking anything in production.

The honest case for moving: a Celery job that fails mid-run re-queues from scratch. If your jobs are short and idempotent, that re-run is free and you can stop reading. If you have multi-step pipelines, long-running jobs, or workflows you want to pause and resume, Flux records every step in an event log and resumes at the last completed task instead. Durable execution is the reason to move.

Mental-model translation

Code-pattern translation

A task definition.

# Celery
from celery import Celery

app = Celery("jobs", broker="redis://localhost:6379/0")

@app.task(autoretry_for=(IOError,), retry_kwargs={"max_retries": 3})
def fetch_url(url: str) -> str:
    return requests.get(url, timeout=10).text

# Flux
from flux.task import task

@task.with_options(retry_max_attempts=3)
async def fetch_url(url: str) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=10)
        return response.text

The decorator changes, the retry surface becomes declarative options, and the body becomes async. The task is invoked from a workflow rather than from anywhere in your application.

A pipeline of dependent steps.

# Celery — chain of three tasks
from celery import chain

result = chain(
    extract.s(source_url),
    transform.s(),
    load.s(target_url),
)()

# Flux — ordinary await chain inside a workflow
from flux import ExecutionContext
from flux.workflow import workflow

@workflow
async def etl_pipeline(ctx: ExecutionContext[dict]):
    raw = await extract(ctx.input["source_url"])
    cleaned = await transform(raw)
    return await load(cleaned, ctx.input["target_url"])

The Flux version is a normal Python function. If transform succeeds and load crashes, the next worker reads the event log, sees extract and transform already completed, and resumes at load with the recorded intermediate values.

Periodic work.

# Celery beat
app.conf.beat_schedule = {
    "daily-report": {
        "task": "jobs.send_daily_report",
        "schedule": crontab(hour=3, minute=0),
    },
}

# Flux
from flux import cron, ExecutionContext
from flux.workflow import workflow

@workflow.with_options(name="daily_report", schedule=cron("0 3 * * *", timezone="UTC"))
async def daily_report(ctx: ExecutionContext[None]):
    return await send_daily_report()

The schedule lives on the workflow. When the workflow is registered against a Flux server, the schedule is created automatically. Like celery beat run as a singleton, the fire happens once — but Flux enforces that itself: the scheduler cycle is coordinated across server replicas via PostgreSQL advisory locks, so you don’t have to keep a dedicated beat-style process unique by deployment discipline.

Delayed execution.

# Celery
send_reminder.apply_async(args=[user_id], countdown=3600)

# Flux — option A: sleep inside a workflow
from flux.tasks import sleep

@workflow
async def reminder_flow(ctx: ExecutionContext[int]):
    await sleep(3600)
    return await send_reminder(ctx.input)

# Flux — option B: a one-time schedule
from flux import once
from datetime import datetime, timedelta, timezone

@workflow.with_options(
    name="send_reminder",
    schedule=once(run_time=datetime.now(timezone.utc) + timedelta(hours=1)),
)
async def send_reminder_once(ctx: ExecutionContext[int]):
    return await send_reminder(ctx.input)

Pick option A for a delay that’s part of a larger workflow body; pick option B if the delay is the whole job.

Routing work to specific workers.

# Celery — route by queue
@app.task(queue="gpu")
def train_model(config: dict): ...

# Flux — worker affinity
from flux.workflow import workflow

@workflow.with_options(affinity={"gpu": "true"})
async def train_model(ctx: ExecutionContext[dict]):
    return await train(ctx.input)

A worker advertises its labels at startup (flux start worker my-worker --label gpu=true), and the server only dispatches workflows whose affinity clause matches the worker’s labels.

What you give up

Celery’s routing surface is wider than Flux’s affinity model. Per-task priority queues, message-level rate limits, RabbitMQ exchanges, and direct routing keys do not have a one-line equivalent in Flux 0.56.0. The affinity-by-resource pattern covers most cases (GPU workers, region-pinned workers, restricted-network workers), but if your production system relies on, for example, four priority queues with different worker pools and explicit per-message priority, you will rebuild that as an application-layer concern.

The Flux dispatch path adds a server hop. With Celery, the client publishes to the broker and the worker consumes; with Flux, the client calls the server, the server records the execution, and the server pushes work over SSE to a connected worker. For a fire-and-forget workload pushing tens of thousands of small jobs per second, Celery is closer to the metal.

You also give up the option of running zero infrastructure. Flux needs a server process, at least one worker, and a real database. Celery needs a broker and (optionally) a result backend. If you are operating Redis already, “add Celery” is cheaper than “add Flux.” Be clear on whether the durability property is worth the operational delta.

Migration order

Pick one task and port it without ceremony. Wrap it in a @workflow even if the workflow has only one step — the workflow is the unit Flux schedules and records, and you need the surrounding event log to evaluate durability honestly. Switch one trigger point — one cron entry, one API handler, one place in your application that calls task.delay(...) — to call workflow.run(...) (synchronous) or the REST API (asynchronous) instead. Keep the Celery version running side by side for a week. If the Flux version is fine, retire the Celery task.

For schedules, do the same. Move one celery beat entry to a @workflow.with_options(schedule=cron(...)) declaration. Run them in parallel against the same database (with the Flux schedule firing at a slightly different minute to make the runs distinguishable) until you trust the Flux version, then disable the beat entry.

Multi-step Canvas pipelines (chain, group, chord) are usually the highest-payoff migrations because the durability win is concrete: the Celery pipeline re-runs from step 1 on crash, the Flux workflow resumes at the failed step. Port those after you are comfortable with the single-task case.

Do not attempt a big-bang switch. The two systems coexist cleanly — they don’t share infrastructure, they don’t fight for the same database tables, and your application can call both. Coexistence is the migration pattern, not an interim state to feel bad about.

Where to read more

• Concepts — vs. plain Python + queues — the comparative analysis behind this guide.
• Build — Scheduling workflows — the full schedule API.
• Build — Errors and retries — the retry option surface.
• Build — Worker affinity — how affinity maps to worker labels.

Compared against Celery 5.x as of July 2026. Flux 0.56.0.