构建现代 web 应用程序时，管理和与数据库交互是一项基本任务。作为 NoSQL 数据库，MongoDB 提供了出色的文档存储灵活性，这使得它非常适合存储和管理非结构化数据。而 FastAPI 是构建 API 的最快 Python 框架之一。本文档将指导您如何将 MongoDB 无缝集成到 FastAPI 应用程序中，并实现基本的 CRUD（创建、读取、更新、删除）操作，从而使测试和数据库交互更加方便。

1. 使用 FastAPI 设置 MongoDB。
2. 使用 motor，一个异步 MongoDB 驱动程序，与 FastAPI 结合使用。
3. 使用 MongoDB 创建简单的 CRUD（创建、读取、更新、删除）操作。

直接来吧！

你需要先安装以下，

以下是专家建议修改后的翻译，确保符合原文的直接和口语化风格，同时改善了中文的常用表达方式和可读性：

你需要先安装以下项目：

• Python（最好是3.7及以上版本）
• MongoDB Atlas，（或任何MongoDB实例）
• FastAPI，和Motor

运行以下命令安装所需依赖项：

在命令行中输入：

pip install fastapi motor uvicorn python-dotenv

你需要配置MongoDB Atlas，或者运行本地的MongoDB实例。

这里有一个基本的文件夹结构供你参考：

    项目文件夹/  
    │  
    ├── main.py  
    ├── mongodb.py  
    ├── model.py  
    ├── route.py  
    ├── 环境变量文件(.env)

以下代码演示了如何将 MongoDB 链接到 FastAPI。

此文件包含FastAPI的核心应用，以及MongoDB的连接设置。

    import os  
    from fastapi import FastAPI  
    from fastapi.middleware.cors import CORSMiddleware  
    from dotenv import load_dotenv  
    from motor.motor_asyncio import AsyncIOMotorClient  
    from mongodb import MongoDBBase  
    from route import router  

    load_dotenv()  

    app = FastAPI()  

    @app.on_event("startup")  
    async def startup_db_client():  
        # 请根据您的需求更新连接字符串  
        # 更多详情: https://www.mongodb.com/docs/manual/reference/connection-string  
        mongodb_client = AsyncIOMotorClient(  
            f"mongodb+srv://{os.getenv('DB_USERNAME')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_CLUSTER')}.example.mongodb.net/"  
        )  
        database = mongodb_client[os.getenv("DB_DATABASE")]  
        MongoDBBase.set_database(database)  

    @app.on_event("shutdown")  
    async def shutdown_db_client():  
        # 获取数据库客户端  
        mongodb_client = MongoDBBase.get_database().client  
        mongodb_client.close()  

    app.add_middleware(  
        CORSMiddleware,  
        allow_origins=["*"],  
        allow_credentials=True,  
        allow_methods=["*"],  
        allow_headers=["*"],  
    )  

    app.include_router(router)

在以下代码中：

• 我们使用 dotenv 加载环境变量，以保护敏感信息（如数据库凭证）的安全。
• 我们使用 motor 异步建立与 MongoDB 集群的连接。
• startup 事件用来初始化 MongoDB 客户端并建立连接，而 shutdown 事件则用来安全地关闭连接。

接下来，我们来定义 mongodb.py 文件中的 MongoDB 交互操作。

    from typing import List, Dict, Any, Mapping  
    from fastapi import HTTPException  
    from bson import ObjectId  
    from motor.motor_asyncio import AsyncIOMotorDatabase  

    class MongoDBBase:  
        collection_name: str = ""  
        _database: AsyncIOMotorDatabase = None  

        @classmethod  
        def set_database(cls, database: AsyncIOMotorDatabase):  
            cls._database = database  

        @classmethod  
        def get_database(cls) -> AsyncIOMotorDatabase:  
            if cls._database is None:  
                raise HTTPException(status_code=500, detail="MongoDB 连接尚未建立")  
            return cls._database  

        @classmethod  
        def get_collection(cls):  
            database = cls.get_database()  
            try:  
                return database[cls.collection_name]  
            except Exception as e:  
                raise HTTPException(status_code=500, detail=f"访问集合时发生错误: {str(e)}")  

        @classmethod  
        async def create(cls, **kwargs) -> Dict[str, Any]:  
            collection = cls.get_collection()  
            result = await collection.insert_one(kwargs)  
            return {**kwargs, "_id": str(result.inserted_id)}  

        @classmethod  
        async def find(cls, id: str) -> Mapping[str, Any]:  
            collection = cls.get_collection()  
            result = await collection.find_one({"_id": ObjectId(id)})  
            if result is None:  
                raise HTTPException(status_code=404, detail="找不到该项目")  
            result["_id"] = str(result["_id"])  
            return result  

        @classmethod  
        async def all(cls) -> List[Dict[str, Any]]:  
            collection = cls.get_collection()  
            cursor = collection.find()  
            results = await cursor.to_list(length=None)  
            for result in results:  
                result["_id"] = str(result["_id"])  
            return results  

        @classmethod  
        async def update(cls, id: str, **kwargs) -> Mapping[str, Any]:  
            collection = cls.get_collection()  
            result = await collection.update_one({"_id": ObjectId(id)}, {"$set": kwargs})  
            if result.matched_count == 0:  
                raise HTTPException(status_code=404, detail="无法找到该项目")  
            return await cls.find(id)  

        @classmethod  
        async def delete(cls, id: str) -> Dict[str, bool]:  
            collection = cls.get_collection()  
            result = await collection.delete_one({"_id": ObjectId(id)})  
            if result.deleted_count == 0:  
                raise HTTPException(status_code=404, detail="无法找到该项目")  
            return {"success": True}

此文件处理与MongoDB集合之间的交互，并包含基本的 CRUD 操作（即创建（Create）、读取（Read）、更新（Update）和删除（Delete））。

现在我们来定义一下 User 模型（在 model.py 文件中）：

from mongodb import MongoDBBase  

# 用户类，表示用户数据集合名为"users"
class User(MongoDBBase):  
    collection_name = "users"

此模型继承自 MongoDBBase 类，并将 collection_name 设置为 "users"。它提供了对 MongoDB 中 users 集合的访问，并能执行用户的增删改查（CRUD）操作。

为了与 用户 模型进行互动，在 route.py 文件中添加以下代码。

    从 typing 导入 Dict, Any  
    从 fastapi 导入 APIRouter  
    从 model 导入 User 作为 UserModel  

    router = APIRouter()  

    @router.get("/users", response_model=Dict[str, Any])  
    async def index():  
        user = await UserModel.all()  
        return {  
            "data": user,  
        }

在这个部分里：

• index 函数使用 User 模型的 all() 方法从 MongoDB 集合中获取所有用户文档。

• 返回一个包含用户列表的 JSON 响应。

• 要查看结果，请访问: http://127.0.0.1:8000/user

在项目的根目录下创建一个名为.env的文件，并在其中添加您的MongoDB凭证（如用户名和密码）。

    DB_USERNAME=你的MongoDB用户名称  
    DB_PASSWORD=你的MongoDB密码名  
    DB_CLUSTER=你的MongoDB集群名  
    DB_DATABASE=你的数据库名称

以下结论:

到目前为止，你应该已经为将MongoDB与FastAPI集成设置了一个可以工作的环境。我们已经讲解了如何设置MongoDB连接、异步处理CRUD操作、创建模型以及定义路由。FastAPI和MongoDB结合可以构建现代、可扩展的API，是构建Python应用的强大组合。

Github: https://github.com/chethana101 (GitHub主页)
LinkedIn: https://www.linkedin.com/in/chethana-perera (领英主页链接)
你可以在这里找到我的GitHub主页和领英主页。