FastAPI 入门:构建一个完整的 CRUD 应用
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。关键知识点包括:
- Pydantic 模型:定义数据结构并自动验证。
- 路径操作装饰器:如
@app.post()、@app.get()等,绑定 HTTP 方法与函数。 - 路径参数与请求体:FastAPI 根据类型注解自动解析。
- 状态码与异常处理:使用
status模块和HTTPException返回标准响应。 - 自动生成文档:开箱即用的 Swagger UI 极大提升开发体验。
虽然我们使用了内存存储简化逻辑,但在实际项目中,你可以轻松替换为 SQLAlchemy(ORM)、MongoDB、或任何数据库驱动。FastAPI 的依赖注入系统也使得数据库会话管理变得非常清晰。
下一步建议:
- 集成真实数据库(如使用 SQLAlchemy + asyncpg)。
- 添加用户输入验证(如邮箱格式校验)。
- 引入身份认证(如 JWT)。
- 编写单元测试(使用
pytest+TestClient)。
FastAPI 凭借其简洁的语法、强大的类型支持和卓越的性能,已成为 Python 生态中构建现代 API 的首选框架之一。希望这篇入门教程能为你打下坚实基础!

评论 0