我正在尝试连接到FastAPI中的mongodb。我一再得到这个例外。
文件-main.py
app = FastAPI(
title=config.PROJECT_NAME, docs_url="/api/docs", openapi_url="/api"
)
@app.get("/api/testing")
async def testit():
user_collection = readernetwork_db.get_collection("user_collection")
all_users = await user_collection.find_one({"email": "sample_email"})
print("all users --- ", all_users)
return all_users
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", reload=True, port=8888)
文件-session.py
import motor.motor_asyncio
from app.core import config
print("here we go again....")
client = motor.motor_asyncio.AsyncIOMotorClient(
config.MONGOATLAS_DATABASE_URI)
readernetwork_db = client.get_database("readernetwork")
异常-：
all_users = await user_collection.find_one({"email": "sample_email"})
RuntimeError: Task <Task pending name='Task-4' coro=<RequestResponseCycle.run_asgi() running at /usr/local/lib/python3.8/site-packages/uvicorn/protocols/http/h11_impl.py:389> cb=[set.discard()]> got Future <Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/local/lib/python3.8/asyncio/futures.py:360]> attached to a different loop
我不知道我哪里搞错了。我应该指定一个到马达的事件循环吗？
您可以在全局范围内拥有mongodb motor客户端，但创建和关闭它应该在异步函数内完成。最可取的方法是在应用程序的startup和shutdown处理程序中执行此操作。像这样：
# mongodb.py
from motor.motor_asyncio import AsyncIOMotorClient
db_client: AsyncIOMotorClient = None
async def get_db_client() -> AsyncIOMotorClient:
"""Return database client instance."""
return db_client
async def connect_db():
"""Create database connection."""
global db_client
db_client = AsyncIOMotorClient(DB_URL)
async def close_db():
"""Close database connection."""
db_client.close()
# main.py
app = FastAPI(title=PROJECT_NAME)
...
app.add_event_handler("startup", connect_db)
app.add_event_handler("shutdown", close_db)
请注意，您需要global db_client行来修改预先定义的全局变量。
无需使用全局。。。您可以从提供给任何路由的请求中访问应用程序状态。
#main.py
async def open_db() -> AsyncIOMotorClient:
app.state.mongodb = AsyncIOMotorClient(DB_URL)
async def close_db():
app.state.mongodb.close()
app.add_event_handler('startup', open_db)
app.add_event_handler('shutdown', close_db)
在每个到指定路线的请求中，您都可以访问应用程序状态。例如，
@app.route('/{username}')
async def index(request: Request, username: str):
user = await request.app.state.mongodb['auth']['users'].find_one({"username" : username})
您甚至可以通过在open_db函数中执行这样的操作来简化它。指定一个状态值(如"users"(作为特定的Collections实例。
async def open_db() -> AsyncIOMotorClient:
app.state.mongodb = AsyncIOMotorClient(DB_URL)
app.state.users = app.state.mongodb['auth']['users']
现在你可以这么做了，
@app.route('/{username}')
async def index(request: Request, username: str):
user = await request.app.state.users.find_one({"username" : username})
client = AsyncIOMotorClient()
client.get_io_loop = asyncio.get_event_loop
对我有用。