getting task attached to a different loop runtime error when using fastapi with gunicron and motor

See original GitHub issue

First Check

  • I added a very descriptive title to this issue.
  • I used the GitHub search to find a similar issue and didn’t find it.
  • I searched the FastAPI documentation, with the integrated search.
  • I already searched in Google “How to X in FastAPI” and didn’t find any information.
  • I already read and followed all the tutorial in the docs and didn’t find an answer.
  • I already checked if it is not related to FastAPI but to Pydantic.
  • I already checked if it is not related to FastAPI but to Swagger UI.
  • I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

  • I commit to help with one of those options 👆

Example Code

# mongo
from motor import motor_asyncio

mongo_uri = "mongodb://mongoUser:***/?authSource=admin"
mongo_client = motor_asyncio.AsyncIOMotorClient(mongo_uri)

def get_users_db() -> motor_asyncio.AsyncIOMotorDatabase:
    return mongo_client.devdb

# user_routes
import fastapi
import mongo
from bson import objectid

db = mongo.get_users_db()
router = fastapi.APIRouter()


@router.get("/{user_id}")
async def get_user_by_id(user_id: str):
    user = await db.users.find_one({"_id": objectid.ObjectId(user_id)})
    if user:
        user["_id"] = str(user["_id"])
        return user
    return {"error": "no user found"}

# main.py
import fastapi
import user_routes

app = fastapi.FastAPI()

app.include_router(user_routes.router)

Description

i am trying to create an app with mongodb client motor and fastapi. I created seperate routes and a seperate file to setup mongodb client. Running this with uvicorn works fine and i get results after querying the database but when i deploy the app with gunicorn and again use the route, i get the following error.

image I tried bunch of methods, like assigning a loop to motor client, changing uvworker to use asyncio loop etc. i don’t know where to post this error but here i am

  • expected: json of user with given id from database or no user found error
  • got: Runtime error

Operating System

Linux

Operating System Details

Debian 11

FastAPI Version

0.68.1

Python Version

Python 3.9.2

Additional Context

motor==2.2.6 gunicorn==20.1

this issue only happens with gunicorn as server and only when accessing mongodb from a fastapi route (i checked running db query in the mongo file and it works)

this my first ever time creating an issue, so please guide me i did something wrong.

Issue Analytics

  • State:closed
  • Created 2 years ago
  • Reactions:2
  • Comments:10 (3 by maintainers)

github_iconTop GitHub Comments

19reactions
CanD42commented, Jan 14, 2022

That is the problem: https://github.com/encode/starlette/issues/1315

client = AsyncIOMotorClient()
client.get_io_loop = asyncio.get_event_loop

works for me

1reaction
miladvayanicommented, Dec 12, 2021

you should define in start up , like this :

app = FastAPI()
app.add_event_handler("startup", start_app) 


async def start_app():
    await Mongo().get_connection()


from motor.motor_asyncio import AsyncIOMotorClient
from config import CONNECTION_STRING, DB_NAME


class Mongo:
    client = None
    db = None

    async def get_connection(self) -> AsyncIOMotorClient:
        if not Mongo.client:
            Mongo.client = AsyncIOMotorClient(CONNECTION_STRING)

now you can use mongo client every where you want!!!

Read more comments on GitHub >

github_iconTop Results From Across the Web

Future task attached to a different loop - python - Stack Overflow
I am repeatedly getting this exception. File - main.py app = FastAPI( title=config.PROJECT_NAME, docs_url="/api/docs", openapi_url ...
Read more >
tiangolo/fastapi - Gitter
I'm developing a FastAPI app and was experimenting with Celery as a task queue. However, it seems that it often just leaves tasks...
Read more >
Async Tests - FastAPI
If you encounter a RuntimeError: Task attached to a different loop when integrating asynchronous function calls in your tests (e.g. when using MongoDB's ......
Read more >
Developing and Testing an Asynchronous API with FastAPI ...
This tutorial looks at how to develop and test an asynchronous API with FastAPI, Postgres, pytest, and Docker using Test-driven Development ...
Read more >
The Ultimate FastAPI Tutorial Part 9 - Asynchronous ...
Theory Section - Python Asyncio and Concurrent Code. A quick bit of terminology. In programming, concurrency means: Executing multiple tasks at ...
Read more >

github_iconTop Related Medium Post

No results found

github_iconTop Related StackOverflow Question

No results found

github_iconTroubleshoot Live Code

Lightrun enables developers to add logs, metrics and snapshots to live code - no restarts or redeploys required.
Start Free

github_iconTop Related Reddit Thread

No results found

github_iconTop Related Hackernoon Post

No results found

github_iconTop Related Tweet

No results found

github_iconTop Related Dev.to Post

No results found

github_iconTop Related Hashnode Post

No results found