# Python FastAPI 框架入门

**学习目标**：掌握 FastAPI 的核心特性，理解类型驱动开发的威力，能够构建高性能异步 API 并生成交互式文档。

---

## 1. FastAPI 简介

FastAPI 是一个现代、高性能的 Python Web 框架，基于 Starlette（ASGI）和 Pydantic（数据验证），专为构建 API 而生。

### 1.1 核心特性

- **极速性能**：与 Node.js 和 Go 相当，是 Python 最快框架之一
- **类型驱动**：基于 Python 类型提示，自动数据验证
- **自动文档**：自动生成 OpenAPI 和 Swagger UI 文档
- **异步原生**：原生支持 `async`/`await`
- **依赖注入**：强大的依赖注入系统
- **数据验证**：基于 Pydantic，自动序列化和反序列化

### 1.2 安装

```bash
pip install fastapi uvicorn

# 可选：标准依赖（包含 ORJSON、Jinja2 等）
pip install fastapi[standard]
```

---

## 2. 快速开始

### 2.1 最小应用

```python
# main.py
from fastapi import FastAPI

app = FastAPI(title="我的 API", version="1.0")

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

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}
```

### 2.2 运行服务

```bash
# 开发模式
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# 生产模式
uvicorn main:app --workers 4

# 使用 Hypercorn（HTTP/2 支持）
pip install hypercorn
hypercorn main:app --bind 0.0.0.0:8000
```

### 2.3 自动文档

启动后访问：
- **Swagger UI**：`http://localhost:8000/docs`
- **ReDoc**：`http://localhost:8000/redoc`
- **OpenAPI JSON**：`http://localhost:8000/openapi.json`

---

## 3. 路径操作

### 3.1 HTTP 方法

```python
from fastapi import FastAPI

app = FastAPI()

@app.get("/users")
def list_users():
    """获取用户列表"""
    return [{"id": 1, "name": "张三"}]

@app.post("/users")
def create_user():
    """创建用户"""
    return {"id": 2, "created": True}

@app.put("/users/{user_id}")
def update_user(user_id: int):
    """更新用户"""
    return {"id": user_id, "updated": True}

@app.patch("/users/{user_id}")
def partial_update(user_id: int):
    """部分更新"""
    return {"id": user_id, "patched": True}

@app.delete("/users/{user_id}")
def delete_user(user_id: int):
    """删除用户"""
    return {"id": user_id, "deleted": True}
```

### 3.2 路径参数

```python
from fastapi import FastAPI
from enum import Enum

app = FastAPI()

# 基础路径参数
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

# 枚举限制
class Role(str, Enum):
    admin = "admin"
    user = "user"
    guest = "guest"

@app.get("/roles/{role}")
def get_role(role: Role):
    if role == Role.admin:
        return {"role": role, "permissions": "all"}
    return {"role": role, "permissions": "limited"}

# 路径包含斜杠
@app.get("/files/{file_path:path}")
def read_file(file_path: str):
    return {"file_path": file_path}
```

---

## 4. 查询参数与请求体

### 4.1 查询参数

```python
from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# 可选参数带默认值
@app.get("/items/")
def read_items(
    skip: int = 0,
    limit: int = 10,
    search: Optional[str] = None
):
    return {
        "skip": skip,
        "limit": limit,
        "search": search
    }

# 布尔参数
@app.get("/items/filter/")
def filter_items(
    published: bool = False,
    category: Optional[str] = None
):
    return {
        "published_only": published,
        "category": category
    }

# 必需查询参数（无默认值）
@app.get("/calculate/")
def calculate(a: int, b: int, operation: str = "add"):
    if operation == "add":
        result = a + b
    elif operation == "multiply":
        result = a * b
    else:
        result = None
    return {"result": result}
```

### 4.2 Pydantic 模型（请求体）

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

app = FastAPI()

class UserBase(BaseModel):
    """用户基础模型"""
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    """创建用户"""
    password: str = Field(..., min_length=6)

class UserResponse(UserBase):
    """用户响应"""
    id: int
    created_at: datetime
    is_active: bool = True
    
    class Config:
        from_attributes = True

class Item(BaseModel):
    """商品模型"""
    name: str = Field(..., description="商品名称")
    description: Optional[str] = Field(None, max_length=300)
    price: float = Field(..., gt=0, description="必须大于 0")
    tax: Optional[float] = None
    tags: list[str] = []

# 创建用户
@app.post("/users", response_model=UserResponse)
def create_user(user: UserCreate):
    """创建新用户
    
    - **username**: 用户名（3-50 字符）
    - **email**: 有效邮箱
    - **password**: 至少 6 位
    """
    # 模拟创建
    return {
        "id": 1,
        **user.model_dump(),
        "created_at": datetime.now(),
        "is_active": True
    }

# 创建商品
@app.post("/items")
def create_item(item: Item):
    """创建商品"""
    item_dict = item.model_dump()
    if item.tax:
        item_dict["total"] = item.price + item.tax
    return item_dict

# 更新商品（部分更新）
@app.patch("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.model_dump()}
```

---

## 5. 响应处理

### 5.1 自定义响应

```python
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse, HTMLResponse, PlainTextResponse

app = FastAPI()

# 自定义状态码
@app.post("/items", status_code=201)
def create_item():
    return {"message": "创建成功"}

# 显式响应模型
@app.get("/legacy/")
def get_legacy():
    data = {"message": "旧版数据"}
    return JSONResponse(
        content=data,
        headers={"X-API-Version": "1.0"}
    )

# HTML 响应
@app.get("/page", response_class=HTMLResponse)
def read_page():
    return "<h1>Hello HTML</h1>"

# 设置 Cookie
@app.get("/set-cookie")
def set_cookie(response: Response):
    response.set_cookie(key="session", value="abc123")
    return {"message": "Cookie 已设置"}

# 文件下载
from fastapi.responses import FileResponse

@app.get("/download")
def download_file():
    return FileResponse(
        path="report.pdf",
        filename="年度报告.pdf",
        media_type="application/pdf"
    )
```

---

## 6. 依赖注入

### 6.1 基础依赖

```python
from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Optional

app = FastAPI()

# 定义依赖函数
async def get_token_header(x_token: str = Header(...)):
    """验证 Token"""
    if x_token != "secret-token":
        raise HTTPException(status_code=400, detail="无效的 Token")
    return x_token

async def common_parameters(
    q: Optional[str] = None,
    skip: int = 0,
    limit: int = 100
):
    """通用分页参数"""
    return {"q": q, "skip": skip, "limit": limit}

# 使用依赖
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/admin/items/")
def read_admin_items(
    commons: dict = Depends(common_parameters),
    token: str = Depends(get_token_header)
):
    return {"items": [], "token": token}
```

### 6.2 类依赖

```python
class CommonQueryParams:
    """通用查询参数类"""
    def __init__(
        self,
        q: Optional[str] = None,
        skip: int = 0,
        limit: int = 100
    ):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/books/")
def read_books(commons: CommonQueryParams = Depends()):
    return {
        "query": commons.q,
        "pagination": {
            "skip": commons.skip,
            "limit": commons.limit
        }
    }
```

### 6.3 数据库依赖

```python
from fastapi import FastAPI, Depends
from contextlib import contextmanager

# 模拟数据库连接
class Database:
    def connect(self):
        print("数据库连接")
        return self
    
    def disconnect(self):
        print("数据库断开")
    
    def query(self, sql: str):
        return [{"id": 1, "name": "Test"}]

db = Database()

# 依赖注入获取数据库连接
async def get_db():
    conn = db.connect()
    try:
        yield conn
    finally:
        conn.disconnect()

@app.get("/users/")
def get_users(db_conn = Depends(get_db)):
    return db_conn.query("SELECT * FROM users")
```

---

## 7. 错误处理

```python
from fastapi import FastAPI, HTTPException, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

# 自定义异常
class ItemNotFoundException(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# 注册异常处理器
@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request, exc: ItemNotFoundException):
    return JSONResponse(
        status_code=404,
        content={"detail": f"商品 {exc.item_id} 不存在"}
    )

@app.exception_handler(RequestValidationError)
async def validation_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "数据验证失败",
            "errors": exc.errors()
        }
    )

# 使用 HTTPException
@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id < 1:
        raise HTTPException(
            status_code=400,
            detail="ID 必须大于 0"
        )
    
    # 模拟查找
    items = {1: "Apple", 2: "Banana"}
    if item_id not in items:
        raise ItemNotFoundException(item_id)
    
    return {"item": items[item_id]}

# 使用状态码常量
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    # ... 删除逻辑
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="无权删除此商品"
    )
```

---

## 8. 异步支持

```python
from fastapi import FastAPI
import asyncio

app = FastAPI()

# 异步路由
@app.get("/async-data")
async def get_async_data():
    await asyncio.sleep(1)  # 模拟异步操作
    return {"data": "异步获取的数据"}

# 混合使用（同步函数在线程池中运行）
def cpu_intensive_task(n: int):
    """CPU 密集型任务"""
    return sum(range(n))

@app.get("/compute/{n}")
async def compute(n: int):
    # FastAPI 自动在线程池中运行同步函数
    result = cpu_intensive_task(n)
    return {"result": result}

# 异步数据库操作（示例）
async def fetch_from_db(query: str):
    await asyncio.sleep(0.1)
    return [{"id": 1, "data": "result"}]

@app.get("/users-async")
async def get_users_async():
    users = await fetch_from_db("SELECT * FROM users")
    return {"users": users}
```

---

## 9. 后台任务

```python
from fastapi import FastAPI, BackgroundTasks
from typing import Optional
import time

app = FastAPI()

def send_email(email: str, message: str):
    """发送邮件（模拟）"""
    time.sleep(2)
    print(f"邮件已发送至 {email}: {message}")

def write_log(message: str):
    """写入日志"""
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")

@app.post("/send-notification/{email}")
def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    """发送通知（后台执行）"""
    background_tasks.add_task(send_email, email, "感谢您的注册！")
    background_tasks.add_task(write_log, f"发送邮件至 {email}")
    
    return {"message": "通知将在后台发送"}
```

---

## 10. 中间件

```python
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# 使用装饰器创建中间件
@app.middleware("http")
async def add_process_time_header(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)
    return response

# 使用类创建中间件
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        print(f"请求: {request.method} {request.url}")
        response = await call_next(request)
        print(f"响应: {response.status_code}")
        return response

app.add_middleware(LoggingMiddleware)

# CORS 中间件
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
```

---

## 11. 小结

| 特性 | FastAPI 实现 | 说明 |
|------|-------------|------|
| 路由 | `@app.get()` 等 | 方法装饰器 |
| 数据验证 | Pydantic 模型 | 类型提示驱动 |
| 自动文档 | `/docs`, `/redoc` | OpenAPI 自动生成 |
| 依赖注入 | `Depends()` | 函数/类依赖 |
| 异步 | `async def` | 原生 ASGI 支持 |
| 后台任务 | `BackgroundTasks` | 非阻塞执行 |
| 中间件 | `@app.middleware()` | 请求处理管道 |

---

## 12. 练习题

1. 使用 FastAPI 创建一个完整的 TODO API，包含 CRUD 和过滤功能。
2. 为上述 API 添加 JWT 认证（使用 `fastapi-users` 或手写）。
3. 使用 Pydantic 模型实现复杂的数据验证（如邮箱格式、密码强度）。
4. 对比 FastAPI 与 Flask/Django 的相同功能实现，总结各自优势。

---

> **下节预告**：我们将对比 Django、Flask 和 FastAPI 三个主流框架，分析各自适用场景，帮助你做出技术选型决策。