当前位置：首页 > news >正文

FastAPI快速构建完成一个简单的demo，（curd）

news 2025/7/14 8:39:25

整体结构

代码主要分为四个部分：

环境设置与导入：引入所有需要的库和模块。
应用和数据初始化：创建 FastAPI 应用实例和一个内存中的 “数据库”。
数据模型定义 (Pydantic)：定义 API 请求和响应的数据结构。这是 FastAPI 的核心之一。
API 端点实现 (CURD)：编写处理创建、读取、更新、删除操作的函数。

1. 环境设置与导入

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

from fastapi import ...: 从 fastapi 库中导入核心组件。
- FastAPI: 这是创建你的 Web 应用的主类。所有的 API 路由、配置等都将附着在这个类的实例上。
- HTTPException: 这是一个特殊的 Python 异常。当你的 API 逻辑需要返回一个 HTTP 错误时（例如，“404 Not Found” 或 “400 Bad Request”），你应该 raise 这个异常。FastAPI 会捕获它并自动生成一个格式正确的 JSON 错误响应。
- status: 这是一个辅助模块，包含所有标准的 HTTP 状态码常量，如 status.HTTP_201_CREATED。使用它比直接写数字（如 201）更清晰、更不容易出错。
from pydantic import BaseModel: Pydantic 是一个数据验证和设置管理的库，是 FastAPI 的基石。
- BaseModel: 你将通过继承这个类来创建自己的数据模型。这些模型将告诉 FastAPI 你期望接收什么样的数据（请求体）以及你将返回什么样的数据（响应体）。
from typing import Optional, List: 从 Python 的标准 typing 模块中导入类型提示。
- Optional[T]: 表示一个字段可以是类型 T 或者 None。例如 Optional[str] 意味着 “一个字符串或None”。这用于定义可选字段。
- List[T]: 表示一个列表，其中每个元素都是类型 T。例如 List[Item] 意味着 “一个包含多个 Item 对象的列表”。

2. 应用和数据初始化

# 1. 初始化 FastAPI 应用
app = FastAPI(title="简单的 CURD API",description="一个使用内存数据库的 FastAPI CURD 示例。",version="1.0.0",
)# 2. 创建一个内存 "数据库"
fake_db = {1: {"id": 1, "name": "笔记本电脑", "price": 5999.99, "description": "一台高性能笔记本电脑"},2: {"id": 2, "name": "机械键盘", "price": 799.00, "description": "手感极佳的机械键盘"},
}
item_id_counter = len(fake_db)

app = FastAPI(...):
- 这一行创建了 FastAPI 应用的实例，变量名 app 是一个惯例。
- title, description, version: 这些参数不是必需的，但强烈推荐填写。它们会被用来自动生成 API 文档的头部信息。当你访问 /docs 时，你将看到这些信息。
fake_db = {...}:
- 这是一个 Python 字典，我们用它来模拟一个简单的数据库。
- 键 (Key): 1, 2 是商品的唯一 ID (item_id)。
- 值 (Value): 是另一个字典，包含了商品的详细信息。
- 注意: 在真实世界的应用中，这里会被替换为与真实数据库（如 PostgreSQL, MySQL, MongoDB）的连接和交互逻辑。
item_id_counter = len(fake_db):
- 这是一个简单的计数器，用来生成新商品的唯一 ID。每次创建一个新商品，我们就会把这个计数器加一。
- 在真实数据库中，这个功能通常由数据库的自增主键 (Auto-Increment Primary Key) 实现。

3. 数据模型定义 (Pydantic)

这是 FastAPI 最强大的功能之一。模型定义了数据的 “形状”。

# 用于创建 Item 的请求体模型
class ItemCreate(BaseModel):name: strprice: floatdescription: Optional[str] = None# 用于更新 Item 的请求体模型
class ItemUpdate(BaseModel):name: Optional[str] = Noneprice: Optional[float] = Nonedescription: Optional[str] = None# 完整的 Item 模型 (用于响应，包含 id)
class Item(BaseModel):id: intname: strprice: floatdescription: Optional[str] = None

为什么要分三个模型？
这是遵循 API 设计的最佳实践：输入和输出的数据结构往往是不同的。
1. 创建 (Create): 用户在创建商品时，不应该提供 id，因为 id 是由服务器生成的。所以 ItemCreate 模型没有 id 字段。
2. 更新 (Update): 用户在更新商品时，可能只想更新其中一个或几个字段（例如，只改价格）。所以 ItemUpdate 模型中的所有字段都是可选的 (Optional)。
3. 响应 (Response): 当服务器返回一个商品信息时，它应该是完整的，必须包含 id。所以 Item 模型包含了所有字段，且 id 是必需的。
class ItemCreate(BaseModel):
- name: str: 定义了一个名为 name 的字段，它必须是字符串。
- price: float: 定义了 price 字段，必须是浮点数。
- description: Optional[str] = None: 定义了 description 字段，它可以是一个字符串，也可以不提供（默认为 None）。

4. API 端点实现 (CURD)

Create (创建)

@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED, tags=["Items"])
def create_item(item_create: ItemCreate):# ... 函数体 ...

@app.post(...): 这是一个 “装饰器”，它告诉 FastAPI：
- "/items/": 当收到一个向 /items/ 路径发起的 HTTP POST 请求时，调用下面的 create_item 函数来处理它。
- response_model=Item: 指定了成功响应的数据模型。FastAPI 会确保函数返回的数据符合 Item 模型的结构，并自动过滤掉多余的字段。它也用此模型生成文档。
- status_code=status.HTTP_201_CREATED: 指定成功时的 HTTP 状态码为 201 Created，这是 RESTful API 中创建资源成功的标准状态码。
- tags=["Items"]: 用于在自动文档 (/docs) 中对 API进行分组。所有 tag 为 “Items” 的端点会显示在一起。
def create_item(item_create: ItemCreate):: 这是函数的定义。
- item_create: ItemCreate: 这是 FastAPI 的 “依赖注入”。它告诉 FastAPI：
  1. 期望请求体中包含一个 JSON 对象。
  2. 使用 ItemCreate 模型来解析和验证这个 JSON。
  3. 如果验证通过，将 JSON 数据转换成一个 ItemCreate 类的实例，并作为 item_create 参数传入函数。
  4. 如果 JSON 格式错误或缺少 name、price 等必填字段，FastAPI 会自动返回一个 422 Unprocessable Entity 错误，你无需编写任何验证代码。
函数体:
- global item_id_counter: 声明我们要修改全局变量 item_id_counter。
- item_id_counter += 1: ID 自增。
- new_item = Item(id=new_item_id, **item_create.model_dump()):
  - item_create.model_dump(): 将 Pydantic 模型 item_create 转换回 Python 字典。
  - ** 是 Python 的解包语法。Item(id=..., name=..., price=...) 就这样被创建出来了。
  - 我们创建了一个完整的 Item 对象（包含ID），用于返回给客户端。
- fake_db[new_item_id] = new_item.model_dump(): 将新商品存入我们的内存数据库。
- return new_item: 返回这个 Pydantic 对象。FastAPI 会自动将其序列化为 JSON 字符串并发送给客户端。

Read (读取)

# 读取所有
@app.get("/items/", response_model=List[Item], tags=["Items"])
def read_items():return list(fake_db.values())# 读取单个
@app.get("/items/{item_id}", response_model=Item, tags=["Items"])
def read_item(item_id: int):if item_id not in fake_db:raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")return fake_db[item_id]

read_items:
- @app.get(...): 处理对 /items/ 的 GET 请求。
- response_model=List[Item]: 响应将是一个列表，列表中的每个元素都符合 Item 模型。
- return list(fake_db.values()): 返回数据库中所有值的列表。FastAPI 会处理序列化。
read_item:
- "/items/{item_id}": 这是一个 “路径参数”。{item_id} 是一个占位符。
- def read_item(item_id: int): FastAPI 会自动从 URL 路径中提取 item_id 的值，并尝试将其转换为一个整数 (int)。如果转换失败（例如 URL 是 /items/abc），会自动返回 422 错误。
- if item_id not in fake_db:: 检查商品是否存在。
- raise HTTPException(...): 如果不存在，抛出 HTTPException 异常。FastAPI 会将其转换为一个包含 404 Not Found 状态码和 { "detail": "Item not found" } 消息的 JSON 响应。

Update (更新)

@app.put("/items/{item_id}", response_model=Item, tags=["Items"])
def update_item(item_id: int, item_update: ItemUpdate):# ... 函数体 ...

@app.put(...): 处理对 /items/{item_id} 的 PUT 请求。PUT 通常用于完全替换或更新资源。
item_update: ItemUpdate: 像 create 一样，FastAPI 会用 ItemUpdate 模型验证请求体。因为 ItemUpdate 的所有字段都是可选的，所以客户端可以只发送他们想更新的字段。
update_data = item_update.model_dump(exclude_unset=True):
- 这是更新操作的关键！exclude_unset=True 参数告诉 Pydantic，在转换模型为字典时，只包含那些在请求中被明确提供值的字段。
- 例如，如果请求体是 {"price": 999}，那么 update_data 就是 {'price': 999}，而不会是 {'name': None, 'price': 999, 'description': None}。这可以防止我们用 None 覆盖掉已有的 name 和 description。
updated_item = stored_item_model.model_copy(update=update_data):
- 这是一个安全的更新方式。我们先加载旧数据 (stored_item_model)，然后用 .model_copy(update=...) 方法创建一个新的、已更新的副本。update 参数会用 update_data 字典里的值覆盖旧值。
剩下的逻辑就是将更新后的数据存回数据库并返回。

Delete (删除)

@app.delete("/items/{item_id}", status_code=status.HTTP_200_OK, tags=["Items"])
def delete_item(item_id: int):# ... 函数体 ...

@app.delete(...): 处理对 /items/{item_id} 的 DELETE 请求。
status_code=status.HTTP_200_OK: 删除成功后，可以返回 200 OK 或 204 No Content。这里我们返回 200 OK 并附带一条消息。
del fake_db[item_id]: 从字典中删除指定的条目。
return {"message": "Item deleted successfully"}: 返回一个简单的 JSON 消息，告知客户端操作成功。因为没有指定 response_model，FastAPI 会直接将这个字典序列化为 JSON。

具体代码

# main.py  from fastapi import FastAPI, HTTPException, status  
from pydantic import BaseModel  
from typing import Optional, List  # 1. 初始化 FastAPI 应用  
app = FastAPI(  title="简单的 CURD API",  description="一个使用内存数据库的 FastAPI CURD 示例。",  version="1.0.0",  
)  # 2. 创建一个内存 "数据库"  
# 使用字典来存储数据，key 是 item_id，value 是 Item 对象  
fake_db = {  1: {"id": 1, "name": "笔记本电脑", "price": 5999.99, "description": "一台高性能笔记本电脑"},  2: {"id": 2, "name": "机械键盘", "price": 799.00, "description": "手感极佳的机械键盘"},  
}  
# 用于生成新的唯一 IDitem_id_counter = len(fake_db)  # 3. 定义 Pydantic 数据模型  
# Pydantic 模型用于数据验证、转换和文档生成  # 用于创建 Item 的请求体模型 (不需要传入 id)class ItemCreate(BaseModel):  name: str  price: float  description: Optional[str] = None  # 用于更新 Item 的请求体模型 (所有字段都是可选的)  
class ItemUpdate(BaseModel):  name: Optional[str] = None  price: Optional[float] = None  description: Optional[str] = None  # 完整的 Item 模型 (用于响应，包含 id)class Item(BaseModel):  id: int  name: str  price: float  description: Optional[str] = None  # 4. 实现 CURD 操作的 API 端点  # --- Create ---  
@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED, tags=["Items"])  
def create_item(item_create: ItemCreate):  """  创建一个新的商品。  - **name**: 商品名称 (必填)  - **price**: 商品价格 (必填)  - **description**: 商品描述 (可选)  """    global item_id_counter  item_id_counter += 1  new_item_id = item_id_counter  # 将 Pydantic 模型转换为字典，并添加 id    new_item = Item(id=new_item_id, **item_create.model_dump())  # 存入 "数据库"  fake_db[new_item_id] = new_item.model_dump()  return new_item  # --- Read (All) ---  
@app.get("/items/", response_model=List[Item], tags=["Items"])  
def read_items():  """  获取所有商品的列表。  """    return list(fake_db.values())  # --- Read (One) ---  
@app.get("/items/{item_id}", response_model=Item, tags=["Items"])  
def read_item(item_id: int):  """  根据 ID 获取单个商品。  """    if item_id not in fake_db:  raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")  return fake_db[item_id]  # --- Update ---  
@app.put("/items/{item_id}", response_model=Item, tags=["Items"])  
def update_item(item_id: int, item_update: ItemUpdate):  """  根据 ID 更新一个商品。  只会更新请求体中提供的字段。  """    if item_id not in fake_db:  raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")  # 获取旧数据  stored_item_data = fake_db[item_id]  stored_item_model = Item(**stored_item_data)  # 将传入的更新数据转换为字典，只包含用户设置了值的字段  update_data = item_update.model_dump(exclude_unset=True)  # 使用新数据更新旧模型  updated_item = stored_item_model.model_copy(update=update_data)  # 保存回数据库  fake_db[item_id] = updated_item.model_dump()  return updated_item  # --- Delete ---  
@app.delete("/items/{item_id}", status_code=status.HTTP_200_OK, tags=["Items"])  
def delete_item(item_id: int):  """  根据 ID 删除一个商品。  """    if item_id not in fake_db:  raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")  del fake_db[item_id]  return {"message": "Item deleted successfully"}