FastAPI 入门教程 —— 现代 Python API 开发

本章目标

• 理解 FastAPI 的异步架构与性能优势
• 掌握 Pydantic v2 数据模型深度用法
• 构建完整 CRUD API（含 SQLAlchemy 异步数据库）
• 理解 FastAPI 与 Flask/Django 的定位差异

1. FastAPI 为何这么快？

传统 WSGI（Flask/Django）
 请求 → 线程池 → 阻塞等待（DB、网络）→ 响应
        每个请求占一个线程

FastAPI ASGI
 请求 → 事件循环 → async/await（非阻塞）→ 响应
        一个线程处理成千上万并发

性能基准（每秒请求数，粗略对比）：

• Flask（sync）：~5,000 req/s
• FastAPI（async）：~25,000 req/s
• Node.js（Express）：~20,000 req/s

2. Pydantic v2 深度用法

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional, List, Union
from datetime import date
from enum import Enum
import re


class ProductCategory(str, Enum):
    ELECTRONICS = "electronics"
    CLOTHING = "clothing"
    FOOD = "food"


class ProductCreate(BaseModel):
    """创建产品请求模型"""
    name: str = Field(..., min_length=1, max_length=200)
    price: float = Field(..., gt=0, description="价格必须大于0")
    category: ProductCategory
    tags: List[str] = Field(default=[], max_length=10)
    discount_percent: Optional[float] = Field(default=None, ge=0, le=100)
    release_date: Optional[date] = None

    # 字段验证器（v2 风格）
    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("名称不能为空或全是空格")
        return v.strip()

    @field_validator("tags")
    @classmethod
    def tags_must_be_unique(cls, v: List[str]) -> List[str]:
        if len(v) != len(set(v)):
            raise ValueError("标签不能重复")
        return [tag.strip().lower() for tag in v]

    # 模型级验证器（跨字段）
    @model_validator(mode="after")
    def validate_pricing(self):
        if self.discount_percent is not None and self.price < 10:
            raise ValueError("价格低于10元的产品不能设置折扣")
        return self


class ProductResponse(BaseModel):
    """产品响应模型"""
    id: int
    name: str
    price: float
    category: ProductCategory
    tags: List[str]
    final_price: float  # 计算字段

    model_config = {"from_attributes": True}

Pydantic v2 vs v1 关键变化

变化	v1	v2
字段验证器	@validator("field")	@field_validator("field")
根验证器	@root_validator	@model_validator(mode="after")
ORM 模式	class Config: orm_mode = True	model_config = {"from_attributes": True}
正则	Field(regex=...)	Field(pattern=...)

3. 异步 SQLAlchemy 集成

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Float, Enum as SAEnum, DateTime, func
import enum

DATABASE_URL = "sqlite+aiosqlite:///./products.db"

engine = create_async_engine(DATABASE_URL, echo=False)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


class ProductCategory(str, enum.Enum):
    ELECTRONICS = "electronics"
    CLOTHING = "clothing"
    FOOD = "food"


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(primary_key=True, index=True)
    name: Mapped[str] = mapped_column(String(200))
    price: Mapped[float] = mapped_column(Float)
    category: Mapped[ProductCategory] = mapped_column(SAEnum(ProductCategory))
    tags: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )


# 依赖项 —— 获取数据库会话
async def get_db() -> AsyncSession:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# main.py - CRUD 路由
@app.post("/products", response_model=ProductResponse, status_code=201)
async def create_product(
    product: ProductCreate,
    db: AsyncSession = Depends(get_db),
):
    db_product = Product(
        name=product.name,
        price=product.price,
        category=product.category,
        tags=",".join(product.tags),
    )
    db.add(db_product)
    await db.flush()
    await db.refresh(db_product)
    return db_product


@app.get("/products", response_model=List[ProductResponse])
async def list_products(
    skip: int = 0,
    limit: int = 20,
    category: Optional[ProductCategory] = None,
    db: AsyncSession = Depends(get_db),
):
    from sqlalchemy import select
    stmt = select(Product).offset(skip).limit(limit)
    if category:
        stmt = stmt.where(Product.category == category)
    result = await db.execute(stmt)
    return result.scalars().all()

4. FastAPI 与 Flask/Django 对比

维度	FastAPI	Flask	Django
类型驱动	✅ 原生	❌	❌（需 DRF）
异步原生	✅ ASGI	⚠️ 2.0+ 部分支持	⚠️ 3.1+ 部分支持
自动文档	✅ OpenAPI 内置	❌ 需插件	❌ 需 DRF + drf-spectacular
数据验证	✅ Pydantic	❌ 需手动	✅ DRF Serializer
性能	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐
生态成熟度	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐
学习曲线	中	低	高
适合场景	API/微服务/ML	小型应用/原型	全栈大型应用

5. 项目结构最佳实践

fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 实例创建 + 路由注册
│   ├── core/
│   │   ├── config.py        # 配置（Pydantic Settings）
│   │   └── security.py      # 认证/密码工具
│   ├── models/              # SQLAlchemy 模型
│   ├── schemas/             # Pydantic 请求/响应模型
│   ├── api/
│   │   ├── __init__.py
│   │   ├── deps.py          # 共享依赖项
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── router.py    # 聚合子路由
│   │       └── endpoints/
│   │           ├── users.py
│   │           └── products.py
│   └── services/            # 业务逻辑层
├── alembic/                 # 数据库迁移
├── tests/
├── requirements.txt
└── .env

思考题

1. FastAPI 的异步是否意味着所有代码都必须是 async def？什么场景同步函数也能获益？
2. Pydantic v2 的 model_validator 和 field_validator 分别何时使用？
3. FastAPI 的依赖注入与类的构造函数注入有何异同？为什么它更适合 Web 场景？
4. 将 Flask 项目迁移到 FastAPI 时，哪些部分最需要重写？