DEV Community: Mehedee Siddique

Configuring The Celery App

Mehedee Siddique — Mon, 18 Sep 2023 17:07:17 +0000

This post is the 2nd of a multipart series on Celery, find the first one here Introduction to Celery

In this post we'll explore Celery configurations and in the process learn a bit about how tasks should be designed.

To recap, lets first set up the Celery app instance

from celery import Celery


# Broker(RabbitMQ) connection str
CELERY_BROKER: str = (
    f"pyamqp://user:Pass1234@rabbitmq:5672//"
)

# Result Backend(Redis) connection str
CELERY_BACKEND: str = f"redis://redis:6379"


# Celery App instance
celery_app = Celery(
    __name__, broker=CELERY_BROKER, backend=CELERY_BACKEND, include=['app.tasks']
)

Lets talk about some configurations now.

task_acks_late (`boolean`)

celery_app.conf.task_acks_late = False

Disabled by default, meaning the worker acknowledges the task as executed right before it starts execution. So, in any case the worker crashes while executing the task, the task won't be redelivered to the worker again. If this setting is enabled, any task lost because of worker crash while in the middle of task execution will be delivered to the worker again by the broker.
This begs the question, why is the setting not enabled by default? To explain, we need to understand what an ideal task looks like. In an ideal scenario, a task should be idempotent. It means, when we send a task to the broker, no matter how many times the task is delivered to the worker for execution(due to worker crash or task retry on failure), the end result is always the same. Now, it is on us developers to write an idempotent task because Celery can't detect if it is not. Also, in reality, it might not always be possible to write an idempotent task. If we enable task_acks_late on a non-idempotent task, we might face unwanted consequences because of the multiple execution of that task. For this reason, assuming our tasks are not idempotent, Celery keeps it disabled by default. If we make sure that our tasks are idempotent, we can take the benefit of enabling this setting and, it is recommended to do so. Read more on this here
Applying this setting on application level would apply it on all tasks but, we can also specify this setting in task level with @celery_app.task(acks_late=True) parameter which will then take precedence over the app level setting for this task only.
My preference is, we keep this setting disabled as it is by default and, for tasks we are sure to be idempotent and must be redelivered to the worker in case of any crash, we enable this setting in task level. Also, I like to explicitly add this config to the app as disabled to make the codebase more explanatory.

task_ignore_result (`boolean`)

celery_app.conf.task_ignore_result = False

Disabled by default, meaning task results won't be saved to the result backend even if you have the backend configured. Enabling this setting would mean results for all tasks will be saved in the backend.
Saving task result in the backend adds and extra overhead to performance, memory and storage. In my opinion, we don't need results for most tasks and, if your code is mostly dependent on task results, you might not be taking the asynchronous execution benefit of Celery. For example, you are handling a newsletter subscribe post request and, have triggered an email sending task from your web application and, waiting for the task to finish in the worker before moving on further with request. In this case your code will still be behaving as if the email sending task was executed in the current application process, not in a separate worker. Having said that, there will still be many valid cases where keeping and tracking task results will be necessary or required(some Celery Canvas Workflows are dependent on task results).

This setting is also available in task level and gives task level setting precedence for that task.

My preference is to keep the default setting as disabled and enable for specific tasks where it is necessary. We should also check task_store_errors_even_if_ignored.

result_expires (`int(seconds) | timedelta`)

celery_app.conf.result_expires = 300

24 hours by default. After the configured time the result will be deleted from the result backend. We should set the minimum time we need to access the result. Setting a larger value unnecessarily will add performance, memory and storage overheads.

worker_concurrency (`int`)

celery_app.conf.worker_concurrency = 2

Number of CPU cores by default. This setting defines how many concurrent processes the worker will use to execute tasks when concurrency pool is not defined or prefork is used explicitly. If concurrency pool is either eventlet or gevent, then this setting denotes the number of green threads spawned.

Concurrency pool of Celery is a different discussion and there is a nice article about it. The rule of thumb is, For CPU bound tasks use prefork concurrency and for I/O bound tasks, use gevent/eventlet. Don't use more number than available CPU cores with prefork(default) pool but, for gevent/eventlet pool, you can use hundreds or even thousands.

broker_pool_limit (`int | None`)

celery_app.conf.broker_pool_limit = 30

10 connection pools by default. This setting denotes the maximum number of connections can be opened in a connection pool. This number can be allocated depending on how many threads/green-threads(eventlets/gevent) are accessing a connection.

worker_prefetch_multiplier (`int`)

celery_app.conf.worker_prefetch_multiplier = 5

4 by default, meaning the worker will prefetch 4 task messages per concurrency, e.g. if worker_concurrency is 4 and worker_prefetch_multiplier is 5 then, 20 tasks in total will be prefetched by the worker.
Prefetching tasks helps us reduce the time a worker stays idle between completing a task and starting the next one. Also, prefetching too much tasks could cause a performance overhead.

This is mostly all the configs we need in most cases and I hope I didn't miss any. Let me know if you think I should have mentioned any other configs. We should also go through Celery's complete list of configs and play with them a bit to understand better.

See you in the next post.

Introduction to Celery

Mehedee Siddique — Wed, 13 Sep 2023 19:38:53 +0000

Celery is an asynchronous task queue for python. We mostly use it to run tasks outside the cycle of our regular application, e.g. HTTP request-response cycle. Also, we can use Celery to schedule tasks for a specific time or run periodic tasks, as in cron-jobs.

A classic example is sending an email on user signup. You want to send a user an on-boarding email when they’re signing up but that means you’d need to hold the request, post the email, wait for success response for email sending and then, return response to the user . As sending email is a network call, this might take some time which means, for this time, your application and the request sits idly, so does the user, thinking why the hell am I signing up for this shitty application!

Here comes Celery to the rescue. With Celery you can just publish an email sending task to a Celery Worker (a celery application containing your email sending task’s codes, more on this later), finish the signup process and return response to the user, all while the Celery worker is doing the task of sending the email to the user. This way your request-response lifecycle is cut short and everyone is happy.

There are two main and one optional components of Celery:

Broker - as the name suggests, an intermediary between the client application/s and worker/s.

Worker - where the task actually executes.

Result Backend aka. Backend (Optional) - where the results of executed tasks are stored for retrieval.

Broker

The Broker is responsible for receiving task messages from the client application/s, queue them and deliver them to available worker/s. It works as an intermediary. Broker is also responsible for maintaining multiple queues according to your need and priority. It also does the job of redelivering failed tasks to workers for retrying(if configured) and more. The most popular broker used with Celery is RabbitMQ. Redis can also be used as broker.

Worker

The worker is the executor of your tasks. There can be multiple workers running at once. Workers constantly monitor broker for new tasks. Whenever a worker has a free slot, it checks with broker for new tasks and picks them if available. Worker then executes the tasks and checks with broker for new tasks again.

Result Backend

Celery gives you an option to store the result of your task. It is achieved with a result backend. Backend stores the task’s execution result along with the return value from your tasks. Backend is also necessary to design some of the canvas workflows(more on this later). One of the most popular choice for Result Backend is Redis because of it’s super fast Key-Value storage that makes fetching results of a task very efficient. RabbitMQ can also be used as Backend.

Lets jump into some actual codes now. We will use RabbitMQ as broker and Redis as backend. Lets create an app directory inside our content root directory and create a python file worker.py.

First we create connections strings for RabbitMQ broker and Redis backend. Then we pass them to create a celery-app instance.

from celery import Celery


# Broker(RabbitMQ) connection str
CELERY_BROKER: str = (
    f"pyamqp://user:Pass1234@rabbitmq:5672//"
)

# Result Backend(Redis) connection str
CELERY_BACKEND: str = f"redis://redis:6379"


# Celery App instance
celery_app = Celery(
    __name__, broker=CELERY_BROKER, backend=CELERY_BACKEND
)

Our celery application is ready. Now lets make a task for it to execute.

First, we will define a function send_newsletter_welcome_email_task that accepts an email address string and sends a welcome email to that address. Lets import and use smtplib and loguru for email sending and logging. Then we will import the celery_app instance from the worker.py and decorate our email sending function with @celery_app.task().

from loguru import logger
from smtplib import SMTP

from .worker import celery_app


@celery_app.task()
def send_newsletter_welcome_email_task(email: str):
    logger.info(f"Send welcome email task received")
    with SMTP(
            host="smtp.freesmtpservers.com",
            port=25,
            timeout=60,
    ) as smtp:
        from_addr = "newsletter@mehedees.dev"
        smtp.sendmail(
            from_addr=from_addr,
            to_addrs=email,
            msg=f"To:{email}\nFrom: {from_addr}\r\nSubject: Welcome\n\nWelcome to the newsletter!",
        )
        logger.info("Email successfully sent")
    logger.info(f"Send welcome email task finished")
    return email

Our task is ready. Now we will define a mock web app to trigger our task.

from fastapi import FastAPI, Body
from loguru import logger

from .tasks import send_newsletter_welcome_email_task


app = FastAPI(
    debug=True,
    title="Test Python Celery",
    description="Test basics of Python Celery",
    openapi_url=f"/openapi.json",
)


@app.post(path='/newsletter/signup')
async def newsletter_signup(email: str = Body(embed=True)):
    logger.info(f"Received newsletter signup request from {email}")
    # Doing some processing bla bla bla
    logger.info("Initiating welcome email sending task")
    send_newsletter_welcome_email_task.delay(email)
    # Return response now, celery will take care of sending the welcome mail
    return {
        'success': 'True',
        'code': 200,
    }

We have created a quick FastAPI web application with newsletter signup route. We’ve also imported our task here. The view function does some dummy processing and then calls the email sending task.
To call the task we call .delay() on our task: send_newsletter_welcome_email_task.delay(email). When triggered, the task will be sent to the broker(RabbitMQ). Then the broker will deliver the task to the available worker. The worker will then execute the task resulting in the welcome email being sent.

We could also trigger the task with .apply_async([email,]). .delay() is actually a shortcut for .apply_async().

Our code is now almost ready. We have defined the celery app instance, defined a task, created a web app to trigger our task. We have one modification left for the celery app instance, we need to introduce our task to the application beforehand. Otherwise our worker won’t recognize the task that has been delivered to it for execution. Lets do it

from celery import Celery


# Broker(RabbitMQ) connection str
CELERY_BROKER: str = (
    f"pyamqp://user:Pass1234@rabbitmq:5672//"
)

# Result Backend(Redis) connection str
CELERY_BACKEND: str = f"redis://redis:6379"


# Celery App instance
celery_app = Celery(
    __name__, broker=CELERY_BROKER, backend=CELERY_BACKEND
)

# Autodiscovery of defined tasks
celery_app.autodiscover_tasks(packages=['app'])

The last line discovers files named task.py inside the package paths provided to the autodiscover_tasks method.

Our code is ready now. Lets run our Celery Worker with following command

celery -A app.worker.celery_app worker --loglevel=INFO

We will see something like

/usr/local/lib/python3.11/site-packages/celery/platforms.py:829: SecurityWarning: You're running the worker with superuser privileges: this is
2023-09-13T18:28:38.979250505Z absolutely not recommended!
2023-09-13T18:28:38.979365023Z 
2023-09-13T18:28:38.979406969Z Please specify a different user using the --uid option.
2023-09-13T18:28:38.979557797Z 
2023-09-13T18:28:38.979587460Z User information: uid=0 euid=0 gid=0 egid=0
2023-09-13T18:28:38.979641167Z 
2023-09-13T18:28:38.979666256Z   warnings.warn(SecurityWarning(ROOT_DISCOURAGED.format(
2023-09-13T18:28:39.124459733Z  
2023-09-13T18:28:39.124513560Z  -------------- celery@e96b71c99d22 v5.3.1 (emerald-rush)
2023-09-13T18:28:39.124523657Z --- ***** ----- 
2023-09-13T18:28:39.124527360Z -- ******* ---- Linux-5.15.49-linuxkit-x86_64-with 2023-09-13 18:28:39
2023-09-13T18:28:39.124530601Z - *** --- * --- 
2023-09-13T18:28:39.124533566Z - ** ---------- [config]
2023-09-13T18:28:39.124536551Z - ** ---------- .> app:         app.worker:0x7fcc3b5f9cd0
2023-09-13T18:28:39.124539840Z - ** ---------- .> transport:   amqp://user:**@rabbitmq:5672//
2023-09-13T18:28:39.124542931Z - ** ---------- .> results:     redis://redis:6379/
2023-09-13T18:28:39.124546424Z - *** --- * --- .> concurrency: 6 (prefork)
2023-09-13T18:28:39.124549579Z -- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
2023-09-13T18:28:39.124552584Z --- ***** ----- 
2023-09-13T18:28:39.124555464Z  -------------- [queues]
2023-09-13T18:28:39.124558372Z                 .> celery           exchange=celery(direct) key=celery
2023-09-13T18:28:39.124561364Z                 
2023-09-13T18:28:39.124564209Z 
2023-09-13T18:28:39.124567139Z [tasks]
2023-09-13T18:28:39.124570067Z   . app.tasks.send_newsletter_welcome_email_task

[2023-09-13 18:29:10,192: INFO/MainProcess] mingle: searching for neighbors
2023-09-13T18:29:11.223127806Z [2023-09-13 18:29:11,222: INFO/MainProcess] mingle: all alone
2023-09-13T18:29:11.268981135Z [2023-09-13 18:29:11,268: INFO/MainProcess] celery@e96b71c99d22 ready.
2023-09-13T18:29:14.031251535Z [2023-09-13 18:29:14,030: INFO/MainProcess] Events of group {task} enabled by remote.

Our worker is now running. Lets run our FastAPI app now on port 12345:

uvicorn app.app:app --reload --workers 1 --host 0.0.0.0 --port 12345

We will see something like:

INFO:     Will watch for changes in these directories: ['/app']
2023-09-13T18:28:36.833486177Z INFO:     Uvicorn running on http://0.0.0.0:12345 (Press CTRL+C to quit)
2023-09-13T18:28:36.833500622Z INFO:     Started reloader process [1] using WatchFiles
2023-09-13T18:28:39.603132930Z INFO:     Started server process [8]
2023-09-13T18:28:39.603200518Z INFO:     Waiting for application startup.
2023-09-13T18:28:39.604262220Z INFO:     Application startup complete.

All is ready now. Lets finally test if Celery can execute our task asynchronously. We will now post an email address to our newsletter signup route

curl -d '{"email":"test.celery@mehedees.dev"}' -H "Content-Type: application/json" -X POST http://0.0.0.0:12345/newsletter/signup

Our mock app logs:

2023-09-13 18:31:28.498 | INFO     | app.app:newsletter_signup:17 - Received newsletter signup request from test.celery@mehedees.dev
2023-09-13T18:31:28.499976407Z 2023-09-13 18:31:28.499 | INFO     | app.app:newsletter_signup:19 - Initiating welcome email sending task
2023-09-13T18:31:29.099523900Z INFO:     172.18.0.1:61574 - "POST /newsletter/signup HTTP/1.1" 200 OK

Mock app has triggered the email sending task(sent to broker) and returned response without waiting for the blocking email sending task to finish executing. Meanwhile Celery worker logs:

[2023-09-13 18:31:29,103: INFO/MainProcess] Task app.tasks.send_newsletter_welcome_email_task[cb2f6812-16d6-48ec-9e04-fa20b57bfead] received
2023-09-13T18:31:29.141559704Z 2023-09-13 18:31:29.129 | INFO     | app.tasks:send_newsletter_welcome_email_task:9 - Send welcome email task received
2023-09-13T18:31:31.923187151Z 2023-09-13 18:31:31.922 | INFO     | app.tasks:send_newsletter_welcome_email_task:21 - Email successfully sent
2023-09-13T18:31:32.237598103Z 2023-09-13 18:31:32.237 | INFO     | app.tasks:send_newsletter_welcome_email_task:22 - Send welcome email task finished
2023-09-13T18:31:32.319138219Z [2023-09-13 18:31:32,313: INFO/ForkPoolWorker-4] Task app.tasks.send_newsletter_welcome_email_task[cb2f6812-16d6-48ec-9e04-fa20b57bfead] succeeded in 3.189990901999977s: 'test.celery@mehedees.dev'

Worker has successfully received the task from broker and executed it asynchronously.

You can find the fully functional and dockerized codebase here.

I started this article originally to share my experience with Celery Canvas Workflows and Concurrency options but then realized, an introductory article might be helpful. So, I'll be writing at least two more articles, one focusing on production ready Celery app development and another on advanced features and issues.

Constructive criticisms and feedback are welcome!

Keep learning, never settle(sorry oneplus) and, till I see you again!

Edit: The 2nd post on Configuring Celery is now available.

DEV Community: Mehedee Siddique

Configuring The Celery App

task_acks_late (boolean)

task_ignore_result (boolean)

result_expires (int(seconds) | timedelta)

worker_concurrency (int)

broker_pool_limit (int | None)

worker_prefetch_multiplier (int)