如何将 FastAPI 路由器与 FastAPI-Users 和 MongoDB 一起使用？

Question

我也可以将 MongoDB 与 FastAPI 一起使用

1. 使用全局 client: motor.motor_asyncio.AsyncIOMotorClient 对象，否则
2. 通过在 startup 事件期间为每个 this SO answer 创建一个，它指的是 this "Real World Example"。

但是，我也想使用fastapi-users，因为它可以很好地与开箱即用的 MongoDB 配合使用。缺点是它似乎只适用于处理我的数据库客户端连接（即全局）的第一种方法。原因是为了配置 fastapi-users，我必须有一个活动的 MongoDB 客户端连接，这样我才能创建 db 对象，如下所示，我需要 db 然后创建 MongoDBUserDatabase 对象fastapi-users 要求：

# main.py
app = FastAPI()


# Create global MongoDB connection 
DATABASE_URL = "mongodb://user:paspsword@localhost/auth_db"
client = motor.motor_asyncio.AsyncIOMotorClient(DATABASE_URL, uuidRepresentation="standard")
db = client["my_db"]

# Set up fastapi_users
user_db = MongoDBUserDatabase(UserDB, db["users"])

cookie_authentication = CookieAuthentication(secret='lame secret' , lifetime_seconds=3600, name='cookiemonster')

fastapi_users = FastAPIUsers(
    user_db,
    [cookie_authentication],
    User,
    UserCreate,
    UserUpdate,
    UserDB,
)

在代码中的那一点之后，我可以导入 fastapi_users 路由器。但是，如果我想将我的项目分解为我自己的 FastAPI 路由器，我会感到很沮丧，因为：

1. 如果我将 client 创建移动到另一个模块以导入到我的 app 和我的路由器中，那么我在不同的事件循环中有不同的客户端并得到像 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 这样的错误（在 this SO question 中提到)
2. 如果我使用“真实世界示例”的解决方案，那么在我的代码示例中我会卡在哪里构建我的 fastapi_users 对象：我无法在 main.py 中执行此操作，因为没有 db对象。

我考虑将MongoDBUserDatabase 对象作为startup 事件代码的一部分（即在真实世界示例中的async def connect_to_mongo() 内），但我也无法让它工作，因为我做不到看看如何让它发挥作用。

我该怎么做

1. 以一种可以在我的主要app 和几个routers 之间共享的方式创建一个全局 MongoDB 客户端和 FastAPI-User 对象，而不会出现“附加到不同的循环”错误，或者
2. 创建精美的包装类和函数以使用startup 触发器设置 FastAPI 用户？

Answer 1

我认为我的解决方案不完整或不正确，但我想我会发布它以防它激发任何想法，我很难过。我遇到了确切的困境，几乎看起来像是一个设计缺陷..

我关注了这个MongoDB full example并将其命名为main.py

此时我的应用程序无法运行。服务器启动，但每次尝试查询数据库时都会导致上述“附加到不同的循环”。

寻找指导，我偶然发现了同一个“现实世界”example

在main.py 中添加了启动和关闭事件处理程序

# Event handlers
app.add_event_handler("startup", create_start_app_handler(app=app))
app.add_event_handler("shutdown", create_stop_app_handler(app=app))

在dlw_api.db.events.py这个：

import logging

from dlw_api.user import UserDB
from fastapi import FastAPI
from fastapi_users.db.mongodb import MongoDBUserDatabase
from motor.motor_asyncio import AsyncIOMotorClient


LOG = logging.getLogger(__name__)
DB_NAME = "dlwLocal"
USERS_COLLECTION = "users"
DATABASE_URI = "mongodb://dlw-mongodb:27017"  # protocol://container_name:port


_client: AsyncIOMotorClient = None
_users_db: MongoDBUserDatabase = None


def get_users_db() -> MongoDBUserDatabase:
    return _users_db


async def connect_to_db() -> None:
    global _users_db
    # logger.info("Connecting to {0}", repr(DATABASE_URL))
    client = AsyncIOMotorClient(DATABASE_URI)
    db = client[DB_NAME]
    collection = db[USERS_COLLECTION]
    _users_db = MongoDBUserDatabase(UserDB, collection)
    LOG.info(f"Connected to {DATABASE_URI}")


async def close_db_connection(app: FastAPI) -> None:
    _client.close()
    LOG.info("Connection closed")

还有dlw_api.events.py:

from typing import Callable
from fastapi import FastAPI
from dlw_api.db.events import close_db_connection, connect_to_db
from dlw_api.user import configure_user_auth_routes
from fastapi_users.authentication import CookieAuthentication
from dlw_api.db.events import get_users_db


COOKIE_SECRET = "THIS_NEEDS_TO_BE_SET_CORRECTLY" # TODO: <--|
COOKIE_LIFETIME_SECONDS: int = 3_600
COOKIE_NAME = "c-is-for-cookie"

# Auth stuff:
_cookie_authentication = CookieAuthentication(
    secret=COOKIE_SECRET,
    lifetime_seconds=COOKIE_LIFETIME_SECONDS,
    name=COOKIE_NAME,
)

auth_backends = [
    _cookie_authentication,
]


def create_start_app_handler(app: FastAPI) -> Callable:
    async def start_app() -> None:
        await connect_to_db(app)
        configure_user_auth_routes(
            app=app,
            auth_backends=auth_backends,
            user_db=get_users_db(),
            secret=COOKIE_SECRET,
        )

    return start_app


def create_stop_app_handler(app: FastAPI) -> Callable:
    async def stop_app() -> None:
        await close_db_connection(app)

    return stop_app

这对我来说不正确，这是否意味着所有使用Depends 进行用户身份验证的路由都必须包含在服务器启动事件处理程序中？？

Answer 2

fastapi-users created a repl.it 的作者 (frankie567) 展示了各种解决方案。 My discussion 关于此解决方案可能会提供更多上下文，但该解决方案的关键部分是：

1. 不要费心使用 FastAPI startup 触发器和 Depends 进行 MongDB 连接管理。相反，创建一个单独的文件（即db.py）来创建您的数据库连接和客户端对象。在需要时导入此 db 对象，例如您的路由器，然后将其用作全局对象。
2. 另外创建一个单独的users.py 来做两件事：
   1. 创建全局使用的fastapi_users = FastAPIUsers(...) 对象，用于与其他路由器一起处理授权。
   2. 创建一个 FastAPI.APIRouter() 对象并将所有 fastapi-user 路由器附加到它 (router.include_router(...))
3. 在所有其他路由器中，根据需要从上面导入 db 和 fastapi_users
4. 关键：将您的主代码拆分为
   1. main.py 仅导入 uvicorn 并服务于 app:app。
   2. 一个app.py，它有你的主要FastAPI对象（即app），然后连接我们所有的路由器，包括来自users.py的路由器，所有fastapi-users路由器都连接到它。

通过将上面的每 4 个代码分开，您可以避免“附加到不同的循环”错误。

Answer 3

我遇到了类似的问题，我所要做的就是让 motor 和 fastapi 在同一个循环中运行：

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

我没有设置 on_startup 或其他任何设置。