使用 FastAPI 和 Motor 实现用户管理 API

FastAPI 是一个现代、高性能的 Python Web 框架，非常适合构建异步 API。Motor 是一个异步的 MongoDB 驱动程序，专为 Python 的asyncio生态设计。结合 FastAPI 和 Motor，我们可以轻松构建一个高效的、基于 MongoDB 的用户管理 API。

本文将详细介绍如何使用 FastAPI 和 Motor 实现以下用户管理功能：

查询所有用户

分页查询用户

根据id查询用户

创建用户

更新用户

删除用户

1. 环境准备

安装依赖

首先，确保安装了以下依赖：

pip install fastapi motor uvicorn

fastapi: Web 框架。

motor: 异步 MongoDB 驱动程序。

uvicorn: 用于运行 FastAPI 应用的 ASGI 服务器。

2. 项目结构

以下是项目的文件结构：

user_api/

├── main.py              # FastAPI 应用入口

├── models.py            # 数据模型定义

├── crud.py              # 数据库操作逻辑

└── schemas.py           # Pydantic 模型（请求/响应格式）

3. 实现步骤

3.1 定义数据模型 (models.py)

在 MongoDB 中，用户文档的结构如下：

from pydantic import BaseModel, Field

from bson import ObjectId

from typing import Optional

# Pydantic 模型（用于请求/响应）

class UserBase(BaseModel):

  name: str

  age: int

class UserCreate(UserBase):

  pass

class UserUpdate(BaseModel):

  name: Optional[str] = None

  age: Optional[int] = None

class UserInDB(UserBase):

  id: str = Field(alias="_id")

  class Config:

      json_encoders = {ObjectId: str}  # 将 ObjectId 转换为字符串

3.2 数据库操作逻辑 (crud.py)

使用 Motor 实现与 MongoDB 的交互：

from motor.motor_asyncio import AsyncIOMotorClient

from bson import ObjectId

from models import UserInDB, UserCreate, UserUpdate

# 连接到 MongoDB

client = AsyncIOMotorClient("mongodb://localhost:27017")

db = client["user_db"]

collection = db["users"]

# 创建用户

async def create_user(user: UserCreate):

  user_dict = user.dict()

  result = await collection.insert_one(user_dict)

  new_user = await collection.find_one({"_id": result.inserted_id})

  return UserInDB(**new_user)

# 查询所有用户

async def get_all_users(skip: int = 0, limit: int = 10):

  users = []

  async for user in collection.find().skip(skip).limit(limit):

      users.append(UserInDB(**user))

  return users

# 根据 id 查询用户

async def get_user_by_id(user_id: str):

  user = await collection.find_one({"_id": ObjectId(user_id)})

  if user:

      return UserInDB(**user)

  return None

# 更新用户

async def update_user(user_id: str, user: UserUpdate):

  user_dict = user.dict(exclude_unset=True)

  await collection.update_one({"_id": ObjectId(user_id)}, {"$set": user_dict})

  updated_user = await collection.find_one({"_id": ObjectId(user_id)})

  return UserInDB(**updated_user)

# 删除用户

async def delete_user(user_id: str):

  result = await collection.delete_one({"_id": ObjectId(user_id)})

  return result.deleted_count > 0

3.3 定义 API 路由 (main.py)

使用 FastAPI 定义用户管理 API：

from fastapi import FastAPI, HTTPException, Query

from models import UserCreate, UserUpdate, UserInDB

from crud import create_user, get_all_users, get_user_by_id, update_user, delete_user

from bson import ObjectId

app = FastAPI()

# 创建用户

@app.post("/users/", response_model=UserInDB)

async def create_new_user(user: UserCreate):

  return await create_user(user)

# 查询所有用户（分页）

@app.get("/users/", response_model=list[UserInDB])

async def read_users(skip: int = Query(0, ge=0), limit: int = Query(10, le=100)):

  return await get_all_users(skip, limit)

# 根据 id 查询用户

@app.get("/users/{user_id}", response_model=UserInDB)

async def read_user(user_id: str):

  user = await get_user_by_id(user_id)

  if user is None:

      raise HTTPException(status_code=404, detail="User not found")

  return user

# 更新用户

@app.put("/users/{user_id}", response_model=UserInDB)

async def update_existing_user(user_id: str, user: UserUpdate):

  updated_user = await update_user(user_id, user)

  if updated_user is None:

      raise HTTPException(status_code=404, detail="User not found")

  return updated_user

# 删除用户

@app.delete("/users/{user_id}")

async def delete_existing_user(user_id: str):

  deleted = await delete_user(user_id)

  if not deleted:

      raise HTTPException(status_code=404, detail="User not found")

  return {"message": "User deleted successfully"}

3.4 运行应用

使用 Uvicorn 运行 FastAPI 应用：

uvicorn main:app --reload

访问http://127.0.0.1:8000/docs，可以看到自动生成的 Swagger 文档，方便测试 API。

4. 设计思路

4.1 异步编程

FastAPI 和 Motor 都支持异步编程，能够高效处理并发请求。

使用async/await语法，确保数据库操作不会阻塞事件循环。

4.2 数据模型

使用 Pydantic 模型定义请求和响应的数据结构，确保输入输出的合法性。

将 MongoDB 的ObjectId转换为字符串，方便前端处理。

4.3 分页查询

通过skip和limit参数实现分页查询，避免一次性返回大量数据。

4.4 错误处理

使用 FastAPI 的HTTPException处理用户未找到等错误情况，返回合适的 HTTP 状态码。

5. 完整代码

以下是完整的代码文件：

from pydantic import BaseModel, Field

from bson import ObjectId

from typing import Optional

class UserBase(BaseModel):

  name: str

  age: int

class UserCreate(UserBase):

  pass

class UserUpdate(BaseModel):

  name: Optional[str] = None

  age: Optional[int] = None

class UserInDB(UserBase):

  id: str = Field(alias="_id")

  class Config:

      json_encoders = {ObjectId: str}

crud.py

main.py

6. 总结

本文详细介绍了如何使用 FastAPI 和 Motor 实现一个用户管理 API，包括查询、分页、创建、更新和删除功能。通过结合 FastAPI 的高性能和 Motor 的异步 MongoDB 操作，我们可以构建一个高效、易扩展的后端服务。希望这篇文章对你有所帮助！如果有任何问题，欢迎留言讨论！