Integrating Celery task queue with Pyramid

Question

I am integrating the Celery 4 task queue into my Pyramid web server. The documentation seems a little sparse though. I found a somewhat dated pyramic_celery module that’s supposed to handle Pyramid’s .ini files, and to make them usable to configure Celery. There is also an even older blog entry which uses a similar idea but a different Celery signal.

After some experimenting I found a solution which seems much simpler (i.e. doesn’t require any of the Celery signals), it works, and it even spins up the Celery Worker inspired by this SO question.

I am curious to hear people’s thoughts on this implementation. In particular, I am curious to find out if spawning the Celery worker as a sub-process of Pyramid is considered “good” or “bad” practice.

The Pyramid server’s development.ini file contains

pyramid.includes =                                                              
    pyramid_tm                                                                  
    srv.celery

where srv.celery is a module and part of the Pyramid server. This module’s __init__.py file contains all the magic:

import multiprocessing                                                          
import celery                                                                            

def includeme(config):                                                          
    pass                                                                        

# Configuration settings for the Celery instance.
_celery_config = dict(                                                          
    broker_url = "redis://localhost:6379/0",                                    
    result_backend = "redis://localhost:6379/0",                                
    imports = "srv.celery.tasks",                                               
    # More configuration can go here.
    )                                                                           

# Instantiate Celery.                                                                              
celery_app = celery.Celery("my-websrv")                                                  
celery_app.conf.update(**_celery_config)                                        

# Define a class that wraps the Celery worker process.
class CeleryWorkerProcess(multiprocessing.Process):                                   
    def __init__(self):                                                         
        super().__init__(name='celery_worker_process')                  

    # This function is called when the process is started, thus
    # creating a new child-process of the web-server that runs
    # the Celery worker.
    def run(self):                                                              
        argv = [                                                                
            'worker',                                                           
            '--loglevel=info',                                                  
            '--quiet',                                                          
        ]                                                                       
        celery_app.worker_main(argv)                                            

# Create the Celery worker process and run it.
_celery_worker = CeleryWorkerProcess()                                                 
_celery_worker.start()                                                           

print("Started Celery", celery_app, "and worker in PID", celery_worker.pid)

This solution does not consider Pyramid’s .ini file configuration, but maybe that’s just ok. This solution does not work in a daemonic setup.

When I pserve the project, both the web server and the Celery worker process spin up and respond. For now, the only task is the add() from the basic example:

Started Celery <Celery my-websrv:0x10f6f20f0> and worker in PID 51926
Starting server in PID 51923.
Serving on http://0.0.0.0:6543
[2016-12-04 10:00:26,796: INFO/celery_worker_process] Connected to redis://localhost:6379/0
[2016-12-04 10:00:26,805: INFO/celery_worker_process] mingle: searching for neighbors
[2016-12-04 10:00:27,825: INFO/celery_worker_process] mingle: all alone
[2016-12-04 10:00:27,840: INFO/celery_worker_process] [email protected] ready.
[2016-12-04 10:01:00,067: INFO/celery_worker_process] Received task: srv.celery.tasks.add[ddf286c9-f34c-4d70-8c49-2570a5830843]  
[2016-12-04 10:01:00,075: INFO/PoolWorker-1] Task srv.celery.tasks.add[ddf286c9-f34c-4d70-8c49-2570a5830843] succeeded in 0.005526618973817676s: 4

Graipher · Accepted Answer · 2016-12-04 12:01:08Z

I think you should avoid using global variables. You can just pass the celery_app as a parameter to the __init__ function and not have to worry too much about anything modifying your global state. Your class is then also testable:

class CeleryWorkerProcess(multiprocessing.Process):
    def __init__(self, celery_app):
        super().__init__(name='celery_worker_process')
        self.celery_app = celery_app

    # This function is called when the process is started, thus
    # creating a new child-process of the web-server that runs
    # the Celery worker.
    def run(self):
        argv = ('worker', '--loglevel=info', '--quiet')
        self.celery_app.worker_main(argv)

# Create the Celery worker process and run it.
_celery_worker = CeleryWorkerProcess(celery_app)
_celery_worker.start()

I also put your argv in one line, it is not too big for that. I also made it a tuple, because it will not change and tuples are slightly more memory efficient.

While trying to copy and paste your class definition, I noticed that all your lines are filled with spaces until (I guess) 80 characters. That is a very weird convention (which also messes up the formatting when pasting into a textbox).

In addition to this, I would put all your code into a main method and guard it with if __name__ == "__main__":. This enables you to import parts of this code easily from other scripts.

def main():
    # Instantiate Celery.
    celery_app = celery.Celery("my-websrv")
    celery_app.conf.update(**_celery_config)

    # Create the Celery worker process and run it.
    _celery_worker = CeleryWorkerProcess(celery_app)
    _celery_worker.start()

if __name__ == "__main__":
    main()

This way you adhere to the python way of structuring your files:

#!/usr/bin/python

import stdlib_module

import custom_module

CONSTANTS


class Definitions


def functions


if __name__ == "__main__":
    run_code()

Thank you for the comments. One thing: the code is not supposed to run standalone (no need for __main__) but instead is initialized during Pyramid’s startup. It probably would be cleaner to call it from Pyramid’s main though, instead of initializing it as a module... — Jens
– Jens, Commented Dec 4, 2016 at 11:50
@Jens Yeah, a ` from utils import start_celery` and then in the pyramid a call to start_celery is more expressive and clear than an import or plain execution of another module within the startup code. (Renaming the main I defined to start_celery) — Graipher
– Graipher, Commented Dec 4, 2016 at 12:04
While I agree with trying to avoid globals in general, a Celery task's decorator refers to the Celery application instance (see here). I don't see how this can not be a global, and is used by Celery itself that way. — Jens
– Jens, Commented Dec 4, 2016 at 23:49

Stack Exchange Network

Integrating Celery task queue with Pyramid

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Integrating Celery task queue with Pyramid

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions