# fastapi

**Repository Path**: wangjllong/fastapi

## Basic Information

- **Project Name**: fastapi
- **Description**: python-fastapi项目
- **Primary Language**: Python
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2025-11-13
- **Last Updated**: 2026-07-09

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# FastAPI 从入门到精通教程

## 目录
1. [FastAPI 简介](#1-fastapi-简介)
2. [环境搭建（Python 3.12）](#2-环境搭建python-312)
3. [核心知识点详解（含示例）](#3-核心知识点详解含示例)
   - 3.1 第一个 FastAPI 应用
   - 3.2 路径参数与查询参数
   - 3.3 请求体（Pydantic 模型）
   - 3.4 响应处理（响应模型、状态码）
   - 3.5 依赖注入（核心特性）
   - 3.6 表单数据与文件上传
   - 3.7 安全性（OAuth2、JWT 认证）
   - 3.8 CORS 跨域配置
   - 3.9 中间件
   - 3.10 背景任务
   - 3.11 异步支持
   - 3.12 测试（pytest）
4. [数据库集成（MySQL + 连接池）](#4-数据库集成mysql--连接池)
5. [完整项目实战（用户管理系统）](#5-完整项目实战用户管理系统)
   - 5.1 项目架构设计
   - 5.2 核心配置
   - 5.3 代码实现（分层架构）
   - 5.4 项目运行与测试
6. [生产环境部署建议](#6-生产环境部署建议)

## 1. FastAPI 简介
FastAPI 是一个 **现代、快速（高性能）、简洁** 的 Python Web 框架，基于 Starlette（异步 Web 框架）和 Pydantic（数据验证）构建，专为 API 开发设计，核心优势：
- 极致性能：可与 Node.js、Go 媲美（基于 Starlette 和 Uvicorn）
- 自动生成 API 文档：支持 Swagger UI（`/docs`）和 ReDoc（`/redoc`）
- 强类型提示：依赖 Python 类型注解，代码更易维护
- 数据验证：基于 Pydantic 2.x，自动验证请求数据
- 异步原生支持：完美兼容 Python 3.7+ 异步语法
- 依赖注入：灵活的依赖管理系统
- 安全性：内置 OAuth2、JWT、API 密钥等认证方案
- 扩展性强：支持中间件、背景任务、WebSocket 等

## 2. 环境搭建（Python 3.12）
### 2.1 安装 Python 3.12
- 从 [Python 官网](https://www.python.org/downloads/release/python-3120/) 下载并安装
- 验证安装：`python --version` 或 `python3 --version`（需显示 3.12.x）

### 2.2 创建虚拟环境（推荐）
```bash
# 创建虚拟环境
python -m venv fastapi-env

# 激活虚拟环境（Windows）
fastapi-env\Scripts\activate

# 激活虚拟环境（Mac/Linux）
source fastapi-env/bin/activate
```

### 2.3 安装核心依赖
```bash
# 核心依赖（FastAPI + 异步服务器 Uvicorn）
pip install fastapi uvicorn==0.27.0

# 数据库依赖（SQLAlchemy 2.x + MySQL 异步驱动 + 连接池）
pip install sqlalchemy[asyncio]==2.0.23 asyncmy==0.2.9

# 安全依赖（JWT 认证 + 密码加密）
pip install python-jose[cryptography]==3.3.0 passlib[bcrypt]==1.7.4

# 其他依赖（配置管理 + 表单/文件上传 + 测试）
pip install pydantic-settings==2.1.0 python-multipart==0.0.6 pytest==7.4.3
```

## 3. 核心知识点详解（含示例）
### 3.1 第一个 FastAPI 应用
创建 `main.py`：
```python
from fastapi import FastAPI

# 创建 FastAPI 实例
app = FastAPI(
    title="FastAPI 入门教程",
    description="Python 3.12 + FastAPI 从入门到精通",
    version="1.0.0"
)

# 路径操作装饰器（GET 方法 + 根路径）
@app.get("/")
def read_root():
    """根路径接口"""
    return {"message": "Hello FastAPI!", "python_version": "3.12"}

# 运行命令：uvicorn main:app --reload
# 访问：http://127.0.0.1:8000
# 文档：http://127.0.0.1:8000/docs 或 http://127.0.0.1:8000/redoc
```

运行应用：
```bash
uvicorn main:app --reload  # --reload 开发模式热重载
```

### 3.2 路径参数与查询参数
#### 3.2.1 路径参数（URL 路径中的参数）
```python
from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# 路径参数（类型注解自动验证）
@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
    """
    - item_id: 路径参数（int 类型，自动验证，非 int 会返回 422 错误）
    - q: 查询参数（可选，默认 None）
    """
    return {"item_id": item_id, "q": q}

# 路径参数枚举（限制可选值）
from enum import Enum

class ItemType(str, Enum):
    book = "book"
    electronic = "electronic"

@app.get("/items/{item_id}/{item_type}")
def read_item_with_type(item_id: int, item_type: ItemType):
    return {"item_id": item_id, "item_type": item_type.value}
```

#### 3.2.2 查询参数（URL 中 `?` 后的参数）
```python
# 必需查询参数 + 可选查询参数 + 默认值
@app.get("/users/")
def get_users(
    page: int = 1,  # 必需查询参数（无默认值）
    limit: int = 10,  # 可选查询参数（默认 10）
    sort: Optional[str] = "create_time"  # 可选查询参数（默认 "create_time"）
):
    return {"page": page, "limit": limit, "sort": sort}

# 访问示例：http://127.0.0.1:8000/users/?page=2&limit=20&sort=name
```

### 3.3 请求体（Pydantic 模型）
FastAPI 使用 Pydantic 模型定义请求体，自动验证数据类型并生成文档。

```python
from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# 定义 Pydantic 模型（请求体结构）
class Item(BaseModel):
    name: str = Field(..., description="物品名称（必填）", min_length=1, max_length=50)
    price: float = Field(..., description="物品价格（必填）", gt=0)
    description: Optional[str] = Field(None, description="物品描述（可选）", max_length=200)
    tags: Optional[list[str]] = Field(default=[], description="标签列表（可选）")

# POST 方法接收请求体
@app.post("/items/", response_model=Item, summary="创建物品")
def create_item(item: Item):
    """创建新物品，请求体为 Item 模型"""
    # 实际场景：将数据存入数据库
    return item  # 自动序列化 Item 为 JSON
```

### 3.4 响应处理
#### 3.4.1 响应模型（`response_model`）
```python
# 定义响应模型（可隐藏敏感字段）
class UserIn(BaseModel):
    username: str
    password: str  # 敏感字段，响应中隐藏

class UserOut(BaseModel):
    id: int
    username: str
    email: Optional[str] = None

@app.post("/users/", response_model=UserOut, summary="创建用户")
def create_user(user: UserIn):
    # 模拟数据库插入，返回带 id 的用户信息（隐藏 password）
    return {"id": 1, "username": user.username, "email": f"{user.username}@example.com"}
```

#### 3.4.2 状态码与响应头
```python
from fastapi import status
from fastapi.responses import JSONResponse

@app.post("/items/{item_id}/update", summary="更新物品")
def update_item(item_id: int, item: Item):
    # 模拟更新逻辑
    updated_item = item.model_dump()
    updated_item["item_id"] = item_id
    # 返回 201 Created 状态码 + 自定义响应头
    return JSONResponse(
        content=updated_item,
        status_code=status.HTTP_201_CREATED,
        headers={"X-Item-Updated": "true"}
    )
```

### 3.5 依赖注入（核心特性）
依赖注入是 FastAPI 的核心，用于复用逻辑（如认证、数据库连接、参数验证）。

#### 3.5.1 基础依赖函数
```python
from fastapi import FastAPI, Depends

app = FastAPI()

# 定义依赖函数（可复用逻辑）
def get_current_user(token: str):
    """模拟从 token 中获取当前用户"""
    return {"username": "admin", "token": token}

# 路径操作依赖
@app.get("/profile/", summary="获取用户资料")
def get_profile(current_user: dict = Depends(get_current_user)):
    """current_user 依赖 get_current_user 函数"""
    return {"profile": current_user}

# 访问示例：http://127.0.0.1:8000/profile/?token=abc123
```

#### 3.5.2 类依赖与全局依赖
```python
# 类依赖（更灵活的复用）
class Pagination:
    def __init__(self, page: int = 1, limit: int = 10):
        self.page = page
        self.limit = limit

@app.get("/articles/", summary="获取文章列表")
def get_articles(pagination: Pagination = Depends()):
    """依赖 Pagination 类，自动解析 page 和 limit 参数"""
    skip = (pagination.page - 1) * pagination.limit
    return {"skip": skip, "limit": pagination.limit, "total": 100}

# 全局依赖（所有接口都生效）
async def global_dependency():
    """模拟全局日志记录"""
    print("全局依赖：接口被访问")

app = FastAPI(dependencies=[Depends(global_dependency)])
```

### 3.6 表单数据与文件上传
#### 3.6.1 表单数据（需安装 `python-multipart`）
```python
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login/", summary="用户登录（表单）")
def login(username: str = Form(...), password: str = Form(...)):
    """接收表单数据（而非 JSON）"""
    return {"username": username, "message": "登录成功"}
```

#### 3.6.2 文件上传
```python
from fastapi import FastAPI, File, UploadFile
from typing import List

app = FastAPI()

# 单文件上传
@app.post("/upload/file/", summary="单文件上传")
async def upload_file(file: UploadFile = File(...)):
    """UploadFile 支持异步读取，包含文件名、大小、内容类型等信息"""
    content = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "file_size": len(content)
    }

# 多文件上传
@app.post("/upload/files/", summary="多文件上传")
async def upload_files(files: List[UploadFile] = File(...)):
    return [
        {"filename": file.filename, "content_type": file.content_type}
        for file in files
    ]
```

### 3.7 安全性（OAuth2 + JWT 认证）
#### 3.7.1 JWT 工具函数
```python
from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext

# 密码加密上下文（bcrypt 算法）
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# JWT 配置（实际项目中应放在环境变量）
SECRET_KEY = "your-secret-key-keep-it-safe"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 密码加密
def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

# 密码验证
def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

# 生成 JWT Token
def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt
```

#### 3.7.2 OAuth2 认证依赖
```python
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

# OAuth2 令牌获取地址
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# 模拟数据库用户
fake_users_db = {
    "admin": {
        "username": "admin",
        "hashed_password": get_password_hash("admin123"),
        "email": "admin@example.com"
    }
}

# 获取当前登录用户（依赖）
async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无效的令牌",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # 解码 Token
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    # 从数据库查询用户（模拟）
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return user

# Token 接口（OAuth2 密码模式）
@app.post("/token", summary="获取 JWT Token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    # 验证用户
    user = fake_users_db.get(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="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # 生成 Token
    access_token = create_access_token(data={"sub": user["username"]})
    return {"access_token": access_token, "token_type": "bearer"}

# 受保护的接口
@app.get("/protected/", summary="受 JWT 保护的接口")
async def protected_route(current_user: dict = Depends(get_current_user)):
    return {"message": "这是受保护的接口", "user": current_user}
```

### 3.8 CORS 跨域配置
前端与后端分离时需配置 CORS：
```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 配置允许跨域的源、方法、请求头
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境需指定具体域名（如 ["https://your-frontend.com"]）
    allow_credentials=True,
    allow_methods=["*"],  # 允许所有 HTTP 方法
    allow_headers=["*"],  # 允许所有请求头
)
```

### 3.9 中间件
中间件用于处理请求/响应的全局逻辑（如日志、计时）：
```python
from fastapi import FastAPI
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
import time

app = FastAPI()

# 自定义中间件（记录请求耗时）
class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()
        # 处理请求
        response = await call_next(request)
        # 计算耗时
        process_time = time.time() - start_time
        # 添加响应头
        response.headers["X-Process-Time"] = str(process_time)
        print(f"请求 {request.url.path} 耗时: {process_time:.4f}s")
        return response

# 注册中间件
app.add_middleware(TimingMiddleware)

@app.get("/")
def read_root():
    return {"message": "Hello Middleware!"}
```

### 3.10 背景任务
用于处理请求响应后需要异步执行的任务（如发送邮件、日志写入）：
```python
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

# 背景任务函数
def send_email(to: str, subject: str):
    """模拟发送邮件（实际场景需用 aiosmtplib 异步发送）"""
    print(f"发送邮件到 {to}，主题：{subject}")

@app.post("/users/register/", summary="用户注册（发送欢迎邮件）")
def register_user(email: str, background_tasks: BackgroundTasks):
    # 注册逻辑（模拟）
    user_id = 1001
    # 添加背景任务（请求响应后执行）
    background_tasks.add_task(send_email, to=email, subject="欢迎注册！")
    return {"user_id": user_id, "message": "注册成功，欢迎邮件已发送"}
```

### 3.11 异步支持
FastAPI 原生支持异步路由和异步依赖，配合异步数据库驱动可提升性能：
```python
from fastapi import FastAPI
import asyncio

app = FastAPI()

# 异步路由（async def）
@app.get("/async/")
async def async_route():
    """异步路由，支持 await 异步操作"""
    await asyncio.sleep(1)  # 模拟异步 IO 操作（如数据库查询、API 调用）
    return {"message": "异步路由响应", "async": True}

# 异步依赖
async def async_dependency():
    await asyncio.sleep(0.5)
    return {"data": "异步依赖数据"}

@app.get("/async/depends/")
async def async_with_depends(data: dict = Depends(async_dependency)):
    return {"dependency_data": data, "message": "异步依赖示例"}
```

### 3.12 测试（pytest）
创建 `test_main.py`：
```python
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_root():
    """测试根路径"""
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello FastAPI!", "python_version": "3.12"}

def test_read_item():
    """测试路径参数和查询参数"""
    response = client.get("/items/10?q=test")
    assert response.status_code == 200
    assert response.json() == {"item_id": 10, "q": "test"}

def test_create_item():
    """测试请求体"""
    response = client.post(
        "/items/",
        json={"name": "iPhone", "price": 999.99, "description": "手机"}
    )
    assert response.status_code == 200
    assert response.json()["name"] == "iPhone"
```

运行测试：
```bash
pytest test_main.py -v
```

## 4. 数据库集成（MySQL + 连接池）
使用 **SQLAlchemy 2.x（异步）+ asyncmy + 连接池** 操作 MySQL，连接池可复用数据库连接，提升性能。

### 4.1 数据库配置
创建 `app/config.py`：
```python
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Optional

class Settings(BaseSettings):
    """应用配置（从环境变量或 .env 文件读取）"""
    # 数据库配置
    DB_HOST: str = "localhost"
    DB_PORT: int = 3306
    DB_USER: str = "root"
    DB_PASSWORD: str = "123456"
    DB_NAME: str = "fastapi_db"
    # 数据库连接池配置
    DB_POOL_SIZE: int = 10  # 连接池默认大小
    DB_MAX_OVERFLOW: int = 20  # 连接池最大溢出连接数
    DB_POOL_RECYCLE: int = 3600  # 连接自动回收时间（秒）
    DB_POOL_PRE_PING: bool = True  # 检查连接可用性（避免失效连接）

    # JWT 配置
    SECRET_KEY: str = "your-secret-key-keep-it-safe-2025"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30

    # 模型配置（读取 .env 文件）
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")

# 实例化配置
settings = Settings()

# 构建数据库 URL（异步 MySQL 格式）
DATABASE_URL = (
    f"mysql+asyncmy://{settings.DB_USER}:{settings.DB_PASSWORD}@"
    f"{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}"
)
```

### 4.2 数据库会话（连接池）
创建 `app/db/session.py`：
```python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.config import settings, DATABASE_URL

# 创建异步引擎（配置连接池）
engine = create_async_engine(
    DATABASE_URL,
    pool_size=settings.DB_POOL_SIZE,
    max_overflow=settings.DB_MAX_OVERFLOW,
    pool_recycle=settings.DB_POOL_RECYCLE,
    pool_pre_ping=settings.DB_POOL_PRE_PING,
    echo=False,  # 生产环境关闭 SQL 日志
)

# 创建异步会话工厂
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    autoflush=False,  # 关闭自动刷新
    autocommit=False,  # 关闭自动提交
    expire_on_commit=False,  # 提交后不失效对象
)

# 依赖函数：获取数据库会话（自动关闭）
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()
```

### 4.3 SQLAlchemy 模型（数据表）
创建 `app/db/base.py`：
```python
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, DateTime, func

# 基础模型（所有数据表的父类）
Base = declarative_base()

class BaseModel(Base):
    """基础模型，包含公共字段"""
    __abstract__ = True  # 抽象类，不创建数据表

    id = Column(Integer, primary_key=True, index=True, comment="主键 ID")
    created_at = Column(
        DateTime, 
        default=func.now(), 
        nullable=False, 
        comment="创建时间"
    )
    updated_at = Column(
        DateTime, 
        default=func.now(), 
        onupdate=func.now(), 
        nullable=False, 
        comment="更新时间"
    )
```

创建 `app/models/user.py`（用户表）：
```python
from sqlalchemy import Column, String, Boolean
from app.db.base import BaseModel

class User(BaseModel):
    """用户表"""
    __tablename__ = "users"

    username = Column(String(50), unique=True, index=True, nullable=False, comment="用户名")
    email = Column(String(100), unique=True, index=True, nullable=False, comment="邮箱")
    hashed_password = Column(String(100), nullable=False, comment="加密密码")
    is_active = Column(Boolean, default=True, comment="是否激活")
    is_admin = Column(Boolean, default=False, comment="是否为管理员")
```

### 4.4 数据库迁移（可选）
使用 `alembic` 管理数据库迁移：
```bash
# 安装 alembic
pip install alembic==1.12.1

# 初始化 alembic
alembic init alembic

# 修改 alembic.ini 中的数据库 URL（同步格式，用于迁移）
sqlalchemy.url = mysql+pymysql://root:123456@localhost:3306/fastapi_db

# 修改 alembic/env.py，导入 Base 模型
from app.db.base import Base
target_metadata = Base.metadata

# 创建迁移脚本
alembic revision --autogenerate -m "init users table"

# 执行迁移（创建数据表）
alembic upgrade head
```

## 5. 完整项目实战（用户管理系统）
### 5.1 项目架构设计（分层架构）
```
fastapi-project/
├── app/                      # 应用核心目录
│   ├── __init__.py
│   ├── main.py               # 应用入口（注册路由、中间件）
│   ├── config.py             # 配置文件
│   ├── api/                  # 路由层（API 接口定义）
│   │   ├── __init__.py
│   │   ├── dependencies.py   # 路由依赖（认证、权限）
│   │   └── v1/               # API 版本 1
│   │       ├── __init__.py
│   │       └── endpoints/    # 具体接口
│   │           ├── auth.py   # 认证接口（登录、刷新 Token）
│   │           └── users.py  # 用户接口（CRUD）
│   ├── core/                 # 核心组件
│   │   ├── __init__.py
│   │   └── security.py       # 安全工具（密码、JWT）
│   ├── crud/                 # 数据访问层（DAO）
│   │   ├── __init__.py
│   │   ├── base.py           # 基础 CRUD
│   │   └── user.py           # 用户 CRUD
│   ├── db/                   # 数据库相关
│   │   ├── __init__.py
│   │   ├── base.py           # 基础模型
│   │   └── session.py        # 数据库会话（连接池）
│   ├── models/               # 数据模型（SQLAlchemy）
│   │   ├── __init__.py
│   │   └── user.py           # 用户模型
│   ├── schemas/              # Pydantic 模型（请求/响应）
│   │   ├── __init__.py
│   │   ├── auth.py           # 认证相关模型
│   │   └── user.py           # 用户相关模型
│   └── utils/                # 工具函数
│       ├── __init__.py
│       └── exceptions.py     # 自定义异常
├── tests/                    # 测试用例
│   ├── __init__.py
│   ├── conftest.py           # 测试配置
│   └── test_api/             # API 测试
│       ├── test_auth.py
│       └── test_users.py
├── .env                      # 环境变量（不提交到 Git）
├── requirements.txt          # 依赖清单
└── README.md                 # 项目说明
```

### 5.2 核心代码实现
#### 5.2.1 核心安全工具（`app/core/security.py`）
```python
from datetime import datetime, timedelta
from typing import Optional, Union

from jose import JWTError, jwt
from passlib.context import CryptContext

from app.config import settings

# 密码加密上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """验证密码"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """生成密码哈希"""
    return pwd_context.hash(password)

def create_access_token(
    subject: Union[str, int], expires_delta: Optional[timedelta] = None
) -> str:
    """生成 JWT Access Token"""
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    
    to_encode = {"exp": expire, "sub": str(subject)}
    encoded_jwt = jwt.encode(
        to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM
    )
    return encoded_jwt
```

#### 5.2.2 Pydantic 模型（`app/schemas/`）
`app/schemas/user.py`：
```python
from typing import Optional
from pydantic import BaseModel, EmailStr, Field, validator

# 基础用户模型（共享字段）
class UserBase(BaseModel):
    username: str = Field(..., min_length=3, max_length=50, description="用户名")
    email: EmailStr = Field(..., description="邮箱")
    is_active: Optional[bool] = Field(True, description="是否激活")

# 创建用户请求模型
class UserCreate(UserBase):
    password: str = Field(..., min_length=6, max_length=20, description="密码")

    @validator("password")
    def password_not_all_numbers(cls, v):
        """自定义验证：密码不能全为数字"""
        if v.isdigit():
            raise ValueError("密码不能全为数字")
        return v

# 更新用户请求模型
class UserUpdate(BaseModel):
    username: Optional[str] = Field(None, min_length=3, max_length=50, description="用户名")
    email: Optional[EmailStr] = Field(None, description="邮箱")
    password: Optional[str] = Field(None, min_length=6, max_length=20, description="密码")

# 用户响应模型
class UserOut(UserBase):
    id: int = Field(..., description="用户 ID")
    is_admin: bool = Field(False, description="是否为管理员")
    created_at: str = Field(..., description="创建时间")
    updated_at: str = Field(..., description="更新时间")

    class Config:
        orm_mode = True  # 支持 SQLAlchemy 模型直接转换
```

`app/schemas/auth.py`：
```python
from pydantic import BaseModel

# Token 响应模型
class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

# Token  payload 模型
class TokenPayload(BaseModel):
    sub: int  # 用户 ID
```

#### 5.2.3 CRUD 层（`app/crud/`）
`app/crud/base.py`：
```python
from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.base import BaseModel

ModelType = TypeVar("ModelType", bound=BaseModel)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
    """基础 CRUD 类"""
    def __init__(self, model: Type[ModelType]):
        self.model = model

    async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]:
        """根据 ID 获取单条数据"""
        result = await db.execute(select(self.model).filter(self.model.id == id))
        return result.scalars().first()

    async def get_multi(
        self, db: AsyncSession, *, skip: int = 0, limit: int = 100
    ) -> List[ModelType]:
        """获取多条数据（分页）"""
        result = await db.execute(select(self.model).offset(skip).limit(limit))
        return result.scalars().all()

    async def create(self, db: AsyncSession, *, obj_in: CreateSchemaType) -> ModelType:
        """创建数据"""
        db_obj = self.model(**obj_in.model_dump())
        db.add(db_obj)
        await db.commit()
        await db.refresh(db_obj)  # 刷新对象，获取数据库生成的字段（如 id、created_at）
        return db_obj

    async def update(
        self,
        db: AsyncSession,
        *,
        db_obj: ModelType,
        obj_in: Union[UpdateSchemaType, Dict[str, Any]]
    ) -> ModelType:
        """更新数据"""
        if isinstance(obj_in, dict):
            update_data = obj_in
        else:
            update_data = obj_in.model_dump(exclude_unset=True)  # 只更新提供的字段
        
        # 逐个更新字段
        for field in update_data:
            if field in update_data:
                setattr(db_obj, field, update_data[field])
        
        db.add(db_obj)
        await db.commit()
        await db.refresh(db_obj)
        return db_obj

    async def remove(self, db: AsyncSession, *, id: int) -> ModelType:
        """删除数据"""
        obj = await db.get(self.model, id)
        await db.delete(obj)
        await db.commit()
        return obj
```

`app/crud/user.py`：
```python
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional

from app.crud.base import CRUDBase
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
from app.core.security import get_password_hash

class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
    """用户 CRUD 类"""

    async def get_by_username(self, db: AsyncSession, *, username: str) -> Optional[User]:
        """根据用户名获取用户"""
        result = await db.execute(select(User).filter(User.username == username))
        return result.scalars().first()

    async def get_by_email(self, db: AsyncSession, *, email: str) -> Optional[User]:
        """根据邮箱获取用户"""
        result = await db.execute(select(User).filter(User.email == email))
        return result.scalars().first()

    async def create(self, db: AsyncSession, *, obj_in: UserCreate) -> User:
        """创建用户（加密密码）"""
        db_obj = User(
            username=obj_in.username,
            email=obj_in.email,
            hashed_password=get_password_hash(obj_in.password),
            is_active=obj_in.is_active,
        )
        db.add(db_obj)
        await db.commit()
        await db.refresh(db_obj)
        return db_obj

    async def update(
        self,
        db: AsyncSession,
        *,
        db_obj: User,
        obj_in: Union[UserUpdate, dict]
    ) -> User:
        """更新用户（如果提供密码则加密）"""
        if isinstance(obj_in, dict):
            update_data = obj_in
        else:
            update_data = obj_in.model_dump(exclude_unset=True)
        
        # 如果更新密码，需要加密
        if "password" in update_data and update_data["password"]:
            hashed_password = get_password_hash(update_data["password"])
            del update_data["password"]
            update_data["hashed_password"] = hashed_password
        
        return await super().update(db, db_obj=db_obj, obj_in=update_data)

# 实例化用户 CRUD
user = CRUDUser(User)
```

#### 5.2.4 路由依赖（`app/api/dependencies.py`）
```python
from typing import Annotated, Optional

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.db.session import get_db
from app.crud.user import user as user_crud
from app.models.user import User
from app.schemas.auth import TokenPayload

# OAuth2 令牌获取地址
reusable_oauth2 = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")

async def get_current_user(
    db: Annotated[AsyncSession, Depends(get_db)],
    token: Annotated[str, Depends(reusable_oauth2)],
) -> User:
    """获取当前登录用户（依赖）"""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="无效的认证令牌",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        # 解码 Token
        payload = jwt.decode(
            token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
        )
        user_id: int = int(payload.get("sub"))
        if user_id is None:
            raise credentials_exception
        token_data = TokenPayload(sub=user_id)
    except JWTError:
        raise credentials_exception
    
    # 查询用户
    user = await user_crud.get(db, id=token_data.sub)
    if user is None:
        raise credentials_exception
    if not user.is_active:
        raise HTTPException(status_code=400, detail="用户已被禁用")
    return user

async def get_current_admin(
    current_user: Annotated[User, Depends(get_current_user)],
) -> User:
    """获取当前管理员用户（依赖）"""
    if not current_user.is_admin:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="无管理员权限"
        )
    return current_user
```

#### 5.2.5 API 接口（`app/api/v1/endpoints/`）
`app/api/v1/endpoints/auth.py`：
```python
from typing import Annotated

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import get_db
from app.crud.user import user as user_crud
from app.core.security import verify_password, create_access_token
from app.schemas.auth import Token

router = APIRouter(prefix="/auth", tags=["认证"])

@router.post("/token", response_model=Token, summary="获取 JWT Token")
async def login_access_token(
    db: Annotated[AsyncSession, Depends(get_db)],
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
):
    """
    OAuth2 密码模式登录，获取访问令牌：
    - 用户名/密码验证
    - 返回 JWT Token（有效期 30 分钟）
    """
    # 查询用户
    user = await user_crud.get_by_username(db, username=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="用户名或密码错误",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    # 生成 Token
    access_token = create_access_token(subject=user.id)
    return {"access_token": access_token, "token_type": "bearer"}
```

`app/api/v1/endpoints/users.py`：
```python
from typing import Annotated, List

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import get_db
from app.api.dependencies import get_current_user, get_current_admin
from app.crud.user import user as user_crud
from app.models.user import User
from app.schemas.user import UserCreate, UserOut, UserUpdate

router = APIRouter(prefix="/users", tags=["用户管理"])

@router.post("/", response_model=UserOut, summary="创建用户（管理员）", dependencies=[Depends(get_current_admin)])
async def create_user(
    *,
    db: Annotated[AsyncSession, Depends(get_db)],
    user_in: UserCreate,
):
    """创建新用户（仅管理员可操作）"""
    # 检查用户名/邮箱是否已存在
    user = await user_crud.get_by_username(db, username=user_in.username)
    if user:
        raise HTTPException(status_code=400, detail="用户名已存在")
    user = await user_crud.get_by_email(db, email=user_in.email)
    if user:
        raise HTTPException(status_code=400, detail="邮箱已存在")
    # 创建用户
    return await user_crud.create(db, obj_in=user_in)

@router.get("/", response_model=List[UserOut], summary="获取用户列表（管理员）", dependencies=[Depends(get_current_admin)])
async def read_users(
    db: Annotated[AsyncSession, Depends(get_db)],
    skip: int = Query(0, ge=0, description="跳过数量"),
    limit: int = Query(10, ge=1, le=100, description="每页数量"),
):
    """获取用户列表（分页，仅管理员可操作）"""
    return await user_crud.get_multi(db, skip=skip, limit=limit)

@router.get("/me", response_model=UserOut, summary="获取当前用户信息")
async def read_user_me(
    current_user: Annotated[User, Depends(get_current_user)],
):
    """获取当前登录用户的个人信息"""
    return current_user

@router.get("/{user_id}", response_model=UserOut, summary="获取单个用户（管理员）", dependencies=[Depends(get_current_admin)])
async def read_user(
    *,
    db: Annotated[AsyncSession, Depends(get_db)],
    user_id: int,
):
    """根据 ID 获取单个用户（仅管理员可操作）"""
    user = await user_crud.get(db, id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return user

@router.put("/{user_id}", response_model=UserOut, summary="更新用户（管理员）", dependencies=[Depends(get_current_admin)])
async def update_user(
    *,
    db: Annotated[AsyncSession, Depends(get_db)],
    user_id: int,
    user_in: UserUpdate,
):
    """更新用户信息（仅管理员可操作）"""
    user = await user_crud.get(db, id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    # 检查更新的用户名/邮箱是否已被占用
    if user_in.username:
        existing_user = await user_crud.get_by_username(db, username=user_in.username)
        if existing_user and existing_user.id != user_id:
            raise HTTPException(status_code=400, detail="用户名已存在")
    if user_in.email:
        existing_user = await user_crud.get_by_email(db, email=user_in.email)
        if existing_user and existing_user.id != user_id:
            raise HTTPException(status_code=400, detail="邮箱已存在")
    # 更新用户
    return await user_crud.update(db, db_obj=user, obj_in=user_in)

@router.delete("/{user_id}", response_model=UserOut, summary="删除用户（管理员）", dependencies=[Depends(get_current_admin)])
async def delete_user(
    *,
    db: Annotated[AsyncSession, Depends(get_db)],
    user_id: int,
):
    """删除用户（仅管理员可操作）"""
    user = await user_crud.get(db, id=user_id)
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    return await user_crud.remove(db, id=user_id)
```

#### 5.2.6 应用入口（`app/main.py`）
```python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.api.v1.endpoints import auth, users
from app.api.v1 import api_router
from app.config import settings

# 创建 FastAPI 实例
app = FastAPI(
    title="FastAPI 用户管理系统",
    description="Python 3.12 + FastAPI + MySQL + 连接池 完整项目",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

# 配置 CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境替换为具体前端域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册 API 路由（版本控制）
app.include_router(api_router, prefix="/api/v1")

# 根路径
@app.get("/", summary="健康检查")
async def root():
    return {
        "message": "FastAPI 用户管理系统运行中",
        "version": "1.0.0",
        "docs": "/docs",
        "redoc": "/redoc"
    }
```

`app/api/v1/__init__.py`：
```python
from fastapi import APIRouter

from app.api.v1.endpoints import auth, users

# 创建版本 1 的路由聚合器
api_router = APIRouter()
api_router.include_router(auth.router)
api_router.include_router(users.router)
```

### 5.3 项目运行与测试
#### 5.3.1 环境变量配置（`.env`）
```env
# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=123456
DB_NAME=fastapi_db

# JWT 配置
SECRET_KEY=your-secret-key-2025-keep-it-safe-please
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
```

#### 5.3.2 初始化数据库
1. 手动创建 MySQL 数据库：`CREATE DATABASE fastapi_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;`
2. 使用 alembic 迁移（参考 4.4 节）

#### 5.3.3 运行项目
```bash
# 开发模式（热重载）
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 生产模式（无热重载）
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
```

#### 5.3.4 测试接口
访问 API 文档：`http://127.0.0.1:8000/docs`，可直接在文档中测试所有接口：
1. 先通过 `/api/v1/auth/token` 接口登录（默认无用户，可手动插入测试用户）
2. 用返回的 Token 测试其他受保护接口
3. 管理员接口需手动设置用户 `is_admin=True`

## 6. 生产环境部署建议
1. **使用 Gunicorn + Uvicorn**：Gunicorn 作为进程管理器，Uvicorn 作为工作进程
   ```bash
   pip install gunicorn
   gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000
   ```

2. **配置 HTTPS**：使用 Nginx 反向代理 + Let's Encrypt 证书

3. **环境变量安全**：生产环境中使用环境变量或密钥管理服务存储 `SECRET_KEY`、数据库密码等敏感信息

4. **日志配置**：添加日志记录（如 `logging` 模块），记录请求、错误等信息

5. **数据库优化**：
   - 调整连接池参数（根据服务器性能）
   - 添加数据库索引
   - 定期备份数据库

6. **Docker 部署**：
   创建 `Dockerfile` 和 `docker-compose.yml`，实现容器化部署

7. **监控**：使用 Prometheus + Grafana 监控应用性能，或使用 Sentry 监控错误

## 总结
本教程覆盖了 FastAPI 从入门到进阶的所有核心知识点，包括路由、参数、请求体、响应、依赖注入、认证、数据库集成等，并提供了一个架构完整、代码规范的用户管理系统项目。项目使用 MySQL 数据库和连接池，支持用户 CRUD、JWT 认证、权限控制等功能，可直接用于生产环境或二次开发。

FastAPI 凭借其高性能、简洁的语法和丰富的特性，已成为 Python API 开发的首选框架，尤其适合构建异步、高性能的后端服务。