Migrating from Celery

Side-by-side concept mapping and step-by-step migration from Celery to flexiq.

This guide maps Celery concepts to their flexiq equivalents. If you're coming from Celery, you'll find that most concepts translate directly — with less infrastructure and simpler configuration.

Concept mapping

Celery	flexiq	Notes
`Celery()` app	`Queue()`	`Queue(db_path=...)` replaces `Celery(broker=..., backend=...)` — no broker URL, no result-backend URL, no `app.conf` setup.
`@app.task`	`@queue.task()`	Same decorator shape; `max_retries`, `rate_limit`, `bind=True`, `name=` carry over with identical names.
`.apply_async()`	`.apply_async()`	`priority` and `queue` carry over. Celery's `countdown` becomes `delay`; there is no `eta` (use `delay` with relative seconds). flexiq adds `unique_key`, `expires`, `depends_on`.
`.delay()`	`.delay()`	Drop-in. Returns `JobResult` instead of `AsyncResult`.
`AsyncResult`	`JobResult`	`.result(timeout=...)` replaces `.get(timeout=...)`; async variant `await job.aresult()`.
Canvas (`chain`, `group`, `chord`)	`chain`, `group`, `chord`	API-compatible. flexiq adds a DAG `Workflow` builder with conditions, gates, and fan-in.
`celery beat`	`@queue.periodic()`	No separate beat daemon. Cron expressions support seconds (6 fields) and live in the worker process.
Result backend (Redis/DB)	Built-in (SQLite)	Results store in the same SQLite file as jobs. Per-job `result_ttl_ms` controls retention.
Broker (Redis/RabbitMQ)	Not needed	A single SQLite (or Postgres) connection replaces the broker. WAL mode allows concurrent reads.
`celery worker`	`flexiq worker`	`--concurrency`, `--queues`, `--loglevel` carry over. flexiq adds `--pool prefork` for CPU parallelism.
`celery inspect`	`flexiq info`	Covers `inspect active/scheduled/registered`. `flexiq dashboard` is the live UI version.

Side-by-side examples

App setup

python

from celery import Celery

app = Celery(
    "myapp",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
app.conf.task_serializer = "json"
app.conf.result_serializer = "json"

python

from flexiq import Queue

queue = Queue(db_path="myapp.db")
# That's it. No broker, no backend, no serializer config.

Task definition

python

@app.task(bind=True, max_retries=3)
def send_email(self, to, subject, body):
    try:
        do_send(to, subject, body)
    except SMTPError as exc:
        raise self.retry(exc=exc, countdown=60)

python

@queue.task(max_retries=3, retry_backoff=2.0, retry_on=[SMTPError])
def send_email(to, subject, body):
    do_send(to, subject, body)
    # Retries happen automatically on matching exceptions.
    # Use retry_on/dont_retry_on for selective retries.

Automatic retries

In Celery, you must explicitly catch exceptions and call self.retry(). In flexiq, any unhandled exception triggers a retry automatically (up to max_retries).

Enqueueing tasks

python

# Simple
send_email.delay("user@example.com", "Hello", "World")

# With options
send_email.apply_async(
    args=("user@example.com", "Hello", "World"),
    countdown=60,       # delay in seconds
    queue="emails",
    priority=5,
)

python

# Simple
send_email.delay("user@example.com", "Hello", "World")

# With options
send_email.apply_async(
    args=("user@example.com", "Hello", "World"),
    delay=60,            # delay in seconds
    queue="emails",
    priority=5,
)

The only change: countdown becomes delay.

Getting results

python

result = send_email.delay("user@example.com", "Hi", "Body")

# Block for result
value = result.get(timeout=30)

# Check status
result.status  # "PENDING", "SUCCESS", "FAILURE"

python

job = send_email.delay("user@example.com", "Hi", "Body")

# Block for result
value = job.result(timeout=30)

# Check status
job.status  # "pending", "running", "complete", "failed", "dead"

Key differences:

.get() becomes .result()
Status values are lowercase
"SUCCESS" becomes "complete"

Workflows (canvas)

python

from celery import chain, group, chord

# Chain
chain(fetch.s(url), parse.s(), store.s()).apply_async()

# Group
group(process.s(item) for item in items).apply_async()

# Chord
chord(
    [download.s(url) for url in urls],
    merge.s()
).apply_async()

python

from flexiq import chain, group, chord

# Chain
chain(fetch.s(url), parse.s(), store.s()).apply()

# Group
group(process.s(item) for item in items).apply()

# Chord
chord(
    [download.s(url) for url in urls],
    merge.s()
).apply()

Almost identical. The only change: .apply_async() becomes .apply().

`.apply()` is not Celery's eager `.apply()`

In Celery, .apply() runs the canvas locally and synchronously. In flexiq, .apply(queue) submits it to the queue for workers to run — await results with .result(). flexiq has no eager local execution mode.

Periodic tasks

python

# celery.py
app.conf.beat_schedule = {
    "cleanup-every-hour": {
        "task": "myapp.cleanup",
        "schedule": crontab(minute=0),
    },
}

# Requires a separate process:
# celery -A myapp beat

python

@queue.periodic(cron="0 0 * * * *")
def cleanup():
    ...

# No separate process — the worker handles scheduling.
# flexiq worker --app myapp:queue

flexiq uses 6-field cron expressions (with seconds). Celery's crontab() maps to the last 5 fields, with 0 prepended for seconds.

Celery `crontab()`	flexiq cron
`crontab()` (every minute)	`0 * * * * *`
`crontab(minute=0)` (every hour)	`0 0 * * * *`
`crontab(minute=0, hour=0)` (daily)	`0 0 0 * * *`
`crontab(minute=30, hour=9, day_of_week='1-5')`	`0 30 9 * * 1-5`

Rate limiting

python

@app.task(rate_limit="100/m")
def call_api(endpoint):
    ...

python

@queue.task(rate_limit="100/m")
def call_api(endpoint):
    ...

Identical syntax.

Worker

bash

celery -A myapp worker --loglevel=info -Q emails,default

bash

flexiq worker --app myapp:queue --queues emails,default

Testing

python

# Celery has CELERY_ALWAYS_EAGER mode
app.conf.task_always_eager = True
app.conf.task_eager_propagates = True

result = add.delay(2, 3)
assert result.get() == 5

python

with queue.test_mode() as results:
    add.delay(2, 3)
    assert results[0].return_value == 5

flexiq's test mode uses a context manager instead of a global setting, so it's safe to use in parallel test runs.

What flexiq doesn't have

Some Celery features don't have flexiq equivalents:

Celery feature	Status in flexiq
Distributed workers (multi-server)	Use Postgres backend
Message routing (exchanges, topics)	Use named queues instead
`celery multi` (process management)	Use systemd, supervisor, or Docker
Custom serializers (JSON, msgpack)	`SmartSerializer` (default), `JsonSerializer`, `CloudpickleSerializer`, or custom `Serializer` protocol
Task cancellation (mid-execution)	Cancel pending or running jobs (`cancel_running_job()` + `check_cancelled()`)
ETA (absolute datetime scheduling)	Use `delay` (relative seconds)
`bind=True` (self argument)	Use `current_job` context instead
Custom result backends	Built-in SQLite or Postgres

Migrating from Celery

Concept mapping

Side-by-side examples

App setup

Task definition

Enqueueing tasks

Getting results

Workflows (canvas)

Periodic tasks

Rate limiting

Worker

Testing

What flexiq doesn't have

Migration checklist

Concept mapping

Side-by-side examples

App setup

Task definition

Enqueueing tasks

Getting results

Workflows (canvas)

Periodic tasks

Rate limiting

Worker

Testing

What flexiq doesn't have

Migration checklist