返回文章列表

FastAPI现代Web API开发入门

FastAPI是近年来最受关注的Python Web框架之一。它基于现代Python特性,性能优异,开发效率高,自动生成文档。本教程将带你从零开始学习FastAPI,构建高性能的Web API。

一、FastAPI简介与安装

FastAPI是一个现代、快速(高性能)的Web框架,基于Python 3.6+的标准类型提示。它的主要特点包括:

安装

# 安装FastAPI和ASGI服务器uvicorn
pip install fastapi uvicorn

# 安装完整依赖(包含额外功能)
pip install "fastapi[all]"

# 如果需要表单支持
pip install python-multipart

二、第一个API路由

让我们从一个最简单的"Hello World"开始。

# main.py - 第一个FastAPI应用
from fastapi import FastAPI

# 创建FastAPI应用实例
app = FastAPI(
    title="我的第一个API",
    description="学习FastAPI的示例项目",
    version="1.0.0"
)

# 定义根路由
@app.get("/")
async def root():
    """根路径,返回欢迎信息"""
    return {"message": "Hello, FastAPI!"}

# 定义另一个路由
@app.get("/hello/{name}")
async def say_hello(name: str):
    """带路径参数的路由"""
    return {"message": f"Hello, {name}!"}

运行应用

# 使用uvicorn启动开发服务器
# --reload 参数会在代码修改后自动重启
uvicorn main:app --reload

# 指定主机和端口
uvicorn main:app --host 0.0.0.0 --port 8000 --reload

# 访问 http://127.0.0.1:8000 查看应用
# 访问 http://127.0.0.1:8000/docs 查看自动生成的Swagger文档
# 访问 http://127.0.0.1:8000/redoc 查看ReDoc文档

三、路径参数与查询参数

FastAPI通过Python的类型提示自动处理参数解析和验证。

# parameters.py - 参数处理示例
from fastapi import FastAPI, Query
from typing import Optional
from enum import Enum

app = FastAPI()

# ===== 路径参数 =====
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    """
    路径参数,自动类型转换和验证
    如果传入非整数,FastAPI会返回422错误
    """
    return {"item_id": item_id}

# 使用枚举限定参数值
class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    """使用枚举限制路径参数的可选值"""
    return {"model": model_name, "message": f"你选择了 {model_name.value} 模型"}

# 路径顺序很重要:更具体的路径要放在前面
@app.get("/users/me")
async def read_user_me():
    """这个路由要在 /users/{user_id} 之前定义"""
    return {"user_id": "当前用户"}

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}


# ===== 查询参数 =====
@app.get("/items/")
async def list_items(
    skip: int = 0,                    # 有默认值的查询参数
    limit: int = 10,                  # 有默认值的查询参数
    q: Optional[str] = None           # 可选查询参数
):
    """
    查询参数示例
    访问 /items/?skip=0&limit=10&q=python
    """
    result = {"skip": skip, "limit": limit}
    if q:
        result["q"] = q
    return result

# 使用Query进行更精细的验证
@app.get("/search/")
async def search(
    q: str = Query(
        ...,                          # ...表示必填参数
        min_length=3,                 # 最小长度
        max_length=50,                # 最大长度
        title="搜索关键词",
        description="要搜索的内容关键词",
        regex="^[a-zA-Z0-9]+$"       # 正则验证
    ),
    category: Optional[str] = Query(
        None,
        regex="^(book|movie|music)$"  # 限定可选值
    )
):
    """带验证的查询参数"""
    return {"q": q, "category": category}

# 查询参数列表(接收多个值)
@app.get("/tags/")
async def get_tags(tags: list[str] = Query([])):
    """
    接收多个同名查询参数
    访问 /tags/?tags=python&tags=java&tags=go
    """
    return {"tags": tags}

四、请求体与Pydantic模型

请求体用于接收客户端发送的数据(通常是POST/PUT请求)。FastAPI使用Pydantic模型来定义和验证请求体。

# models.py - Pydantic模型定义
from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum

# 基础模型
class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100,
                     description="物品名称")
    description: Optional[str] = Field(None, max_length=500,
                                       description="物品描述")
    price: float = Field(..., gt=0, description="价格,必须大于0")
    tax: Optional[float] = Field(None, ge=0, description="税率")

# 创建物品时的模型
class ItemCreate(ItemBase):
    """创建物品请求模型,继承自ItemBase"""
    pass

# 物品响应模型(包含id和创建时间)
class ItemResponse(ItemBase):
    """物品响应模型"""
    id: int
    created_at: datetime
    is_active: bool = True

    class Config:
        orm_mode = True  # 允许从ORM对象读取数据

# 用户模型示例
class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=20)
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    password: str = Field(..., min_length=6, max_length=100)

    @validator('password')
    def password_must_contain_number(cls, v):
        if not any(c.isdigit() for c in v):
            raise ValueError('密码必须包含至少一个数字')
        return v

class UserResponse(UserBase):
    id: int
    is_active: bool = True
    created_at: datetime

    class Config:
        orm_mode = True

# 嵌套模型
class OrderItem(BaseModel):
    item_id: int
    quantity: int = Field(..., gt=0)

class OrderCreate(BaseModel):
    user_id: int
    items: List[OrderItem] = Field(..., min_items=1)
    note: Optional[str] = None

class OrderResponse(BaseModel):
    id: int
    user_id: int
    items: List[OrderItem]
    total_price: float
    created_at: datetime
    status: str
# request_body.py - 请求体处理
from fastapi import FastAPI, HTTPException, status
from models import ItemCreate, ItemResponse, UserCreate, UserResponse
from datetime import datetime

app = FastAPI()

# 模拟数据库
fake_items_db = []
fake_users_db = []

# ===== POST请求:创建资源 =====
@app.post("/items/",
          response_model=ItemResponse,
          status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate):
    """
    创建新物品
    - 接收ItemCreate模型作为请求体
    - 返回ItemResponse模型
    - 状态码为201 Created
    """
    # 生成新ID
    new_id = len(fake_items_db) + 1

    # 创建响应数据
    item_data = item.dict()
    item_data.update({
        "id": new_id,
        "created_at": datetime.now(),
        "is_active": True
    })

    # 保存到数据库
    fake_items_db.append(item_data)

    return item_data

# ===== GET请求:获取资源 =====
@app.get("/items/", response_model=list[ItemResponse])
async def list_items(skip: int = 0, limit: int = 10):
    """获取物品列表"""
    return fake_items_db[skip: skip + limit]

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int):
    """获取单个物品"""
    for item in fake_items_db:
        if item["id"] == item_id:
            return item
    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"物品 {item_id} 不存在"
    )

# ===== PUT请求:更新资源 =====
@app.put("/items/{item_id}", response_model=ItemResponse)
async def update_item(item_id: int, item: ItemCreate):
    """更新物品"""
    for i, existing_item in enumerate(fake_items_db):
        if existing_item["id"] == item_id:
            # 更新数据
            item_data = item.dict()
            item_data.update({
                "id": item_id,
                "created_at": existing_item["created_at"],
                "is_active": existing_item["is_active"]
            })
            fake_items_db[i] = item_data
            return item_data

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"物品 {item_id} 不存在"
    )

# ===== DELETE请求:删除资源 =====
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    """删除物品"""
    for i, item in enumerate(fake_items_db):
        if item["id"] == item_id:
            fake_items_db.pop(i)
            return  # 204 No Content不需要返回body

    raise HTTPException(
        status_code=status.HTTP_404_NOT_FOUND,
        detail=f"物品 {item_id} 不存在"
    )

五、响应模型

响应模型控制API返回的数据结构,可以过滤敏感字段,保证返回数据的一致性。

# response_models.py - 响应模型示例
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime

app = FastAPI()

# 用户数据库模型(包含敏感信息)
class UserInDB(BaseModel):
    id: int
    username: str
    email: str
    hashed_password: str  # 敏感信息
    is_active: bool
    created_at: datetime

# 用户输入模型
class UserIn(BaseModel):
    username: str
    email: str
    password: str

# 用户输出模型(不含密码)
class UserOut(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool
    created_at: datetime

# 模拟数据库
fake_db = [
    UserInDB(
        id=1,
        username="admin",
        email="admin@example.com",
        hashed_password="$2b$12$xxxx",
        is_active=True,
        created_at=datetime.now()
    )
]

@app.post("/users/", response_model=UserOut)
async def create_user(user: UserIn):
    """
    创建用户
    请求体使用UserIn(包含明文密码)
    响应体使用UserOut(不包含密码)
    FastAPI会自动过滤掉不在UserOut中的字段
    """
    # 模拟密码哈希
    hashed_password = f"$2b$12${user.password}_hashed"

    new_user = UserInDB(
        id=len(fake_db) + 1,
        username=user.username,
        email=user.email,
        hashed_password=hashed_password,
        is_active=True,
        created_at=datetime.now()
    )
    fake_db.append(new_user)

    # 返回时,hashed_password会被自动过滤掉
    return new_user

@app.get("/users/", response_model=List[UserOut])
async def list_users():
    """获取用户列表,响应中不会包含密码"""
    return fake_db

@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
    """获取单个用户"""
    for user in fake_db:
        if user.id == user_id:
            return user
    from fastapi import HTTPException
    raise HTTPException(status_code=404, detail="用户不存在")

六、依赖注入

依赖注入是FastAPI的强大特性,用于共享逻辑(如数据库连接、认证、分页参数等)。

# dependencies.py - 依赖注入示例
from fastapi import FastAPI, Depends, HTTPException, Header
from pydantic import BaseModel
from typing import Optional
import os

app = FastAPI()

# ===== 1. 简单依赖 =====
def common_parameters(
    skip: int = 0,
    limit: int = 10,
    q: Optional[str] = None
):
    """
    公共查询参数依赖
    多个路由可以共享这段逻辑
    """
    return {"skip": skip, "limit": limit, "q": q}

@app.get("/items/")
async def list_items(commons: dict = Depends(common_parameters)):
    return {"message": "获取物品列表", "params": commons}

@app.get("/users/")
async def list_users(commons: dict = Depends(common_parameters)):
    return {"message": "获取用户列表", "params": commons}

# ===== 2. 数据库连接依赖 =====
class Database:
    def __init__(self):
        self.connected = False

    def connect(self):
        self.connected = True
        print("数据库已连接")

    def disconnect(self):
        self.connected = False
        print("数据库已断开")

    def query(self, sql: str):
        if not self.connected:
            raise RuntimeError("数据库未连接")
        return [{"id": 1, "name": "示例数据"}]

db = Database()

def get_db():
    """数据库依赖:在请求开始时连接,结束后断开"""
    db.connect()
    try:
        yield db  # yield使依赖变为生成器,可以在请求结束后执行清理
    finally:
        db.disconnect()

@app.get("/db-test/")
async def db_test(database: Database = Depends(get_db)):
    """使用数据库依赖"""
    result = database.query("SELECT * FROM items")
    return {"data": result}

# ===== 3. 认证依赖 =====
API_KEY = os.getenv("API_KEY", "secret-key-123")

async def verify_api_key(x_api_key: Optional[str] = Header(None)):
    """
    API密钥验证依赖
    检查请求头中的X-API-Key
    """
    if x_api_key != API_KEY:
        raise HTTPException(
            status_code=401,
            detail="无效的API密钥",
            headers={"WWW-Authenticate": "ApiKey"}
        )
    return x_api_key

@app.get("/protected/")
async def protected_route(api_key: str = Depends(verify_api_key)):
    """需要认证才能访问的路由"""
    return {"message": "你已通过认证", "api_key": api_key[:8] + "..."}

# ===== 4. 全局依赖 =====
# 对整个应用生效的依赖
app_with_auth = FastAPI(dependencies=[Depends(verify_api_key)])

# ===== 5. 类作为依赖 =====
class Pagination:
    """分页依赖类"""
    def __init__(self, page: int = 1, size: int = 10):
        self.page = max(1, page)
        self.size = min(max(1, size), 100)  # 限制每页最多100条
        self.offset = (self.page - 1) * self.size

    @property
    def limit(self):
        return self.size

@app.get("/articles/")
async def list_articles(pagination: Pagination = Depends()):
    """使用类依赖进行分页"""
    return {
        "page": pagination.page,
        "size": pagination.size,
        "offset": pagination.offset,
        "message": f"获取第{pagination.page}页,每页{pagination.size}条"
    }

七、异常处理

# exception_handling.py - 异常处理
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()

# ===== 1. HTTP异常 =====
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in [1, 2, 3]:
        raise HTTPException(
            status_code=404,
            detail="物品不存在",
            headers={"X-Error": "Item Not Found"}
        )
    return {"item_id": item_id}

# ===== 2. 自定义异常 =====
class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

# 注册异常处理器
@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={
            "message": f"哎呀,{exc.name} 出了点问题",
            "request_url": str(request.url)
        }
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "error":
        raise UnicornException(name=name)
    return {"name": name}

# ===== 3. 全局异常处理 =====
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """捕获所有未处理的异常"""
    return JSONResponse(
        status_code=500,
        content={
            "message": "服务器内部错误",
            "detail": str(exc),
            "path": str(request.url.path)
        }
    )

# ===== 4. 验证错误处理 =====
from fastapi.exceptions import RequestValidationError
from fastapi.encoders import jsonable_encoder

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    """自定义验证错误响应格式"""
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "message": "参数验证失败",
            "errors": jsonable_encoder(exc.errors()),
            "path": str(request.url.path)
        }
    )

八、自动文档(Swagger UI)

FastAPI自动生成API文档,这是它最受欢迎的特性之一。

# docs_example.py - 文档配置示例
from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI(
    title="待办事项API",
    description="""
    ## 这是一个待办事项管理API

    ### 功能特性
    * 创建待办事项
    * 查询待办列表
    * 更新事项状态
    * 删除事项

    ### 使用说明
    1. 先创建用户
    2. 获取认证token
    3. 使用token操作待办事项
    """,
    version="1.0.0",
    terms_of_service="https://example.com/terms/",
    contact={
        "name": "API支持",
        "url": "https://example.com/support",
        "email": "support@example.com",
    },
    license_info={
        "name": "MIT License",
        "url": "https://opensource.org/licenses/MIT",
    },
    docs_url="/docs",       # Swagger UI路径
    redoc_url="/redoc",     # ReDoc路径
    openapi_url="/openapi.json"  # OpenAPI Schema路径
)

class TodoCreate(BaseModel):
    """创建待办事项的模型"""
    title: str = Field(
        ...,
        title="标题",
        description="待办事项的标题",
        min_length=1,
        max_length=100,
        example="买牛奶"
    )
    description: str = Field(
        "",
        title="描述",
        description="待办事项的详细描述",
        max_length=500,
        example="去楼下超市买2升装的全脂牛奶"
    )
    priority: int = Field(
        1,
        title="优先级",
        description="优先级(1-5),5为最高",
        ge=1,
        le=5,
        example=3
    )

class TodoResponse(BaseModel):
    """待办事项响应模型"""
    id: int = Field(..., example=1)
    title: str = Field(..., example="买牛奶")
    description: str = Field("", example="去楼下超市买2升装的全脂牛奶")
    priority: int = Field(..., example=3)
    completed: bool = Field(False, example=False)

    class Config:
        json_schema_extra = {
            "example": {
                "id": 1,
                "title": "买牛奶",
                "description": "去楼下超市买2升装的全脂牛奶",
                "priority": 3,
                "completed": False
            }
        }

@app.post(
    "/todos/",
    response_model=TodoResponse,
    summary="创建待办事项",
    description="创建一个新的待办事项,返回创建后的完整信息。",
    response_description="创建成功的待办事项",
    tags=["待办事项"],
    status_code=201
)
async def create_todo(todo: TodoCreate):
    """
    创建一个新的待办事项。

    参数:
    - **title**: 标题(必填,1-100字符)
    - **description**: 描述(可选,最多500字符)
    - **priority**: 优先级(1-5,默认1)

    返回创建成功的待办事项,包含分配的ID。
    """
    # 实际实现中这里会保存到数据库
    return {
        "id": 1,
        "title": todo.title,
        "description": todo.description,
        "priority": todo.priority,
        "completed": False
    }

@app.get(
    "/todos/",
    response_model=list[TodoResponse],
    summary="获取待办列表",
    tags=["待办事项"]
)
async def list_todos():
    """获取所有待办事项列表"""
    return [
        {"id": 1, "title": "买牛奶", "description": "", "priority": 3, "completed": False},
        {"id": 2, "title": "写代码", "description": "完成API开发", "priority": 5, "completed": False},
    ]

# 访问 /docs 可以看到带标签分组的、有详细描述的Swagger文档

九、实战案例:待办事项CRUD API

综合以上知识,构建一个完整的待办事项CRUD API。

# todo_app.py - 完整的待办事项API
from fastapi import FastAPI, HTTPException, Depends, status, Query
from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetime
from enum import Enum

# ===== 配置 =====
app = FastAPI(
    title="待办事项API",
    description="一个完整的待办事项CRUD API示例",
    version="1.0.0"
)

# ===== 枚举 =====
class Priority(int, Enum):
    LOW = 1
    MEDIUM = 3
    HIGH = 5

class TodoStatus(str, Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"

# ===== 模型 =====
class TodoBase(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    description: str = Field("", max_length=500)
    priority: Priority = Priority.MEDIUM

class TodoCreate(TodoBase):
    pass

class TodoUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = Field(None, max_length=500)
    priority: Optional[Priority] = None
    status: Optional[TodoStatus] = None

class TodoResponse(TodoBase):
    id: int
    status: TodoStatus
    created_at: datetime
    updated_at: datetime

    class Config:
        json_schema_extra = {
            "example": {
                "id": 1,
                "title": "学习FastAPI",
                "description": "完成FastAPI教程",
                "priority": 3,
                "status": "pending",
                "created_at": "2024-01-15T10:00:00",
                "updated_at": "2024-01-15T10:00:00"
            }
        }

# ===== 模拟数据库 =====
class TodoDatabase:
    def __init__(self):
        self.todos: dict[int, dict] = {}
        self.next_id = 1

    def get_all(self) -> List[dict]:
        return list(self.todos.values())

    def get_by_id(self, todo_id: int) -> Optional[dict]:
        return self.todos.get(todo_id)

    def create(self, todo_data: dict) -> dict:
        todo_id = self.next_id
        now = datetime.now()
        todo = {
            "id": todo_id,
            "status": TodoStatus.PENDING,
            "created_at": now,
            "updated_at": now,
            **todo_data
        }
        self.todos[todo_id] = todo
        self.next_id += 1
        return todo

    def update(self, todo_id: int, update_data: dict) -> Optional[dict]:
        if todo_id not in self.todos:
            return None
        self.todos[todo_id].update(update_data)
        self.todos[todo_id]["updated_at"] = datetime.now()
        return self.todos[todo_id]

    def delete(self, todo_id: int) -> bool:
        return self.todos.pop(todo_id, None) is not None

db = TodoDatabase()

# ===== 依赖 =====
def get_todo_or_404(todo_id: int) -> dict:
    """根据ID获取待办事项,不存在则返回404"""
    todo = db.get_by_id(todo_id)
    if not todo:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"待办事项 {todo_id} 不存在"
        )
    return todo

# ===== 路由 =====
@app.post(
    "/todos/",
    response_model=TodoResponse,
    status_code=status.HTTP_201_CREATED,
    summary="创建待办事项",
    tags=["待办事项"]
)
async def create_todo(todo: TodoCreate):
    """创建一个新的待办事项"""
    created = db.create(todo.dict())
    return created

@app.get(
    "/todos/",
    response_model=List[TodoResponse],
    summary="获取待办列表",
    tags=["待办事项"]
)
async def list_todos(
    status_filter: Optional[TodoStatus] = Query(None, alias="status"),
    priority_min: Optional[Priority] = Query(None, alias="priority_min"),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    """
    获取待办事项列表,支持过滤和分页

    - **status**: 按状态过滤
    - **priority_min**: 最低优先级
    - **skip**: 跳过记录数
    - **limit**: 返回记录数
    """
    todos = db.get_all()

    # 过滤
    if status_filter:
        todos = [t for t in todos if t["status"] == status_filter]
    if priority_min:
        todos = [t for t in todos if t["priority"] >= priority_min]

    # 分页
    todos = todos[skip: skip + limit]

    return todos

@app.get(
    "/todos/{todo_id}",
    response_model=TodoResponse,
    summary="获取单个待办事项",
    tags=["待办事项"]
)
async def get_todo(todo: dict = Depends(get_todo_or_404)):
    """根据ID获取待办事项详情"""
    return todo

@app.put(
    "/todos/{todo_id}",
    response_model=TodoResponse,
    summary="更新待办事项",
    tags=["待办事项"]
)
async def update_todo(
    todo_id: int,
    todo_update: TodoUpdate,
    existing: dict = Depends(get_todo_or_404)
):
    """
    更新待办事项

    所有字段都是可选的,只更新提供的字段
    """
    # 只更新非None的字段
    update_data = todo_update.dict(exclude_unset=True)
    if not update_data:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="没有提供要更新的字段"
        )

    updated = db.update(todo_id, update_data)
    return updated

@app.patch(
    "/todos/{todo_id}/status",
    response_model=TodoResponse,
    summary="更新待办状态",
    tags=["待办事项"]
)
async def update_todo_status(
    todo_id: int,
    new_status: TodoStatus,
    todo: dict = Depends(get_todo_or_404)
):
    """快速更新待办事项的状态"""
    updated = db.update(todo_id, {"status": new_status})
    return updated

@app.delete(
    "/todos/{todo_id}",
    status_code=status.HTTP_204_NO_CONTENT,
    summary="删除待办事项",
    tags=["待办事项"]
)
async def delete_todo(
    todo_id: int,
    todo: dict = Depends(get_todo_or_404)
):
    """删除指定的待办事项"""
    db.delete(todo_id)
    return None

# ===== 统计接口 =====
@app.get(
    "/todos/stats/summary",
    summary="获取统计信息",
    tags=["统计"]
)
async def get_stats():
    """获取待办事项的统计信息"""
    todos = db.get_all()
    total = len(todos)

    status_count = {"pending": 0, "in_progress": 0, "completed": 0}
    priority_count = {1: 0, 3: 0, 5: 0}

    for todo in todos:
        status_count[todo["status"]] += 1
        priority_count[todo["priority"]] += 1

    completion_rate = (status_count["completed"] / total * 100) if total > 0 else 0

    return {
        "total": total,
        "by_status": status_count,
        "by_priority": priority_count,
        "completion_rate": round(completion_rate, 2)
    }

# 启动命令: uvicorn todo_app:app --reload

总结

本教程涵盖了FastAPI的核心知识:

  1. 了解了FastAPI的特点和优势
  2. 学会了创建第一个API路由
  3. 掌握了路径参数和查询参数的使用和验证
  4. 学会了使用Pydantic模型定义请求体
  5. 理解了响应模型的作用和数据过滤
  6. 掌握了依赖注入的使用方法
  7. 学会了异常处理和自定义错误响应
  8. 了解了自动文档功能的使用和配置
  9. 通过待办事项CRUD API实战综合运用了所有知识

进阶建议:

动手挑战

学到这里,不妨动手试一试以下练习,巩固你的理解:

  1. 基础练习:回顾本文核心概念,用自己的话总结关键知识点。
  2. 进阶实践:将文中的示例代码运行一遍,尝试修改参数观察变化。
  3. 拓展思考:想一想这个技术/方法还能应用在哪些场景中?

小贴士:遇到问题时,先独立思考,再查阅资料,最后请教他人——这是成长最快的学习方式。

赞赏支持

本文更新于 2026-08-22,环境 Python 3.12