一、FastAPI 框架基础认知

1. 核心定位与技术栈

FastAPI 是一个现代、高性能、强类型的 Python Web 框架，核心基于两个关键库：

• Starlette：提供异步 Web 能力（路由、中间件、WebSocket 等）；
• Pydantic：基于 Python 类型提示实现数据验证和序列化。

核心优势：

• 🚀 性能接近 Node.js/Go（异步 I/O 驱动）；
• 📝 自动生成交互式 API 文档（Swagger UI/Redoc）；
• ✅ 强类型校验，减少运行时错误；
• 🧩 依赖注入系统，代码解耦更优雅。

二、环境搭建与项目结构

1. 基础环境

# 安装核心依赖
pip install fastapi uvicorn
# 安装常用工具（数据验证、环境变量、异步数据库等）
pip install pydantic-settings python-multipart aiofiles

2. 规范项目结构（生产级推荐）

fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py          # 应用入口
│   ├── api/             # 路由模块（按版本/业务划分）
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── users.py
│   │       └── items.py
│   ├── core/            # 核心配置
│   │   ├── __init__.py
│   │   ├── config.py    # 环境变量、全局配置
│   │   └── security.py  # 密码加密、JWT 工具
│   ├── models/          # ORM 模型（数据库表结构）
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/         # Pydantic 模型（请求/响应数据结构）
│   │   ├── __init__.py
│   │   └── user.py
│   ├── crud/            # 数据库操作层（CRUD 逻辑）
│   │   ├── __init__.py
│   │   └── user.py
│   └── dependencies/    # 依赖注入（权限校验、数据库会话等）
│       ├── __init__.py
│       └── auth.py
├── tests/               # 测试用例
├── .env                 # 环境变量（不提交到 Git）
├── requirements.txt
└── Dockerfile           # 容器化部署

三、核心组件深度解析

1. 路由系统（基础中的基础）

FastAPI 支持所有 HTTP 方法（GET/POST/PUT/DELETE 等），通过装饰器定义路由：

from fastapi import FastAPI, Path, Query

app = FastAPI()

# 路径参数 + 类型校验
@app.get("/users/{user_id}")
def get_user(
    user_id: int = Path(..., title="用户ID", ge=1),  # ... 表示必填，ge=1 表示大于等于1
):
    return {"user_id": user_id}

# 查询参数 + 默认值
@app.get("/items/")
def get_items(
    skip: int = Query(0, ge=0),  # 默认0，大于等于0
    limit: int = Query(10, le=100),  # 默认10，小于等于100
):
    return {"skip": skip, "limit": limit}

2. Pydantic 数据模型（核心校验层）

Pydantic 是 FastAPI 的 “灵魂”，负责请求体解析、数据验证、响应序列化：

from pydantic import BaseModel, Field, EmailStr
from typing import Optional, List

# 基础模型
class UserBase(BaseModel):
    name: str = Field(..., min_length=2, max_length=20)  # 必填，长度限制
    email: EmailStr  # 自动校验邮箱格式
    age: Optional[int] = Field(None, ge=0, le=120)  # 可选，年龄范围

# 创建用户请求模型（继承基础模型，不含id）
class UserCreate(UserBase):
    password: str = Field(..., min_length=6)  # 密码字段

# 响应模型（继承基础模型，含id，隐藏密码）
class UserResponse(UserBase):
    id: int

    class Config:
        orm_mode = True  # 支持 ORM 对象直接转 Pydantic 模型

# 嵌套模型示例
class Item(BaseModel):
    name: str
    price: float

class Order(BaseModel):
    order_id: int
    items: List[Item]  # 嵌套列表

3. 依赖注入系统（代码解耦神器）

依赖注入是 FastAPI 最强大的特性之一，用于权限校验、数据库会话管理、参数复用等：

from fastapi import Depends, HTTPException, status
from typing import Annotated

# 1. 基础依赖：模拟获取当前用户
def get_current_user(token: str):
    if token != "valid_token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的token"
        )
    return {"user_id": 1, "name": "张三"}

# 2. 依赖嵌套：依赖上面的 get_current_user
def get_current_active_user(
    current_user: Annotated[dict, Depends(get_current_user)]
):
    if current_user.get("disabled"):
        raise HTTPException(status_code=400, detail="用户已禁用")
    return current_user

# 3. 在路由中使用依赖
@app.get("/users/me")
def read_users_me(
    current_user: Annotated[dict, Depends(get_current_active_user)]
):
    return current_user

四、进阶特性实战

1. 异步编程（高性能关键）

FastAPI 原生支持异步，IO 密集型操作（数据库、网络请求）必须用异步才能发挥性能优势：

from fastapi import FastAPI
import httpx  # 异步 HTTP 客户端

app = FastAPI()

# 异步路由函数（async def）
@app.get("/async-example")
async def async_example():
    # 异步发起 HTTP 请求（不阻塞事件循环）
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.github.com")
    return {"github_status": response.status_code}

2. 数据库集成（SQLAlchemy 2.0 异步版）

生产环境常用 SQLAlchemy 2.0 作为 ORM，配合异步数据库驱动（如 asyncpg）：

# core/config.py 配置数据库连接
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str = "postgresql+asyncpg://user:password@localhost/dbname"

settings = Settings()

# models/user.py 定义 ORM 模型
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(20), index=True)
    email = Column(String(100), unique=True, index=True)
    hashed_password = Column(String(255))

# dependencies/db.py 数据库会话依赖
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings

engine = create_async_engine(settings.DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# crud/user.py 数据库操作
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.schemas.user import UserCreate
from app.core.security import get_password_hash

async def create_user(db: AsyncSession, user_in: UserCreate):
    hashed_password = get_password_hash(user_in.password)
    db_user = User(
        name=user_in.name,
        email=user_in.email,
        hashed_password=hashed_password
    )
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user

async def get_user_by_email(db: AsyncSession, email: str):
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()

3. 认证授权（OAuth2 + JWT）

FastAPI 内置 OAuth2 支持，结合 JWT 实现无状态认证：

# core/security.py 安全工具
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

# 配置（生产环境用环境变量）
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# dependencies/auth.py 认证依赖
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.security import SECRET_KEY, ALGORITHM
from app.dependencies.db import get_db
from app.crud.user import get_user_by_email

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/v1/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db)
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无法验证凭据",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        email: str = payload.get("sub")
        if email is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    user = await get_user_by_email(db, email=email)
    if user is None:
        raise credentials_exception
    return user

# api/v1/users.py 登录路由
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession
from app.dependencies.db import get_db
from app.schemas.user import UserCreate, UserResponse
from app.crud.user import create_user, get_user_by_email
from app.core.security import create_access_token, verify_password

router = APIRouter()

@router.post("/register", response_model=UserResponse)
async def register(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
    user = await get_user_by_email(db, email=user_in.email)
    if user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="邮箱已注册"
        )
    return await create_user(db, user_in)

@router.post("/login")
async def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: AsyncSession = Depends(get_db)
):
    user = await get_user_by_email(db, email=form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用户名或密码错误"
        )
    access_token = create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

五、工程化与部署

1. 测试（pytest + TestClient）

FastAPI 提供 TestClient 方便进行接口测试：

# tests/test_users.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_register():
    response = client.post(
        "/api/v1/register",
        json={
            "name": "测试用户",
            "email": "test@example.com",
            "password": "123456"
        }
    )
    assert response.status_code == 200
    assert response.json()["email"] == "test@example.com"

2. 部署（Docker + Gunicorn + Uvicorn）

生产环境推荐用 Docker 容器化部署，配合 Gunicorn 管理进程，Uvicorn 处理异步请求：

dockerfile

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["gunicorn", "app.main:app", "--workers", "4", "--worker-class", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]