Example for running celery task which includes db transcation in async fastapi app

[codesutras]

Is there any working example of running a celery task, which has some DB transactions ( async DB transaction) ??

Note: I'm using MongoDB (motor) for db utils.

This URL is a well-documented example for running fastapi with MongoDB in async fashion.

I've checked all project template generator as well. None of them have any example for having a celery task interacting with db transactions. ( I'll be okay to contribute on this once these current issues will get resolve)

I've found that there is no support for an async function in current (stable) celery release. Until then, it's going to be a massive rewriting for me,if I wanted to use fastapi with celery ( Need to drop off MongoDB async way).

Is there any better way to deal with this situation? So, that fastapi can have few endpoints work with MongoDB (motor) transactions and a few endpoints ( which have celery task) can consume db functions in sync way to work well.

In a nutshell,
Q1: How to handle async db transactions in the celery task?
Q2: Can we have async and sync db connection in fast api?

The reason, I'm asking these questions because I'm getting below exception in the celery tasks. which are depending on db calls ( mongodb motor).

kombu.exceptions.EncodeError: Object of type coroutine is not JSON serializable
@tiangolo Any thought would be appreciated. Since I really don't want to lose the power of async and it's performance advantages with fastapi.

[retnikt]

Can you provide a code sample for this problem?

[codesutras]

@retnikt I've achieved to run celery task with DB transactions in async fashion. Though, it was not at all straight forward solutions. I have to do some cleanup in code and will share it after that.

I'll create a git repo with detailed information for using celery task + scheduling + fast API in next few days.

[tiangolo]

Thanks for reporting back and closing the issue. 👍

If using async functions for distributed background tasks, I would try ARQ 😉

[codesutras]

Thanks for reporting back and closing the issue.

If using async functions for distributed background tasks, I would try ARQ

@tiangolo I'd tried ARQ already. But, it can't handle too much load. It got failed in my load testing. Celery was the only solution in my case. Though ARQ works pretty well if you don't have a much-complicated task to be performed.

Nevertheless, Celery coroutine support is in pipeline and will get a release in celery 5.0 as per the github discussion. Code is almost ready for it. But, I'd reported an issue in it.

[lucasgadams]

How do we feel about asyncio options now 4 years later? Not sure if it is actually on the celery roadmap anymore. ARQ has been continued to be developed. Really interested to hear any solutions!

[ZhengRui]

@codesutras I am facing the same issue these days, in my case I have a function that does some image processing and then updates the database, I am trying to figure out how to decorate this function using celery Task as the function contains await. Do you have a repo to show how you did it? Thanks.

[cs-satish-mishra]

@ZhengRui could you find any solution to this?

[ZhengRui]

celery app file worker.py

from celery import Celery

app = Celery('app', backend='rpc://', broker='amqp://localhost')

task definition file tasks/default.py

import asyncio
from database import Database
from ..worker import app

db = Database(f'postgresql://user:password@localhost:port/db')

async def some_task(param1: str, param2: str, current_task_ref):
    await db.connect()

    # interact with db
    tot = 1e4
    for i in range(tot):
        current_task_ref.update_state(state='PROGRESSING', meta={'current': i + 1, 'total': tot})

    await db.close()
    return {'current': tot, 'total': tot}

@app.task(bind=True, acks_late=True)
def some_task_t(self, *args, **kwargs):
    return asyncio.run(some_task(*args, **kwargs, current_task_ref=self))

a file to expose to fastapi tq.py

from celery.result import AsyncResult

from .tasks.default import (
    some_task_t,
    another_task_t,
)

def get_task_progress(task_id):
    result = AsyncResult(task_id)
    return result.info

in your fastapi routes

import tq

@r.post('/sometask')
async def some_task_r(
    param1: str,
    param2: str,
    db: Database = Depends(get_db),
):
    task = tq.some_task_t.delay(
        param1,
        param2,
    )

    return task.task_id

@r.get('/sometask/progress')
async def get_some_task_progress_r(
    task_id: str,
):
    return tq.get_task_progress(task_id)

when rabbitmq server and postgres database is ready, start the celery worker: celery -A worker.app worker --loglevel=INFO --pool=gevent --concurrency=10, depends on your actual task, you can try out different worker pool settings 'processes', 'threads' etc and concurrency settings. @cs-satish-mishra

[cs-satish-mishra]

@ZhengRui thanks a lot for this.

[selfteachingcode]

@ZhengRui
@cs-satish-mishra - did the solution work for you??

I tried the above - I get this error:

Future <Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/lib/python3.8/asyncio/futures.py:360]> attached to a different loop
ERROR/ForkPoolWorker-7] Task app.process_data[64d9492b-0981-4153-90a4-c7040b3e7d4b] raised unexpected: RuntimeError("Task <Task pending name='Task-1' coro=<some_task() running at /home/shavs/Documents/test/app.py:19> cb=[_run_until_complete_cb() at /usr/lib/python3.8/asyncio/base_events.py:184]> got Future <Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/lib/python3.8/asyncio/futures.py:360]> attached to a different loop") Traceback (most recent call last): File "/usr/local/lib/python3.8/dist-packages/celery/app/trace.py", line 412, in trace_task R = retval = fun(*args, **kwargs) File "/usr/local/lib/python3.8/dist-packages/celery/app/trace.py", line 704, in __protected_call__ return self.run(*args, **kwargs) File "/home/shavs/Documents/test/app.py", line 31, in process_data asyncio.run(some_task(data)) File "/usr/lib/python3.8/asyncio/runners.py", line 44, in run return loop.run_until_complete(main) File "/usr/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete return future.result() File "/home/shavs/Documents/test/app.py", line 19, in some_task result = await db['your_collection'].insert_one(data) RuntimeError: Task <Task pending name='Task-1' coro=<some_task() running at /home/shavs/Documents/test/app.py:19> cb=[_run_until_complete_cb() at /usr/lib/python3.8/asyncio/base_events.py:184]> got Future <Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/lib/python3.8/asyncio/futures.py:360]> attached to a different loop

this is my app.py

from fastapi import FastAPI
from celery import Celery
from motor.motor_asyncio import AsyncIOMotorClient
import time
import asyncio

# Create a FastAPI app
app = FastAPI()

# Configure Celery
celery_app = Celery('tasks', broker='redis://:redis_pass@localhost:6379/0')

# Connect to MongoDB using Motor
mongo_client = AsyncIOMotorClient('mongodb://localhost:27017')
db = mongo_client['test_dev']


async def some_task(data):
      result = await db['your_collection'].insert_one(data)
      print("inserted!", result)
      


# Define a Celery task
@celery_app.task
def process_data(data):
    # Perform some processing on the data
    # For example, save it to MongoDB
    time.sleep(50)
    print(data)
    asyncio.run(some_task(data))

# Define a route in FastAPI to trigger the Celery task
@app.get('/process')
async def process_endpoint():
    # Call the Celery task asynchronously
    data = {"a": "A"}
    task = process_data.apply_async(args=[data])
    return {'message': 'Data processing started'}

[jashwant]

@selfteachingcode,

asyncio.run creates a new loop each time it is called.
The task created in one loop cannot be accessed in different loop.

Below is a detailed code of how I dealt with this. Python async/await is still very weird to me. So, feel free to correct.

My celery app was a separate service running in docker.
I needed to call async method fetch_user_api through celery and fetch_user_api was using a async cache library.
So, I needed to start and close cache and also use the same cache for fetch_user_api.
Methods to start cache, close cache and api call to fetch users were all async.

import asyncio
import threading
from utils import init_cache, close_cache

thread_local = threading.local()

async def fetch_user_api():
    # Saving api call in redis cache
   
async def update_all_users():
    await fetch_user_api()

def get_or_create_event_loop():
    if not hasattr(thread_local, "loop") or thread_local.loop.is_closed():
        thread_local.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(thread_local.loop)
    return thread_local.loop


@worker_init.connect
def on_worker_init(*args, **kwargs):
    loop = get_or_create_event_loop()
    loop.run_until_complete(init_cache())
    print("Cache initialized!")


@worker_shutdown.connect
def on_worker_shutdown(*args, **kwargs):
    loop = get_or_create_event_loop()
    loop.run_until_complete(close_cache())
    loop.close()
    print("Cache closed!")


# Task runs every 5 mins
@celery_app.task(acks_late=True)
def periodic_update():
    loop = get_or_create_event_loop()
    loop.run_until_complete(update_all_users())