python API 之 fastapi

 

为什么选择 FastAPI？

• 高性能：基于 Starlette 和 Uvicorn，支持异步请求处理

• 开发效率：自动交互文档、类型提示、代码自动补全

• 现代标准：兼容 OpenAPI 和 JSON Schema

• 易扩展：模块化设计，支持中间件和依赖注入

 

以下是一个使用 FastAPI 最新特性 (0.109+) 的完整示例，涵盖核心功能并包含详细注释：

```python
# main.py
from typing import Annotated, Optional
from datetime import datetime, timedelta
from contextlib import asynccontextmanagerfrom fastapi import (
    FastAPI, 
    Depends, 
    HTTPException, 
    status, 
    Query, 
    Body,
    WebSocket,
    Request
)
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
from pydantic import BaseModel, EmailStr, Field
from jose import JWTError, jwt
from passlib.context import CryptContext# ---------------------------
# 应用生命周期与配置
# ---------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
    """生命周期管理 (FastAPI 2.2+ 新特性)"""
    print("Application startup")
    yield
    print("Application shutdown")app = FastAPI(
    title="Modern FastAPI Demo",
    lifespan=lifespan,
    swagger_ui_parameters={"syntaxHighlight": False}
)# 配置中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)# 静态文件服务
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")# ---------------------------
# 安全配置
# ---------------------------
SECRET_KEY = "your-secret-key-keep-it-secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")# ---------------------------
# 数据模型
# ---------------------------
class UserBase(BaseModel):
    username: str = Field(..., example="john_doe")
    email: EmailStr = Field(..., example="john@example.com")class UserCreate(UserBase):
    password: str = Field(..., min_length=8, example="secret123")class UserInDB(UserBase):
    hashed_password: strclass Token(BaseModel):
    access_token: str
    token_type: strclass TokenData(BaseModel):
    username: Optional[str] = None# ---------------------------
# 工具函数
# ---------------------------
def fake_hash_password(password: str):
    return pwd_context.hash(password)def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)def create_access_token(data: dict, expires_delta: timedelta):
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)# ---------------------------
# 依赖项
# ---------------------------
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    # 这里应查询真实数据库
    user = UserInDB(
        username=token_data.username,
        email="user@example.com", 
        hashed_password="fakehashedsecret"
    )
    return user# ---------------------------
# 路由
# ---------------------------
@app.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    """返回HTML页面 (集成前端示例)"""
    return templates.TemplateResponse(
        "index.html",
        {"request": request, "message": "Welcome to FastAPI!"}
    )@app.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    """创建用户 (数据验证示例)"""
    hashed_password = fake_hash_password(user.password)
    return {
        "username": user.username,
        "email": user.email,
        "hashed_password": hashed_password
    }@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    """OAuth2 认证 (安全示例)"""
    # 验证用户逻辑应替换为真实数据库查询
    user = UserInDB(
        username=form_data.username,
        email="user@example.com",
        hashed_password=fake_hash_password(form_data.password)
    )
    if not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, 
        expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}@app.get("/users/me")
async def read_users_me(
    current_user: Annotated[UserInDB, Depends(get_current_user)]
):
    """需要认证的端点示例"""
    return current_user@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket 示例"""
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message received: {data}")@app.get("/items/")
async def read_items(
    q: Annotated[
        Optional[str], 
        Query(
            title="Query string",
            description="Optional search query",
            min_length=3,
            max_length=50
        )
    ] = None,
    page: Annotated[int, Query(ge=1)] = 1,
    size: Annotated[int, Query(ge=1, le=100)] = 10
):
    """复杂查询参数示例"""
    return {"q": q, "page": page, "size": size}@app.post("/compute/")
async def compute_task(
    numbers: Annotated[list[int], Body(embed=True)],
    delay: Annotated[bool, Body()] = False
):
    """后台任务示例"""
    if delay:
        # 模拟长时间任务
        import time
        time.sleep(2)
    
    total = sum(numbers)
    return {"total": total, "count": len(numbers)}# ---------------------------
# 错误处理
# ---------------------------
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
        headers=exc.headers
    )# ---------------------------
# 运行与测试
# ---------------------------
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:app",
        host="0.0.0.0", 
        port=8000,
        reload=True,
        ssl_keyfile="key.pem",  # HTTPS 示例（需生成证书）
        ssl_certfile="cert.pem"
    )
```

### 配套文件结构
```
project/
├── static/
│   └── style.css
├── templates/
│   └── index.html
├── main.py
├── requirements.txt
├── cert.pem
└── key.pem
```

### 主要功能说明

1. **最新生命周期管理**  
使用 `@asynccontextmanager` 管理应用生命周期（FastAPI 2.2+）

2. **安全认证系统**  
- OAuth2 密码流认证
- JWT Token 生成与验证
- 密码哈希加密（使用 passlib）

3. **现代化数据验证**  
- Pydantic V2 模型验证
- 参数注解式声明（Annotated）
- 复杂查询参数约束

4. **高级路由功能**  
- WebSocket 支持
- 静态文件服务
- Jinja2 模板渲染
- 后台任务处理

5. **生产级配置**  
- CORS 中间件
- HTTPS 支持
- 结构化错误处理
- 自动生成文档（Swagger/ReDoc）

6. **类型提示增强**  
- 使用 Python 3.10+ 的类型注解
- 依赖注入系统增强
- 参数元数据声明

### 运行与测试

1. 安装依赖：
```bash
pip install fastapi uvicorn[standard] python-jose[cryptography] passlib jinja2
```

2. 启动服务：
```bash
python main.py
```

3. 访问功能：
- API 文档：`https://localhost:8000/docs`
- WebSocket 测试：`ws://localhost:8000/ws`
- 前端页面：`https://localhost:8000/`

### 测试示例（使用 HTTPie）
```bash
# 创建用户
http POST :8000/users/ username=john email=john@test.com password=secret123

# 获取Token
http POST :8000/token username=john password=secret123

# 访问受保护端点
http GET :8000/users/me "Authorization: Bearer <token>"

# WebSocket测试
websocat ws://localhost:8000/ws
```

这个示例展示了 FastAPI 的现代用法，包含最新的特性实现方式，建议结合官方文档和实际项目需求进行扩展。