FastAPI 入门:构建一个完整的 CRUD 应用

余胜军说java助理
2025-12-11 15:48
阅读 833

FastAPI 入门:构建一个完整的 CRUD 应用

引言/概述

FastAPI 是一个现代、快速(高性能)的 Web 框架,用于构建 API,基于 Python 3.7+ 的类型提示(Type Hints)。它由 Starlette 提供底层支持,结合 Pydantic 实现数据验证,具有自动生成交互式文档(Swagger UI 和 ReDoc)、异步支持、依赖注入等强大特性。对于初学者而言,掌握 FastAPI 的最佳方式是从实现一个基础的 CRUD(Create, Read, Update, Delete)应用开始。

本文将带你从零开始,使用 FastAPI 构建一个管理“用户信息”的简单 RESTful API。我们将使用内存中的列表模拟数据库(便于理解),并逐步实现创建、读取、更新和删除用户的功能。通过本教程,你将掌握 FastAPI 的核心概念,如路径操作、Pydantic 模型、状态码处理、异常响应等。

注意:实际项目中应使用真实数据库(如 SQLAlchemy + PostgreSQL),但为简化入门流程,本文使用内存存储。


核心内容

环境准备与项目初始化

首先,确保你已安装 Python 3.7 或更高版本。接着,创建一个虚拟环境并安装 FastAPI 及其依赖:

python -m venv fastapi-crud-env
source fastapi-crud-env/bin/activate  # Linux/macOS
# 或 fastapi-crud-env\Scripts\activate  # Windows

pip install fastapi uvicorn
  • fastapi:主框架。
  • uvicorn:ASGI 服务器,用于运行 FastAPI 应用。

创建一个名为 main.py 的文件,作为我们的入口点。

定义数据模型(Pydantic)

FastAPI 使用 Pydantic 模型来定义请求和响应的数据结构,并自动进行数据验证和序列化。

我们先定义一个 User 模型:

# main.py
from pydantic import BaseModel
from typing import Optional

class User(BaseModel):
    id: Optional[int] = None
    name: str
    email: str
  • id 是可选的(创建时由服务端生成)。
  • name 和 email 是必填字段。

模拟数据库(内存存储)

为简化,我们使用一个 Python 列表来存储用户数据,并用一个全局变量记录下一个可用 ID:

# main.py(续)
users_db = []
next_id = 1

⚠️ 在真实应用中,请勿使用全局变量存储数据,这仅用于演示。

创建 FastAPI 应用实例

# main.py(续)
from fastapi import FastAPI, HTTPException, status

app = FastAPI(title="FastAPI CRUD 示例")

实现 Create(创建用户)

使用 POST 方法创建新用户。FastAPI 会自动将请求体解析为 User 模型。

# main.py(续)
@app.post("/users/", response_model=User, status_code=status.HTTP_201_CREATED)
def create_user(user: User):
    global next_id
    new_user = User(id=next_id, name=user.name, email=user.email)
    users_db.append(new_user)
    next_id += 1
    return new_user
  • response_model=User:指定返回的数据结构。
  • status_code=201:创建成功应返回 201 状态码。
  • 请求体通过类型注解 user: User 自动解析。

实现 Read(读取用户)

分为两种情况:获取所有用户(列表)和根据 ID 获取单个用户。

获取所有用户

@app.get("/users/", response_model=list[User])
def get_users():
    return users_db

根据 ID 获取单个用户

若用户不存在,应返回 404 错误:

@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int):
    for user in users_db:
        if user.id == user_id:
            return user
    raise HTTPException(status_code=404, detail="User not found")
  • 路径参数 user_id: int 会被自动转换为整数。
  • 使用 HTTPException 抛出标准 HTTP 错误。

实现 Update(更新用户)

使用 PUT 方法更新指定 ID 的用户信息。需验证用户是否存在。

@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, updated_user: User):
    for i, user in enumerate(users_db):
        if user.id == user_id:
            # 更新除 ID 外的字段
            users_db[i] = User(id=user_id, name=updated_user.name, email=updated_user.email)
            return users_db[i]
    raise HTTPException(status_code=404, detail="User not found")

注意:我们保留原始 ID,不允许多客户端修改 ID。

实现 Delete(删除用户)

使用 DELETE 方法移除用户:

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int):
    for i, user in enumerate(users_db):
        if user.id == user_id:
            del users_db[i]
            return  # 204 No Content 不返回 body
    raise HTTPException(status_code=404, detail="User not found")
  • 成功删除后返回 204 状态码,表示无内容返回。

完整代码整合

将上述代码整合到 main.py 中:

# main.py
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import Optional, List

app = FastAPI(title="FastAPI CRUD 示例")

class User(BaseModel):
    id: Optional[int] = None
    name: str
    email: str

# 模拟数据库
users_db = []
next_id = 1

@app.post("/users/", response_model=User, status_code=status.HTTP_201_CREATED)
def create_user(user: User):
    global next_id
    new_user = User(id=next_id, name=user.name, email=user.email)
    users_db.append(new_user)
    next_id += 1
    return new_user

@app.get("/users/", response_model=List[User])
def get_users():
    return users_db

@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int):
    for user in users_db:
        if user.id == user_id:
            return user
    raise HTTPException(status_code=404, detail="User not found")

@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, updated_user: User):
    for i, user in enumerate(users_db):
        if user.id == user_id:
            users_db[i] = User(id=user_id, name=updated_user.name, email=updated_user.email)
            return users_db[i]
    raise HTTPException(status_code=404, detail="User not found")

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_user(user_id: int):
    for i, user in enumerate(users_db):
        if user.id == user_id:
            del users_db[i]
            return
    raise HTTPException(status_code=404, detail="User not found")

运行与测试

在终端中运行应用:

uvicorn main:app --reload
  • --reload:开发模式下代码变更自动重启。

启动后,访问 http://localhost:8000/docs,即可看到自动生成的 Swagger UI 文档。你可以直接在网页上测试所有 CRUD 接口:

  • POST /users/:创建用户(输入 JSON 如 {"name": "Alice", "email": "alice@example.com"})
  • GET /users/:获取所有用户
  • GET /users/1:获取 ID 为 1 的用户
  • PUT /users/1:更新 ID 为 1 的用户
  • DELETE /users/1:删除 ID 为 1 的用户

总结

通过本文,我们使用 FastAPI 快速构建了一个具备完整 CRUD 功能的 RESTful API。关键知识点包括:

  1. Pydantic 模型:定义数据结构并自动验证。
  2. 路径操作装饰器:如 @app.post()、@app.get() 等,绑定 HTTP 方法与函数。
  3. 路径参数与请求体:FastAPI 根据类型注解自动解析。
  4. 状态码与异常处理:使用 status 模块和 HTTPException 返回标准响应。
  5. 自动生成文档:开箱即用的 Swagger UI 极大提升开发体验。

虽然我们使用了内存存储简化逻辑,但在实际项目中,你可以轻松替换为 SQLAlchemy(ORM)、MongoDB、或任何数据库驱动。FastAPI 的依赖注入系统也使得数据库会话管理变得非常清晰。

下一步建议:

  • 集成真实数据库(如使用 SQLAlchemy + asyncpg)。
  • 添加用户输入验证(如邮箱格式校验)。
  • 引入身份认证(如 JWT)。
  • 编写单元测试(使用 pytest + TestClient)。

FastAPI 凭借其简洁的语法、强大的类型支持和卓越的性能,已成为 Python 生态中构建现代 API 的首选框架之一。希望这篇入门教程能为你打下坚实基础!

评论 0

最热最新
暂无评论
余胜军说java助理Lv.1
0
影响力
0
文章
0
粉丝
聊天室