### 使用 FastAPI 实现 CRUD 操作 #### 创建项目结构 为了更好地管理代码，建议创建清晰的文件夹和模块划分。通常会有一个 `models` 文件用于定义数据库模型，一个 `schemas` 文件用于定义 Pydantic 请求体验证模式，以及一个 `routes` 或者 `endpoints` 文件夹来放置路由逻辑。 #### 定义 SQLAlchemy 数据库模型 首先，在 `models.py` 中定义与数据库表对应的类： ```python from sqlalchemy import Column, Integer, String from database import Base class Item(Base): __tablename__ = "items" id = Column(Integer, primary_key=True, index=True) name = Column(String, unique=True, index=True) ``` 此部分描述了如何利用 SQLAlchemy ORM 来映射 Python 类到 SQL 表格[^1]。 #### 设计请求响应数据格式 接着，在 `schemas.py` 中设计 API 的输入输出 JSON 结构： ```python from pydantic import BaseModel class ItemCreate(BaseModel): name: str class ItemResponse(ItemCreate): id: int class Config: orm_mode = True ``` 上述代码片段展示了怎样借助 Pydantic 库构建有效的入参校验器，并且配置返回给客户端的数据形式[^3]。 #### 编写 CRUD 接口函数 最后一步是在主应用程序文件里编写处理 HTTP 请求的方法。下面的例子来自官方文档推荐的方式之一，它实现了四个标准的操作——增加、读取、更新和删除记录的功能： ```python from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from models import Item as ModelItem from schemas import ItemCreate, ItemResponse import crud # Assume this module contains the actual DB operations logic. router = APIRouter() @router.post("/items/", response_model=ItemResponse) def create_item(item: ItemCreate, db_session: Session = Depends(get_db)): """Creates a new item.""" return crud.create_item(db=db_session, item=item) @router.get("/items/{item_id}", response_model=ItemResponse) def read_item(item_id: int, db_session: Session = Depends(get_db)): """Reads an existing item by ID.""" db_item = crud.read_item_by_id(db=db_session, item_id=item_id) if not db_item: raise HTTPException(status_code=404, detail="Item not found") return db_item @router.put("/items/{item_id}", response_model=ItemResponse) def update_item( item_id: int, updated_data: ItemCreate, db_session: Session = Depends(get_db), ): """Updates an existing item with given data.""" db_item = crud.update_item( db=db_session, item_id=item_id, updates=updated_data.dict() ) if not db_item: raise HTTPException(status_code=404, detail="Item not found") return db_item @router.delete("/items/{item_id}") def delete_item(item_id: int, db_session: Session = Depends(get_db)): """Deletes an item based on its identifier.""" result = crud.delete_item(db=db_session, item_id=item_id) if not result: raise HTTPException(status_code=404, detail="Item not found") return {"detail": f"Deleted {result} record(s)"} ``` 这段代码说明了如何基于路径参数和查询字符串解析传入的信息并调用相应的业务层方法完成特定的任务；同时也体现了 FastAPI 对于状态码及异常情况的良好支持。