笔记 / FastAPI · 更新于 2026-10-15

FastAPI

1. 环境与项目初始化

1.1 依赖安装（Windows + Python 3.12）

# 基础安装（推荐镜像）
pip install fastapi==0.138.0 fastapi-cli==0.0.20 fastapi-core==0.138.0 "uvicorn[standard]" -i https://pypi.tuna.tsinghua.edu.cn/simple

# 表单/文件上传必需
pip install python-multipart==0.0.9

# 数据库（同步方案）
pip install PyMySQL==1.1.0 SQLAlchemy==2.0.28 cryptography==42.0.5

# 异步数据库方案（可选，见后）
pip install aiomysql==0.2.0 sqlalchemy[asyncio]  # 或直接用 asyncpg + SQLAlchemy异步

1.2 启动命令

uvicorn main:app --reload --host 127.0.0.1 --port 8000
# 或
fastapi dev main.py

1.3 PyCharm 运行配置（可选）

2. FastAPI 应用基础

2.1 最小应用

from fastapi import FastAPI

app = FastAPI(title="My API", version="1.0.0")

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

2.2 自动文档访问

3. 路由与参数

3.1 HTTP 方法装饰器

@app.get("/items")      # 读取
@app.post("/items")     # 创建
@app.put("/items/{id}") # 完整更新
@app.patch("/items/{id}") # 部分更新
@app.delete("/items/{id}") # 删除

3.2 路径参数（Path Parameters）

@app.get("/books/{book_id}")
def get_book(book_id: int):   # 自动转换类型，非数字返回422
    return {"id": book_id}

高级校验（使用 Path）：

from fastapi import Path

@app.get("/books/{book_id}")
def get_book(
    book_id: int = Path(..., gt=0, le=10000, title="图书ID", example=1)
):
    return {"id": book_id}

... 表示必填；gt/ge/lt/le 用于数值范围。

路由顺序：静态路径必须写在动态路径之前（否则 /users/me 会被 /users/{user_id} 匹配）。

3.3 查询参数（Query Parameters）

未在路径中声明的参数自动成为查询参数。

@app.get("/items")
def list_items(page: int = 1, limit: int = 10, q: str | None = None):
    return {"page": page, "limit": limit, "q": q}

高级校验（使用 Query）：

from fastapi import Query

@app.get("/search")
def search(
    keyword: str | None = Query(None, min_length=1, max_length=20),
    min_price: float = Query(0, ge=0)
):
    ...

4. 请求体（Request Body）

4.1 使用 Pydantic 模型

from pydantic import BaseModel

class BookCreate(BaseModel):
    title: str
    author: str
    price: float = Field(..., gt=0)
    is_published: bool = True

@app.post("/books")
def create_book(book: BookCreate):
    return book.model_dump()  # v2 转 dict

4.2 字段校验（Field）

from pydantic import Field

class Book(BaseModel):
    title: str = Field(..., min_length=2, max_length=50)
    price: float = Field(..., gt=0, le=1000)
    tags: list[str] = Field(default=[], max_length=5)

4.3 请求体 + 路径 + 查询参数混用

FastAPI 自动根据类型区分： - 路径中占位符 → 路径参数 - 简单类型（未在路径中）→ 查询参数 - BaseModel 子类 → 请求体

@app.put("/books/{book_id}")
def update(book_id: int, book: BookCreate, notify: bool = False):
    ...

4.4 单一简单值请求体（使用 Body）

from fastapi import Body

@app.post("/importance")
def set_importance(importance: int = Body(..., embed=True, ge=1, le=5)):
    ...

embed=True 要求 JSON 必须为 {"importance": 5}。

4.5 表单数据（Form）

from fastapi import Form

@app.post("/login")
def login(username: str = Form(...), password: str = Form(...)):
    ...

必须安装 python-multipart。

4.6 文件上传（UploadFile）

from fastapi import File, UploadFile
import shutil

@app.post("/upload")
def upload(file: UploadFile = File(...)):
    # 校验文件类型
    if file.content_type not in ["image/jpeg", "image/png"]:
        raise HTTPException(400, "只允许jpg/png")
    # 保存
    with open(f"uploads/{file.filename}", "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    return {"filename": file.filename}

UploadFile vs bytes：UploadFile 使用缓冲流，节省内存，推荐。

多文件上传：

@app.post("/upload-multiple")
def upload_multiple(files: list[UploadFile] = File(...)):
    return [f.filename for f in files]

5. 响应模型与状态码

5.1 响应模型（response_model）

用于过滤敏感字段，只返回指定字段。

class UserPublic(BaseModel):
    id: int
    username: str

class UserDB(UserPublic):
    hashed_password: str

@app.get("/users/{id}", response_model=UserPublic)
def get_user(id: int):
    user = UserDB(id=1, username="alice", hashed_password="xxx")
    return user  # 自动过滤掉 hashed_password

5.2 状态码

from fastapi import status

@app.post("/items", status_code=status.HTTP_201_CREATED)
def create():
    return {"id": 1}

5.3 非 JSON 响应

from fastapi.responses import HTMLResponse
@app.get("/page", response_class=HTMLResponse)
def page():
    return "<h1>Hello</h1>"

from fastapi.responses import StreamingResponse
def generate():
    yield "chunk1"
    yield "chunk2"
@app.get("/stream")
def stream():
    return StreamingResponse(generate(), media_type="text/plain")

5.4 自定义响应头

@app.get("/headers")
def headers(response: Response):
    response.headers["X-Custom"] = "value"
    return {"ok": True}

5.5 读取请求头

from fastapi import Header

@app.get("/read-header")
def read(user_agent: str | None = Header(None), x_token: str | None = Header(None, alias="X-Auth-Token")):
    ...

6. 异常处理

6.1 抛出 HTTPException

from fastapi import HTTPException, status

@app.get("/items/{id}")
def get(id: int):
    if id not in db:
        raise HTTPException(status_code=404, detail="Item not found", headers={"X-Error": "missing"})
    return db[id]

6.2 自定义异常与全局处理器

class BusinessException(Exception):
    def __init__(self, code: int, msg: str):
        self.code = code
        self.msg = msg

@app.exception_handler(BusinessException)
def handle_biz(req, exc):
    return JSONResponse(status_code=400, content={"code": exc.code, "msg": exc.msg})

@app.get("/biz")
def biz(score: int):
    if score < 60:
        raise BusinessException(1001, "不及格")
    return {"ok": True}

7. 异步编程基础

7.1 同步 vs 异步

import asyncio

@app.get("/sync")
def sync_task():
    time.sleep(2)          # 阻塞，但被线程池隔离
    return {"ok": True}

@app.get("/async")
async def async_task():
    await asyncio.sleep(2) # 非阻塞，释放事件循环
    return {"ok": True}

黄金法则：使用 await 异步库时用 async def，否则用普通 def。

8. 依赖注入（Depends）

8.1 基本依赖

from fastapi import Depends, Header

def verify_token(x_token: str = Header(...)):
    if x_token != "secret":
        raise HTTPException(401)
    return x_token

@app.get("/admin")
def admin(token: str = Depends(verify_token)):
    return {"token": token}

8.2 依赖嵌套

def pagination(page: int = 1, size: int = 10):
    return {"skip": (page-1)*size, "limit": size}

def search_context(q: str | None = None, pag: dict = Depends(pagination)):
    return {"q": q, **pag}

@app.get("/search")
def search(ctx: dict = Depends(search_context)):
    return ctx

8.3 使用 yield 管理资源（数据库会话）

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

@app.get("/items")
def read_items(db = Depends(get_db)):
    return db.query(Item).all()

9. 数据库集成（MySQL + SQLAlchemy）

9.1 同步方案（教案原版）

目录结构

project/
├── database.py    # 引擎、会话、Base
├── models.py      # ORM 模型
├── schemas.py     # Pydantic 模型
└── main.py        # 路由

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:123456@127.0.0.1:3306/school_db?charset=utf8mb4"
engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=True, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

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

models.py

from sqlalchemy import Column, Integer, String, Float, Boolean, DateTime
from sqlalchemy.sql import func
from database import Base

class BookModel(Base):
    __tablename__ = "books"
    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    title = Column(String(100), nullable=False, index=True)
    author = Column(String(50), nullable=False)
    price = Column(Float, nullable=False)
    is_borrowed = Column(Boolean, default=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

schemas.py

from pydantic import BaseModel, Field
from datetime import datetime

class BookCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=100)
    author: str = Field(..., min_length=1, max_length=50)
    price: float = Field(..., gt=0)

class BookUpdate(BaseModel):
    title: str | None = None
    author: str | None = None
    price: float | None = Field(None, gt=0)
    is_borrowed: bool | None = None

class BookResponse(BookCreate):
    id: int
    is_borrowed: bool
    created_at: datetime
    class Config:
        from_attributes = True   # 支持 ORM 对象转 dict

main.py（CRUD 示例）

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from database import get_db, engine, Base
import models, schemas

Base.metadata.create_all(bind=engine)  # 建表
app = FastAPI()

@app.post("/books", response_model=schemas.BookResponse, status_code=201)
def create(book: schemas.BookCreate, db: Session = Depends(get_db)):
    db_book = models.BookModel(**book.model_dump())
    db.add(db_book)
    db.commit()
    db.refresh(db_book)
    return db_book

@app.get("/books", response_model=list[schemas.BookResponse])
def list_books(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
    return db.query(models.BookModel).offset(skip).limit(limit).all()

@app.get("/books/{id}", response_model=schemas.BookResponse)
def get(id: int, db: Session = Depends(get_db)):
    book = db.query(models.BookModel).filter(models.BookModel.id == id).first()
    if not book:
        raise HTTPException(404, "Not found")
    return book

@app.put("/books/{id}", response_model=schemas.BookResponse)
def update(id: int, book_in: schemas.BookUpdate, db: Session = Depends(get_db)):
    book = db.query(models.BookModel).filter(models.BookModel.id == id).first()
    if not book:
        raise HTTPException(404, "Not found")
    for key, value in book_in.model_dump(exclude_unset=True).items():
        setattr(book, key, value)
    db.commit()
    db.refresh(book)
    return book

@app.delete("/books/{id}", status_code=204)
def delete(id: int, db: Session = Depends(get_db)):
    book = db.query(models.BookModel).filter(models.BookModel.id == id).first()
    if not book:
        raise HTTPException(404, "Not found")
    db.delete(book)
    db.commit()
    return None

9.2 异步数据库方案（高并发通用）

使用 aiomysql + SQLAlchemy 异步支持，配合 async def 实现非阻塞数据库操作。

安装

pip install aiomysql sqlalchemy[asyncio]

database.py（异步引擎）

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import declarative_base

# 注意 URL 前缀：mysql+aiomysql://...
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@127.0.0.1:3306/school_db?charset=utf8mb4"
async_engine = create_async_engine(ASYNC_DATABASE_URL, echo=True, pool_pre_ping=True)
AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)
Base = declarative_base()

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

异步 CRUD 示例（main.py 中）

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

@app.post("/books", response_model=schemas.BookResponse, status_code=201)
async def create(book: schemas.BookCreate, db: AsyncSession = Depends(get_db)):
    db_book = models.BookModel(**book.model_dump())
    db.add(db_book)
    await db.commit()
    await db.refresh(db_book)
    return db_book

@app.get("/books/{id}", response_model=schemas.BookResponse)
async def get(id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(models.BookModel).where(models.BookModel.id == id))
    book = result.scalar_one_or_none()
    if not book:
        raise HTTPException(404, "Not found")
    return book

注意：异步 CRUD 中所有数据库操作（commit, refresh, execute）都要加 await。

优点：不阻塞事件循环，高并发下性能更好。

10. 中间件

10.1 自定义中间件（计时）

import time
from fastapi import Request

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

10.2 CORS 跨域中间件

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "*"],  # 生产环境限定具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

11. 常用功能速查

功能 代码片段
读取环境变量 import os; DB_URL = os.getenv("DATABASE_URL")
配置文件 使用 pydantic_settings 或 python-dotenv
日志 import logging; logging.basicConfig(level=logging.INFO)
分页通用依赖 见第8.2节
文件下载 FileResponse(path, filename="xxx")
重定向 RedirectResponse(url="/docs")
返回图片流 StreamingResponse(open("img.png","rb"), media_type="image/png")

12. 常见问题与坑点

问题 解决方案
表单/文件上传 500 错误 安装 python-multipart
数据库连接报错 cryptography 安装 cryptography
路由顺序错误 静态路径写在动态路径之前
响应模型过滤失效 检查 response_model 是否指定，且字段名一致
Pydantic v2 兼容 ORM 设置 Config.from_attributes = True（原 orm_mode）
异步函数中使用同步库阻塞 使用 def 或 loop.run_in_executor 避免阻塞
8000 端口占用 netstat -ano \| findstr 8000 杀死进程或换端口

13. 异步增强

迁移步骤： 1. 替换数据库驱动为 aiomysql 2. 使用 create_async_engine 和 AsyncSession 3. 所有 ORM 操作加 await 4. 路由函数改为 async def