🐍 第26课：微服务调用

—— FastAPI：现代高性能 Python Web 框架

📌 本课目标

1️⃣ FastAPI 入门

"""FastAPI 三大核心优势：
1. 极致性能 —— 基于 Starlette + Pydantic，性能媲美 NodeJS/Go
2. 开发效率 —— 类型提示即文档，自动生成 OpenAPI 文档
3. 编辑器支持 —— IDE 自动补全，类型检查覆盖全链路

安装：pip install fastapi uvicorn
"""

from fastapi import FastAPI

app = FastAPI(title="我的第一个API", version="1.0.0")

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

@app.get("/health")
def health_check():
    return {"status": "ok", "version": "1.0.0"}

# 启动：uvicorn main:app --reload
# 自动文档：http://localhost:8000/docs
# 备用文档：http://localhost:8000/redoc

2️⃣ Pydantic 数据模型

from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime

app = FastAPI()

class TaskCreate(BaseModel):
    """创建任务的数据模型"""
    title: str = Field(..., min_length=1, max_length=100, description="任务标题")
    description: Optional[str] = Field(None, max_length=500, description="任务描述")
    priority: int = Field(1, ge=1, le=5, description="优先级 1-5")
    due_date: Optional[str] = Field(None, description="截止日期 YYYY-MM-DD")

class TaskResponse(TaskCreate):
    """任务响应模型"""
    id: int
    created_at: str
    status: str = "pending"

# 内存存储（演示用）
tasks_db = []
task_counter = 0

@app.post("/tasks", response_model=TaskResponse)
def create_task(task: TaskCreate):
    """创建任务"""
    global task_counter
    task_counter += 1
    new_task = TaskResponse(
        id=task_counter,
        title=task.title,
        description=task.description,
        priority=task.priority,
        due_date=task.due_date,
        created_at=datetime.now().isoformat(),
    )
    tasks_db.append(new_task)
    return new_task

@app.get("/tasks", response_model=list[TaskResponse])
def list_tasks(priority: Optional[int] = None, status: Optional[str] = None):
    """列出任务（支持筛选）"""
    result = tasks_db
    if priority:
        result = [t for t in result if t.priority == priority]
    if status:
        result = [t for t in result if t.status == status]
    return result

@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    """获取单个任务"""
    for task in tasks_db:
        if task.id == task_id:
            return task
    from fastapi import HTTPException
    raise HTTPException(status_code=404, detail="任务不存在")

@app.put("/tasks/{task_id}/complete")
def complete_task(task_id: int):
    """完成任务"""
    for task in tasks_db:
        if task.id == task_id:
            task.status = "completed"
            return task
    raise HTTPException(status_code=404, detail="任务不存在")

3️⃣ 子进程启动 FastAPI

import subprocess
import time
import requests  # pip install requests

# 用 subprocess 启动 FastAPI 服务
def start_api_server(script_path="api_server.py", port=8000):
    """启动 API 服务"""
    proc = subprocess.Popen(
        ["uvicorn", "api_server:app", "--host", "0.0.0.0", "--port", str(port)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    time.sleep(2)  # 等待服务启动
    return proc

def call_api(method, url, **kwargs):
    """调用 API"""
    resp = getattr(requests, method)(url, **kwargs)
    return resp.json(), resp.status_code

# 使用示例
# proc = start_api_server()
# data, code = call_api("post", "http://localhost:8000/tasks", 
#                       json={"title": "学习FastAPI", "priority": 3})
# print(f"状态码: {code}, 响应: {data}")
# proc.terminate()

4️⃣ 中间件与异常处理

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import time

app = FastAPI()

# 计时中间件
@app.middleware("http")
async def timer_middleware(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    elapsed = time.time() - start
    response.headers["X-Process-Time"] = f"{elapsed:.3f}s"
    return response

# 全局异常处理
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content={"detail": str(exc)}
    )

# CORS 支持
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

5️⃣ 验证脚本

#!/usr/bin/env python3
"""第26课 FastAPI验证"""
from fastapi import FastAPI
from pydantic import BaseModel, Field
from fastapi.testclient import TestClient

app = FastAPI()

class Item(BaseModel):
    name: str = Field(..., min_length=1)
    price: float = Field(..., gt=0)

items = []

@app.get("/items")
def list_items():
    return items

@app.post("/items")
def create_item(item: Item):
    items.append(item)
    return item

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if 0 <= item_id < len(items):
        return items[item_id]
    from fastapi import HTTPException
    raise HTTPException(status_code=404, detail="Not found")

client = TestClient(app)

def test_crud():
    # Create
    resp = client.post("/items", json={"name": "Book", "price": 29.9})
    assert resp.status_code == 200
    data = resp.json()
    assert data["name"] == "Book"
    
    # List
    resp = client.get("/items")
    assert resp.status_code == 200
    assert len(resp.json()) == 1
    
    # Get
    resp = client.get("/items/0")
    assert resp.status_code == 200
    
    # 404
    resp = client.get("/items/999")
    assert resp.status_code == 404
    
    # Validation error
    resp = client.post("/items", json={"name": "", "price": -1})
    assert resp.status_code == 422
    
    print("✅ FastAPI CRUD测试通过")

def test_pydantic():
    item = Item(name="Test", price=10.0)
    assert item.name == "Test"
    assert item.price == 10.0
    
    try:
        Item(name="", price=10.0)
        assert False, "Should have raised"
    except:
        pass
    
    try:
        Item(name="OK", price=-1)
        assert False, "Should have raised"
    except:
        pass
    
    print("✅ Pydantic校验测试通过")

if __name__ == "__main__":
    test_crud()
    test_pydantic()
    print("\n🎉 第26课全部验证通过！")

6️⃣ 文件上传与下载

from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import FileResponse
from pathlib import Path
import uuid

app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    """上传文件"""
    ext = Path(file.filename).suffix
    unique_name = f"{uuid.uuid4().hex}{ext}"
    file_path = UPLOAD_DIR / unique_name
    with open(file_path, "wb") as f:
        content = await file.read()
        f.write(content)
    return {"filename": file.filename, "saved_as": unique_name, "size": len(content)}

@app.get("/download/{filename}")
async def download_file(filename: str):
    """下载文件"""
    file_path = UPLOAD_DIR / filename
    if not file_path.exists():
        raise HTTPException(status_code=404, detail="文件不存在")
    return FileResponse(file_path, filename=filename)

7️⃣ API 认证与鉴权

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

app = FastAPI()

API_TOKENS = {
    "token_admin_123": {"role": "admin", "name": "管理员"},
    "token_user_456": {"role": "user", "name": "普通用户"},
}

async def verify_token(authorization: Optional[str] = Header(None)):
    """验证 API Token"""
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="未提供认证信息")
    token = authorization[7:]
    if token not in API_TOKENS:
        raise HTTPException(status_code=403, detail="无效的Token")
    return API_TOKENS[token]

@app.get("/me")
async def get_me(user=Depends(verify_token)):
    return user

8️⃣ FastAPI 完整 CRUD 服务

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

app = FastAPI(title="任务管理 API", version="1.0.0")

class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None
    priority: int = Field(1, ge=1, le=5)
    due_date: Optional[str] = None

class TaskResponse(TaskCreate):
    id: int
    status: str = "pending"
    created_at: str = ""

# 内存存储
tasks_db = {}
task_counter = 0

@app.post("/tasks", response_model=TaskResponse, status_code=201)
def create_task(task: TaskCreate):
    global task_counter
    task_counter += 1
    new_task = TaskResponse(
        id=task_counter, title=task.title, description=task.description,
        priority=task.priority, due_date=task.due_date,
        created_at=datetime.now().isoformat(),
    )
    tasks_db[task_counter] = new_task
    return new_task

@app.get("/tasks", response_model=List[TaskResponse])
def list_tasks(status: Optional[str] = None, priority: Optional[int] = None):
    result = list(tasks_db.values())
    if status:
        result = [t for t in result if t.status == status]
    if priority:
        result = [t for t in result if t.priority == priority]
    return result

@app.get("/tasks/{task_id}", response_model=TaskResponse)
def get_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(404, "任务不存在")
    return tasks_db[task_id]

@app.put("/tasks/{task_id}", response_model=TaskResponse)
def update_task(task_id: int, task: TaskCreate):
    if task_id not in tasks_db:
        raise HTTPException(404, "任务不存在")
    existing = tasks_db[task_id]
    updated = TaskResponse(
        id=task_id, title=task.title, description=task.description,
        priority=task.priority, due_date=task.due_date,
        status=existing.status, created_at=existing.created_at,
    )
    tasks_db[task_id] = updated
    return updated

@app.delete("/tasks/{task_id}")
def delete_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(404, "任务不存在")
    del tasks_db[task_id]
    return {"message": "删除成功"}

@app.patch("/tasks/{task_id}/complete")
def complete_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(404, "任务不存在")
    tasks_db[task_id].status = "completed"
    return tasks_db[task_id]

9️⃣ API 文档与版本管理

"""FastAPI 自动文档：
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
- OpenAPI JSON: http://localhost:8000/openapi.json

版本管理策略：
1. URL 版本：/api/v1/tasks, /api/v2/tasks
2. Header 版本：Accept: application/vnd.api.v1+json
3. 查询参数：?version=1

推荐 URL 版本，最直观，缓存友好
"""

🔑 本课要点