Python - Celery how to establish async connection per worker

user_11027207 0
Viewed 280

I created several endpoint calling to Celery tasks which preforming different tasks against the DB. Obviously it doesn't make sense to re-connect to DB each time, But from other hand - When the connection should be close? Is it make sense using async connection to DB?

I'm not sure how i can achieve that and if that make sense to use Async with Celery - would appreciate any guidance

import os
import traceback

from celery import Celery
from celery.utils.log import get_task_logger

from config.config import *


app = Celery('proj',
             broker=config('CELERY_BROKER_URL'),
             backend=config('CELERY_RESULT_BACKEND'),
             include=['proj.tasks','proj.fetch_data'])

app.conf.update(
    result_expires=3600,
)


app.autodiscover_tasks()



if __name__ == '__main__':
    app.start()

I came up to worker_process_init, worker_process_shutdown,

Note : database_ps_ms_stg - is the based on Databases (async to postgres).

tasks.py

from .celery import app
from celery.signals import worker_process_init, worker_process_shutdown,task_postrun

from config.db import database_ps_ms_stg 
import asyncio



@worker_process_init.connect
async def init_worker(**kwargs):
 
    if not database_ps_ms_stg.is_connected:
        await database_ps_ms_stg.connect()
        print ("connected to database_ps_ms_stg") 
    

@worker_process_shutdown.connect
async def shutdown_worker(**kwargs):
    
    if  database_ps_ms_stg.is_connected:
        await database_ps_ms_stg.disconnect()
        print ("disconneceting from database_ps_ms_stg")

Getting :

[2021-07-18 16:23:16,951: WARNING/ForkPoolWorker-1] /usr/local/lib/python3.8/site-packages/celery/concurrency/prefork.py:77: RuntimeWarning: coroutine 'init_worker' was never awaited
  signals.worker_process_init.send(sender=None)

```
celery python
1 Answers

Your coroutines are not being scheduled for execution in any event loop. For example, this code

@worker_process_init.connect
async def init_worker(**kwargs):
 
    if not database_ps_ms_stg.is_connected:
        await database_ps_ms_stg.connect()
        print ("connected to database_ps_ms_stg")

just creates a coroutine object when worker_process_init fires, but does nothing with this object afterwards.

It is probably a solution - to wrap your coroutines with kind of a scheduler-decorator, which will start them

def async2sync(func):
    def wrapper(*args, **kwargs):
        loop = asyncio.get_event_loop()
        task = loop.ensure_future(func())
        task.add_done_callback(lambda f: loop.stop())
        loop.run_forever()
        try:
            return task.result()
        except asyncio.CancelledError:
            pass
    return wrapper

...    
@worker_process_init.connect
@async2sync
async def init_worker(**kwargs):
 
    if not database_ps_ms_stg.is_connected:
        await database_ps_ms_stg.connect()
        print ("connected to database_ps_ms_stg")

Please check if this may answer some of your questions. My opinion - it's not worth it, better just use blocking connectors, as long as your code is intended to be used with celery.

user_71832350
Related