FastAPI项目中使用MongoDB做数据存储

以下代码分为两个模块实现：mongo_con.py跟mongo_client.py

mongo_con.py代码如下：

from motor.motor_asyncio import AsyncIOMotorClientclass MongoConnection:def __init__(self, uri: str):self.uri = uri  # 数据库连接 URIself.client = None  # 初始化客户端为空async def connect(self):"""连接到 MongoDB 数据库."""self.client = AsyncIOMotorClient(self.uri)  # 创建异步客户端# 可以选择连接一个特定的数据库，假设这个数据库叫 "test_db"self.database = self.client["test_db"]async def disconnect(self):"""断开与 MongoDB 的连接."""self.client.close()  # 关闭客户端连接

mongo_client.py代码如下:

from motor.motor_asyncio import AsyncIOMotorCollectionclass MongoClient:def __init__(self, database):# 使用传入的数据库实例初始化 MongoClientself.collection: AsyncIOMotorCollection = database["your_collection_name"]  # 指定要操作的集合async def insert_one(self, document):"""向集合中插入单个文档."""result = await self.collection.insert_one(document)  # 执行插入操作return result.inserted_id  # 返回插入文档的 IDasync def insert_many(self, documents):"""向集合中插入多个文档."""result = await self.collection.insert_many(documents)  # 执行插入操作return result.inserted_ids  # 返回插入文档的 ID 列表async def find_one(self, filter):"""根据过滤条件查找单个文档."""document = await self.collection.find_one(filter)  # 执行查找操作return document  # 返回查找到的文档async def find(self, filter, limit=10, skip=0):"""根据过滤条件查找多个文档，支持分页."""cursor = self.collection.find(filter).skip(skip).limit(limit)  # 执行查找操作并添加分页documents = await cursor.to_list(length=limit)  # 将游标转换为列表return documents  # 返回查找到的文档列表async def update_one(self, filter, update):"""根据过滤条件更新单个文档."""result = await self.collection.update_one(filter, update)  # 执行更新操作return result.modified_count  # 返回修改的文档数量async def update_many(self, filter, update):"""根据过滤条件更新多个文档."""result = await self.collection.update_many(filter, update)  # 执行更新操作return result.modified_count  # 返回修改的文档数量async def delete_one(self, filter):"""根据过滤条件删除单个文档."""result = await self.collection.delete_one(filter)  # 执行删除操作return result.deleted_count  # 返回删除的文档数量async def delete_many(self, filter):"""根据过滤条件删除多个文档."""result = await self.collection.delete_many(filter)  # 执行删除操作return result.deleted_count  # 返回删除的文档数量async def paginate(self, filter, page: int, size: int):"""分页查找文档."""cursor = self.collection.find(filter).skip((page - 1) * size).limit(size)  # 添加分页参数documents = await cursor.to_list(length=size)  # 将游标转换为列表return documents  # 返回查找到的文档列表

写一个小的FastAPI项目并启动调用连接数据库

from fastapi import FastAPI, HTTPException, Body
from mongo_con import MongoConnection
from contextlib import asynccontextmanager
from pydantic import BaseModel, Fieldmongo_uri = "mongodb://localhost:27017"  # 你的 MongoDB URI@asynccontextmanager
async def lifespan(app: FastAPI):"""FastAPI 应用的生命周期管理器."""mongo_conn = MongoConnection(mongo_uri)  # 创建 MongoConnection 实例await mongo_conn.connect()  # 连接到 MongoDBapp.mongodb = mongo_conn.database  # 将数据库对象附加到 app 上yield  # 暂停在此等待请求await mongo_conn.disconnect()  # 关闭连接app = FastAPI(lifespan=lifespan)# 定义用户模型
class User(BaseModel):name: str = Field(..., example="John Doe")@app.post("/users/")
async def create_user(user: User):"""插入用户接口，接受 name 字段."""existing_user = await app.mongodb.users.find_one({"name": user.name})if existing_user:raise HTTPException(status_code=400, detail="用户已存在")# 插入新用户await app.mongodb.users.insert_one({"name": user.name})return {"message": "用户创建成功", "user": user.name}@app.get("/")
async def read_root():return {"Hello": "World"}