KAHIBARO
Discord Login Register ☰
☰

17.4. Celery

Table of Contents

Why Use Celery?

Backend applications often need to perform work that is too slow or too heavy to run inside the normal HTTP request. Examples:

If you try to do these tasks inside the request handler, your API will feel slow and may even time out. You also cannot easily retry failures or spread work across multiple machines.

Celery is a Python library that solves this problem. It lets you:

Celery uses a message broker such as Redis or RabbitMQ to send jobs from your web app to worker processes.

Celery’s Core Concepts

To understand Celery, remember three main parts:

PartRole
ProducerCode that sends tasks, often your web application
BrokerMessage system that stores tasks, for example Redis
WorkerCelery process that reads tasks from broker and executes them

Celery also has some extra helpers:

Key idea: Your web app sends tasks to the broker, workers process them in the background, and your app can continue handling requests without waiting.

Installing Celery

Celery is a Python package, so you usually install it into your virtual environment.

bash 
pip install celery

If you want to use Redis as a broker and result backend:

bash 
pip install "celery[redis]"

You will also need a running Redis or RabbitMQ instance. The details of installing Redis are in the Redis chapter, but a typical local development command (with Docker) might look like:

bash 
docker run -d -p 6379:6379 --name redis redis:7

Creating a Basic Celery Application

Celery apps are usually defined in a file, often called celery_app.py or inside your main package.

A minimal setup using Redis as broker and result backend:

python 
# celery_app.py
from celery import Celery
celery_app = Celery(
    "my_app",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
# optional configuration
celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="UTC",
    enable_utc=True,
)

Explanation:

You can keep these values in environment variables and read them from there, instead of hardcoding them.

Defining Celery Tasks

A Celery task is a normal Python function that is registered as a task. The simplest way is to use the @celery_app.task decorator.

python 
# tasks.py
from time import sleep
from .celery_app import celery_app
@celery_app.task
def add(x, y):
    sleep(2)  # simulate slow work
    return x + y

You can also define tasks with more control:

python 
@celery_app.task(name="math.add", bind=True)
def add_bound(self, x, y):
    # self is the task instance, you can access things like self.request.id
    return x + y

Some typical backend tasks:

python 
@celery_app.task
def send_verification_email(user_id: int):
    # lookup user in database
    # generate verification link
    # send email
    return f"Verification email sent to user {user_id}"
@celery_app.task
def resize_image(image_id: str):
    # load image from storage
    # run resize operation
    # save resized image
    return f"Image {image_id} resized"

Running Celery Workers

Once you have a Celery app and tasks, you need worker processes to execute them. In your project root:

bash 
celery -A celery_app.celery_app worker --loglevel=info

Breakdown:

If your Celery app is in a package, the path might look like:

bash 
celery -A myproject.celery_app.celery_app worker --loglevel=info

You can control how many concurrent tasks a worker can process:

bash 
celery -A celery_app.celery_app worker --loglevel=info --concurrency=4

This creates up to 4 worker processes or threads, depending on your pool configuration.

Calling Tasks Asynchronously

You do not call Celery tasks like normal functions if you want them to run in the background. Instead you call them through one of Celery’s methods.

The most common method is .delay():

python 
from tasks import add
result = add.delay(2, 3)  # returns immediately
print(result)             # <AsyncResult: some-task-id>

You can also use .apply_async() for more control:

python 
result = add.apply_async(
    args=[2, 3],
    countdown=10,  # run after 10 seconds
)

Typical usage inside a web endpoint:

python 
def create_user(user_data):
    user = save_user_to_db(user_data)
    # send email in background
    send_verification_email.delay(user.id)
    return {"message": "User created, verification email will be sent"}

The HTTP response is sent immediately. The Celery worker will handle the email later.

Checking Task Results and Status

If you have a result backend configured, every async call returns an AsyncResult object. It lets you check the task state and result.

python 
result = add.delay(2, 3)
# check status
print(result.id)      # task id
print(result.state)   # PENDING, STARTED, SUCCESS, FAILURE, etc.
# wait for completion (blocks)
value = result.get(timeout=10)
print(value)  # 5

Useful properties and methods:

Property / MethodMeaning
result.idUnique ID of this task
result.stateCurrent state, for example SUCCESS
result.ready()Returns True if completed
result.successful()Returns True if completed successfully
result.get()Returns result or raises error

In real APIs, you often:

  1. Start a long-running job, return its task_id to the client.
  2. Provide another endpoint that checks the status of a task by id.

Retrying and Handling Failures

Celery has built-in support for retries. For example, when talking to flaky external services you may want to retry a few times.

To use retries, define a bound task with bind=True and call self.retry.

python 
from celery.utils.log import get_task_logger
from .celery_app import celery_app
logger = get_task_logger(__name__)
@celery_app.task(bind=True, max_retries=3, default_retry_delay=5)
def call_external_api(self, url: str):
    try:
        # for example only, use requests or httpx in real code
        if some_random_failure_condition():
            raise ConnectionError("Temporary failure")
        return "ok"
    except Exception as exc:
        logger.warning("API call failed, retrying...")
        raise self.retry(exc=exc)

Explanation:

You can also configure retry settings when calling the task:

python 
call_external_api.apply_async(
    args=["https://example.com"],
    retry=True,
    retry_policy={
        "max_retries": 5,
        "interval_start": 0,
        "interval_step": 2,
        "interval_max": 10,
    },
)

Scheduling Tasks and Celery Beat

Sometimes you need to run tasks periodically, for example:

You can treat periodic tasks as background jobs that are triggered on a schedule. Celery provides Celery Beat, a scheduler process that creates tasks at defined intervals.

A simple in-code schedule:

python 
# celery_app.py
from celery import Celery
from celery.schedules import crontab
celery_app = Celery(
    "my_app",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
celery_app.conf.beat_schedule = {
    "cleanup-every-hour": {
        "task": "tasks.cleanup_temp_files",
        "schedule": crontab(minute=0),  # at every hour
    },
    "ping-every-10-seconds": {
        "task": "tasks.ping",
        "schedule": 10.0,  # seconds
    },
}

Tasks:

python 
# tasks.py
from .celery_app import celery_app
@celery_app.task
def cleanup_temp_files():
    # delete temporary files, expired tokens, etc.
    return "cleanup done"
@celery_app.task
def ping():
    print("ping")

Start beat in one process and worker in another:

bash 
# Terminal 1: worker
celery -A celery_app.celery_app worker --loglevel=info
# Terminal 2: beat
celery -A celery_app.celery_app beat --loglevel=info

Beat sends scheduled tasks to the broker. The worker executes them.

Configuring Celery

Celery has many configuration options. You can set them through celery_app.conf.update(...) or by using a dedicated config module.

Common configuration examples:

python 
celery_app.conf.update(
    broker_url="redis://localhost:6379/0",
    result_backend="redis://localhost:6379/1",
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="UTC",
    enable_utc=True,
    # task routes
    task_routes={
        "tasks.send_email": {"queue": "emails"},
        "tasks.generate_report": {"queue": "reports"},
    },
    # visibility timeout with Redis broker
    broker_transport_options={"visibility_timeout": 3600},
)

You can route different tasks to different queues and start workers that only handle specific queues:

bash 
celery -A celery_app.celery_app worker -Q emails --loglevel=info

Common Celery Patterns in Backend Apps

Here are some realistic patterns that you will often use.

1. Fire-and-forget

You do not care about the result, you only want the task to run later.

python 
send_verification_email.delay(user_id)

2. Start job, poll for status

You start a long job and return its ID, then the client checks on it.

python 
# start job
result = generate_report.delay(report_params)
return {"task_id": result.id}
# check job
from celery.result import AsyncResult
def get_task_status(task_id: str):
    res = AsyncResult(task_id, app=celery_app)
    return {"state": res.state}

3. Chaining tasks

You want tasks to run one after another.

python 
from celery import chain
from tasks import upload_file, process_file, notify_user
workflow = chain(
    upload_file.s("/path/to/file"),
    process_file.s(),
    notify_user.s(user_id),
)
workflow.delay()

4. Parallel tasks and joining results

Run multiple tasks in parallel, then process all results.

python 
from celery import group
from tasks import resize_image, create_thumbnail
job = group(
    resize_image.s("image1"),
    resize_image.s("image2"),
    create_thumbnail.s("image1"),
)
result = job.apply_async()
all_results = result.get()  # list of individual results

Basic Celery Project Structure Example

Here is a simple directory layout including Celery and a web backend:

text 
myproject/
    app/
        __init__.py
        main.py          # your FastAPI / Flask app
        celery_app.py    # Celery instance
        tasks.py         # Celery tasks
    requirements.txt

celery_app.py:

python 
from celery import Celery
celery_app = Celery(
    "myproject",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)
celery_app.autodiscover_tasks(["app"])

tasks.py:

python 
from .celery_app import celery_app
@celery_app.task
def send_welcome_email(user_id: int):
    # lookup user, send email
    return f"Welcome email sent to user {user_id}"

Your web handler in main.py:

python 
from fastapi import FastAPI
from .tasks import send_welcome_email
app = FastAPI()
@app.post("/register")
def register_user():
    # create user in database
    user_id = 42  # example
    send_welcome_email.delay(user_id)
    return {"message": "User registered"}

Best Practices When Using Celery

Some important rules that help you avoid common problems:

  • Do not pass huge objects or database models as task arguments, pass IDs or small data instead.
  • Keep tasks small and focused, a task should ideally do one thing.
  • Handle expected errors and use retries for temporary failures.
  • Use environment variables for broker and backend configuration.
  • Separate queues for different kinds of work if you have many tasks.
  • Monitor your Celery workers and queues in production.

Following these practices helps keep your background processing reliable and easier to debug.

Views: 8

Comments

Please login to add a comment.

Don't have an account? Register now!

© 2025 Kahibaro
Visitors: 3488 / 24h
Impressum / Imprint
Datenschutz / Data Policy