FastAPIPythonWeb APISQLAlchemy

FastAPI 学习笔记¶

这份笔记整理 FastAPI 开发中的核心知识，包括路由、参数校验、Pydantic、响应、异常处理、中间件、依赖注入、ORM、项目拆分、模型服务化和接口测试。

一、FastAPI 概述¶

FastAPI 是一个用于构建 Web API 的 Python 框架。

主要特点：

性能较高，适合构建 API 服务。
基于 Python 类型注解，能够自动完成参数解析与数据校验。
自动生成 OpenAPI 文档，并提供 Swagger UI 和 ReDoc 页面。
代码结构简洁，适合快速开发和维护。

常见使用场景：

大模型推理接口。
数据处理 API。
RESTful Web 服务。
智能体工具接口。
内部微服务。

二、基础入门¶

1. 第一个 FastAPI 程序¶

from fastapi import FastAPI


app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World"}

假设文件名是 main.py，可以使用 Uvicorn 启动：

uvicorn main:app --reload

其中：

main 对应 main.py。
app 对应文件中的 app = FastAPI()。
--reload 会在代码变化时自动重启，适合开发环境。

启动后可以访问：

Swagger UI：http://127.0.0.1:8000/docs
ReDoc：http://127.0.0.1:8000/redoc

2. 路由¶

通过装饰器将 URL、HTTP 方法和处理函数关联起来：

@app.get("/items")
async def list_items():
    return []


@app.post("/items")
async def create_item():
    return {"message": "创建成功"}


@app.put("/items/{item_id}")
async def update_item(item_id: int):
    return {"item_id": item_id, "message": "更新成功"}


@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"item_id": item_id, "message": "删除成功"}

常见方法：

方法	常见用途
`GET`	查询数据
`POST`	创建数据
`PUT`	完整更新数据
`PATCH`	部分更新数据
`DELETE`	删除数据

3. 路径参数¶

路径参数是 URL 路径的一部分：

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return {"item_id": item_id}

类型注解为 int 时，FastAPI 会自动转换和校验输入。

使用 Path 添加更多约束：

from typing import Annotated

from fastapi import Path


@app.get("/items/{item_id}")
async def get_item(
    item_id: Annotated[int, Path(title="物品 ID", ge=1, le=1000)],
):
    return {"item_id": item_id}

4. 查询参数¶

没有出现在路径中的函数参数通常会被识别为查询参数：

from typing import Annotated

from fastapi import Query


@app.get("/items")
async def list_items(
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
):
    return {"skip": skip, "limit": limit}

请求示例：

GET /items?skip=0&limit=20

5. 请求体¶

使用 Pydantic 模型描述 JSON 请求体：

from pydantic import BaseModel, Field


class ItemCreate(BaseModel):
    name: str = Field(min_length=1, max_length=50)
    price: float = Field(gt=0)
    is_offer: bool = False


@app.post("/items")
async def create_item(item: ItemCreate):
    return item

FastAPI 会根据模型：

解析 JSON 数据。
校验字段类型和取值范围。
在接口文档中生成请求体结构。
在校验失败时返回错误信息。

三、响应与异常处理¶

1. JSON 响应¶

直接返回字典、列表或 Pydantic 模型时，FastAPI 会生成 JSON 响应：

@app.get("/status")
async def get_status():
    return {"status": "running"}

2. HTML 响应¶

from fastapi.responses import HTMLResponse


@app.get("/html", response_class=HTMLResponse)
async def get_html():
    return "<h1>Hello World</h1>"

3. 文件响应¶

from fastapi.responses import FileResponse


@app.get("/file")
async def get_file():
    return FileResponse(
        path="files/report.pdf",
        filename="report.pdf",
    )

实际项目应校验文件路径和访问权限，不能直接使用未经验证的用户输入拼接路径。

4. 异常响应¶

from fastapi import HTTPException


items = {1: {"name": "示例商品"}}


@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
        )

    return items[item_id]

HTTPException 适合返回预期内的业务错误，例如资源不存在、没有权限或参数状态不允许。

四、进阶功能¶

1. 中间件¶

中间件会包围一次请求的处理过程：请求进入时执行一部分逻辑，路由处理完成后再处理响应。

常见用途：

日志记录。
请求耗时统计。
跨域处理。
统一响应头。
身份认证的全局预处理。

from time import perf_counter

from fastapi import Request


@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = perf_counter()
    response = await call_next(request)
    elapsed = perf_counter() - start
    response.headers["X-Process-Time"] = f"{elapsed:.6f}"
    return response

多个中间件会形成嵌套调用：请求按照中间件栈进入，响应按照相反方向返回。

2. 依赖注入¶

依赖注入用于复用参数处理、数据库会话、身份认证和权限检查等逻辑：

from typing import Annotated

from fastapi import Depends


async def common_parameters(
    q: str | None = None,
    skip: int = 0,
    limit: int = 100,
):
    return {"q": q, "skip": skip, "limit": limit}


@app.get("/items")
async def get_items(
    params: Annotated[dict, Depends(common_parameters)],
):
    return params

中间件和依赖注入的区别：

对比维度	中间件	依赖注入
作用范围	通常影响整个应用	可配置在接口、路由组或应用层级
调用方式	请求经过应用时自动进入	通过 `Depends()` 声明
常见用途	日志、跨域、全局响应处理	数据库会话、认证、权限、参数复用

五、SQLAlchemy ORM¶

ORM 将数据库表映射为 Python 类，使开发者可以通过对象操作数据。

1. 定义模型¶

from sqlalchemy import Float, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    price: Mapped[float] = mapped_column(Float)

2. 在依赖中管理数据库会话¶

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker


engine = create_engine("数据库连接地址")
SessionLocal = sessionmaker(bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

3. 在路由中使用会话¶

from typing import Annotated

from fastapi import Depends
from sqlalchemy import select
from sqlalchemy.orm import Session


DbSession = Annotated[Session, Depends(get_db)]


@app.get("/items")
def get_items(db: DbSession):
    statement = select(Item)
    return db.scalars(statement).all()

同步数据库驱动执行的是阻塞操作，因此这个示例使用普通 def 路由。使用异步数据库驱动时，可以改用 SQLAlchemy 的异步会话。

4. 查询数据¶

from sqlalchemy import func, select


# 查询全部
items = db.scalars(select(Item)).all()

# 比较条件
items = db.scalars(
    select(Item).where(Item.price > 100)
).all()

# 模糊查询
items = db.scalars(
    select(Item).where(Item.name.like("%手机%"))
).all()

# 聚合查询
total = db.scalar(select(func.sum(Item.price)))

# 分页查询
items = db.scalars(
    select(Item).offset(skip).limit(limit)
).all()

5. 新增数据¶

new_item = Item(name="示例手机", price=5999)
db.add(new_item)
db.commit()
db.refresh(new_item)

6. 更新数据¶

item = db.get(Item, 1)

if item is None:
    raise HTTPException(status_code=404, detail="Item not found")

item.price = 5499
db.commit()
db.refresh(item)

7. 删除数据¶

item = db.get(Item, 1)

if item is None:
    raise HTTPException(status_code=404, detail="Item not found")

db.delete(item)
db.commit()

六、项目实践规划¶

1. 用户模块¶

用户注册与登录。
JWT 身份认证。
用户权限管理。

2. 内容模块¶

内容的新增、查询、修改和删除。
内容分类与标签。

3. 收藏功能¶

收藏与取消收藏。
收藏列表查询。

4. 浏览历史¶

记录用户浏览行为。
查询和清理历史记录。

5. Redis 缓存¶

缓存热点数据。
为缓存设置合理的过期时间。
在数据更新时删除或更新对应缓存。

6. AI 问答¶

调用大模型 API。
封装智能问答接口。
处理流式输出、超时和异常。

7. 项目拆分¶

可以按照职责拆分目录：

app/
├── main.py
├── api/
├── models/
├── schemas/
├── services/
├── dependencies/
├── core/
└── tests/

拆分重点：

路由只负责接收请求和返回响应。
业务逻辑放入服务层。
数据库模型与请求响应模型分开管理。
配置从代码中分离，并区分开发和生产环境。

8. 模型服务化¶

在应用启动阶段加载模型，避免每次请求重复加载。
对推理请求设置并发限制和超时时间。
记录模型版本、输入参数和异常信息。
根据模型特性选择同步、异步或任务队列。

9. 接口测试¶

使用 Swagger UI、Apifox 或 Postman 进行手动测试。
使用自动化测试覆盖正常响应、参数错误和权限问题。
对数据库、缓存和外部模型服务使用独立的测试环境。

七、常用命令与速查¶

操作	命令或地址
开发模式启动	`uvicorn main:app --reload`
指定主机和端口	`uvicorn main:app --host 0.0.0.0 --port 8000`
Swagger UI	`http://127.0.0.1:8000/docs`
ReDoc	`http://127.0.0.1:8000/redoc`

生产环境中不应直接照搬开发模式配置，应结合进程管理、日志、反向代理、访问控制和监控策略进行部署。

返回笔记开头 ↑