我正在尝试通过创建一个简单的基于 mongo 的 rest api 来玩 fastapi。以下是我到目前为止所做的：
文件：这包含使用框架mongo.py从 mongo db 获取数据的所有代码：motor
"""Module for handling the motor mongo package code."""
from typing import Any, Dict, List
import motor.motor_asyncio
DB = "books"
COLLECTION = "books"
class MongoBackend:
def __init__(self, uri: str) -> None:
self._client = motor.motor_asyncio.AsyncIOMotorClient(uri)
async def get_all_books(self) -> List[Dict[str, Any]]:
cursor = self._client[DB][COLLECTION].find({}, {"_id": 0})
return [doc async for doc in cursor]
async def get_single_book(self, book_id: str) -> Dict[str, Any]:
return await self._client[DB][COLLECTION].find_one(
{"book_id": book_id}, {"_id": 0}
)
async def update_one_book(self, book_id: str, data: Dict[str, Any]) -> None:
data["book_id"] = book_id
await self._client[DB][COLLECTION].update(
{"book_id": book_id}, data, upsert=True
)
async def insert_one_book(self, data: Dict[str, Any]) -> None:
await self._client[DB][COLLECTION].insert_one(data)
async def delete_one_book(self, book_id: str) -> None:
await self._client[DB][COLLECTION].delete_one({"book_id": book_id})
我的routers.py文件如下所示：
"""Module for containing the routes for the application."""
import uuid
from typing import Dict, Any, Tuple, List
from fastapi import APIRouter, Depends
import application.models as models
import application.mongo as mongo
import os
import functools
router = APIRouter()
@functools.lru_cache()
def mongo_backend():
return mongo.MongoBackend(uri=os.getenv("MONGODB_URI"))
@functools.lru_cache()
def base_uri():
return os.getenv("BASE_URI")
def add_hyper_link_to_book(book: Dict[str, Any]) -> Dict[str, Any]:
book_id = book.pop("book_id")
book["_link"] = f"{base_uri()}book/{book_id}"
return book
@router.get("/books")
async def get_all_books():
backend = mongo_backend()
print(id(backend))
all_books = [
add_hyper_link_to_book(book)
for book in await backend.get_all_books()
]
return all_books
@router.post("/books")
async def add_a_book(book: models.Book) -> Tuple[List[Dict[str, Any]], int]:
backend = mongo_backend()
book_to_insert = book.dict()
book_to_insert["book_id"] = str(uuid.uuid1())
await backend.insert_one_book(data=book_to_insert)
return [
add_hyper_link_to_book(book)
for book in await backend.get_all_books()
], 201
@router.get("/book/{book_id}")
async def get_a_single_book(book_id: str, backend: mongo.MongoBackend = Depends(mongo_backend)):
book = await backend.get_single_book(book_id=book_id)
if book is None:
return {"message": "No such book exist!!"}
return add_hyper_link_to_book(book)
所以到目前为止我有3条路线，我想将我的后端对象依赖注入到路线中，我也希望它是一个单例。
为了显示有效和无效之间的区别，我仅将依赖项注入到其中一个路由（/book）中，而对于其余路由（/books），我刚刚调用了函数来mongo_backend直接创建后端（）。
我不太明白我做错了什么，因为当我尝试查询注入依赖项的路由时，后端对象会引发此错误：
ERROR: Exception in ASGI application
Traceback (most recent call last):
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/uvicorn/protocols/http/h11_impl.py", line 396, in run_asgi
result = await app(self.scope, self.receive, self.send)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/uvicorn/middleware/proxy_headers.py", line 45, in __call__
return await self.app(scope, receive, send)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/fastapi/applications.py", line 199, in __call__
await super().__call__(scope, receive, send)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/applications.py", line 111, in __call__
await self.middleware_stack(scope, receive, send)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/middleware/errors.py", line 181, in __call__
raise exc from None
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/middleware/errors.py", line 159, in __call__
await self.app(scope, receive, _send)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/exceptions.py", line 82, in __call__
raise exc from None
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/exceptions.py", line 71, in __call__
await self.app(scope, receive, sender)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/routing.py", line 566, in __call__
await route.handle(scope, receive, send)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/routing.py", line 227, in handle
await self.app(scope, receive, send)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/routing.py", line 41, in app
response = await func(request)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/fastapi/routing.py", line 191, in app
solved_result = await solve_dependencies(
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/fastapi/dependencies/utils.py", line 550, in solve_dependencies
solved = await run_in_threadpool(call, **sub_values)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/starlette/concurrency.py", line 34, in run_in_threadpool
return await loop.run_in_executor(None, func, *args)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/concurrent/futures/thread.py", line 57, in run
result = self.fn(*self.args, **self.kwargs)
File "./application/routers.py", line 17, in mongo_backend
return mongo.MongoBackend(uri=os.getenv("MONGODB_URI"))
File "./application/mongo.py", line 12, in __init__
self._client = motor.motor_asyncio.AsyncIOMotorClient(uri)
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/motor/core.py", line 150, in __init__
io_loop = self._framework.get_event_loop()
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/site-packages/motor/frameworks/asyncio/__init__.py", line 42, in get_event_loop
return asyncio.get_event_loop()
File "/home/subhayan/anaconda3/envs/mongo-book-store-fastapi/lib/python3.8/asyncio/events.py", line 639, in get_event_loop
raise RuntimeError('There is no current event loop in thread %r.'
RuntimeError: There is no current event loop in thread 'ThreadPoolExecutor-1_0'.
两条路线的其余部分工作正常。有人可以帮助我完成这项工作需要做些什么。
如果它有帮助，这是我的app.py：
"""Main module for the fastapi app."""
import pathlib
from dotenv import load_dotenv
from fastapi import FastAPI
import application.routers as routers
# load the environment from the file app.env in the project directory
basedir = pathlib.Path(__file__).parent.parent
load_dotenv(basedir / "app.env")
app = FastAPI()
app.include_router(routers.router)