FastAPI 开发规范与最佳实践

任务目标

• 本 Skill 用于：构建生产级 FastAPI 服务
• 能力包含：项目结构规范、异步防阻塞开发、Bearer Token 认证、日志滚动配置
• 触发条件：用户需要创建 FastAPI 项目、设计异步接口、实现鉴权或配置日志

前置准备

• Python 3.10+
• 推荐使用虚拟环境：python -m venv venv && source venv/bin/activate
• 安装依赖：pip install -r requirements.txt

操作步骤

1. 项目初始化

使用脚手架脚本生成标准项目结构：

python /workspace/projects/fastapi-dev-guide/scripts/init_project.py --name my_api --output ./my_project

生成的目录结构：

my_api/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用入口
│   ├── config.py            # 配置管理
│   ├── api/                 # API 路由
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       ├── endpoints/
│   │       │   ├── __init__.py
│   │       │   ├── auth.py
│   │       │   └── items.py
│   │       └── router.py
│   ├── core/                # 核心模块
│   │   ├── __init__.py
│   │   ├── security.py      # 认证相关
│   │   └── logging.py       # 日志配置
│   ├── models/              # 数据库模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/             # Pydantic 模型
│   │   ├── __init__.py
│   │   └── user.py
│   ├── services/           # 业务逻辑
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── db/                  # 数据库
│       ├── __init__.py
│       └── session.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   └── test_api_v1/
├── logs/                    # 日志目录
├── requirements.txt
└── .env.example

2. 异步防阻塞开发规范

核心原则：任何可能阻塞的事件循环操作必须使用异步版本或在线程池中执行。

2.1 正确使用 sync/async 函数

| 操作类型 | 正确做法 | 错误做法 | |---------|---------|---------| | 数据库查询 | 使用 asyncpg、SQLAlchemy async | 使用同步 psycopg2 | | HTTP 请求 | 使用 httpx.AsyncClient | 使用同步 requests | | 文件 I/O | 使用 aiofiles 或在线程池执行 | 直接 open() 同步读写 | | CPU 密集型 | 使用 run_in_executor() | 直接同步执行 |

2.2 接口设计检查清单

编写接口时检查以下问题：

# ❌ 错误示例：同步数据库操作会阻塞事件循环
@app.get("/items")
def get_items():
    conn = psycopg2.connect(DATABASE_URL)  # 阻塞！
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM items")
    return cursor.fetchall()

# ✅ 正确示例：异步数据库操作
@app.get("/items")
async def get_items(session: AsyncSession):
    result = await session.execute(select(Item))
    return result.scalars().all()

2.3 常见阻塞场景与解决方案

场景 1：同步 HTTP 请求

# ❌ 错误
import requests
@app.get("/external-data")
async def get_external():
    response = requests.get("https://api.example.com/data")  # 阻塞整个事件循环
    return response.json()

# ✅ 正确
import httpx
@app.get("/external-data")
async def get_external():
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get("https://api.example.com/data")
        return response.json()

场景 2：CPU 密集型计算

# ❌ 错误 - 阻塞事件循环
import hashlib
@app.post("/hash")
async def compute_hash(data: str):
    result = hashlib.sha256(data.encode()).hexdigest()  # CPU 密集计算
    return {"hash": result}

# ✅ 正确 - 使用线程池
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)

@app.post("/hash")
async def compute_hash(data: str):
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        executor, 
        lambda: hashlib.sha256(data.encode()).hexdigest()
    )
    return {"hash": result}

场景 3：同步文件操作

# ❌ 错误
@app.post("/upload")
async def upload_file(file: UploadFile):
    content = await file.read()
    with open(f"./uploads/{file.filename}", "wb") as f:
        f.write(content)  # 阻塞
    return {"filename": file.filename}

# ✅ 正确 - 使用 aiofiles
import aiofiles
@app.post("/upload")
async def upload_file(file: UploadFile):
    content = await file.read()
    async with aiofiles.open(f"./uploads/{file.filename}", "wb") as f:
        await f.write(content)
    return {"filename": file.filename}

3. Bearer Token 认证实现

3.1 安全配置（core/security.py）

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

# JWT 配置
SECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# 密码哈希
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# Bearer Token 方案
security = HTTPBearer()

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(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication token",
            headers={"WWW-Authenticate": "Bearer"},
        )

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security)
):
    token = credentials.credentials
    payload = decode_token(token)
    user_id: str = payload.get("sub")
    if user_id is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token payload"
        )
    # 这里应该从数据库获取用户
    return {"user_id": user_id}

3.2 受保护的路由

from fastapi import APIRouter, Depends

router = APIRouter()

@router.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
    return {"user_id": current_user["user_id"], "message": "Protected resource"}

@router.post("/items")
async def create_item(
    item: ItemCreate,
    current_user: dict = Depends(get_current_user)
):
    # 创建逻辑
    return {"item": item, "owner": current_user["user_id"]}

3.3 登录接口

from fastapi import APIRouter, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

@router.post("/login", response_model=Token)
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = await authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password"
        )
    access_token = create_access_token(
        data={"sub": user.username, "role": user.role}
    )
    return {"access_token": access_token, "token_type": "bearer"}

# Schema
from pydantic import BaseModel
class Token(BaseModel):
    access_token: str
    token_type: str

4. 日志滚动配置

4.1 配置说明

使用 TimedRotatingFileHandler 实现按时间滚动的日志：

| 参数 | 说明 | |------|------| | when | 滚动时机：S(秒)、M(分)、H(时)、D(天)、W0-W6(周)、midnight(午夜) | | interval | 滚动间隔，配合 when 使用 | | backupCount | 保留的备份文件数量 | | encoding | 文件编码，建议使用 utf-8 |

4.2 日志配置（core/logging.py）

import logging
import sys
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path

LOG_DIR = Path("logs")
LOG_DIR.mkdir(exist_ok=True)

def setup_logging(
    app_name: str = "fastapi",
    level: int = logging.INFO,
    when: str = "midnight",
    interval: int = 1,
    backup_count: int = 30
) -> logging.Logger:
    """配置日志系统，支持按时间滚动"""
    
    # 创建 logger
    logger = logging.getLogger(app_name)
    logger.setLevel(level)
    logger.handlers.clear()
    
    # 日志格式
    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)-8s | %(name)s:%(funcName)s:%(lineno)d | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )
    
    # 控制台处理器
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(level)
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)
    
    # 文件处理器 - 按时间滚动
    file_handler = TimedRotatingFileHandler(
        filename=LOG_DIR / f"{app_name}.log",
        when=when,
        interval=interval,
        backupCount=backup_count,
        encoding="utf-8",
        utc=True  # 使用 UTC 时间，避免时区问题
    )
    file_handler.setLevel(level)
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)
    
    # 错误日志单独记录
    error_handler = TimedRotatingFileHandler(
        filename=LOG_DIR / f"{app_name}_error.log",
        when="D",  # 每天
        interval=1,
        backupCount=7,
        encoding="utf-8"
    )
    error_handler.setLevel(logging.ERROR)
    error_handler.setFormatter(formatter)
    logger.addHandler(error_handler)
    
    return logger

# 使用 uvicorn 日志配置
def get_uvicorn_log_config():
    return {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "default": {
                "format": "%(asctime)s | %(levelname)-8s | %(message)s",
            },
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "formatter": "default",
                "stream": "ext://sys.stdout",
            },
        },
        "loggers": {
            "uvicorn": {"handlers": ["console"], "level": "INFO"},
            "uvicorn.error": {"level": "INFO"},
            "uvicorn.access": {"handlers": ["console"], "level": "INFO"},
        },
        "root": {"level": "INFO", "handlers": ["console"]},
    }

4.3 在应用中使用

# app/main.py
from app.core.logging import setup_logging, get_uvicorn_log_config

logger = setup_logging("my_api")

app = FastAPI(title="My API")

@app.on_event("startup")
async def startup():
    logger.info("Application starting up")

@app.on_event("shutdown")
async def shutdown():
    logger.info("Application shutting down")

# 带日志的请求示例
@app.middleware("http")
async def log_requests(request: Request, call_next):
    logger.info(f"Request: {request.method} {request.url.path}")
    response = await call_next(request)
    logger.info(f"Response: {response.status_code}")
    return response

5. 生产环境配置

5.1 使用 Gunicorn + Uvicorn Workers

# gunicorn.conf.py
import multiprocessing

bind = "0.0.0.0:8000"
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
keepalive = 65
timeout = 30
graceful_timeout = 30

# 日志
accesslog = "-"
errorlog = "-"
loglevel = "info"

5.2 运行命令

# 开发环境
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 生产环境
gunicorn app.main:app -c gunicorn.conf.py

使用示例

示例 1：创建新项目

• 场景：用户需要快速创建一个符合规范的 FastAPI 项目
• 预期产出：完整的项目目录结构和基础文件
• 关键要点：运行脚手架脚本生成标准结构

示例 2：编写防阻塞接口

• 场景：需要调用外部 API 获取数据
• 预期产出：异步 HTTP 请求的正确实现
• 关键要点：使用 httpx.AsyncClient 而非 requests

示例 3：实现用户认证

• 场景：需要保护某些 API 端点
• 预期产出：带 Bearer Token 验证的受保护路由
• 关键要点：使用 Depends(get_current_user) 保护端点

示例 4：配置日志滚动

• 场景：部署到生产环境需要日志管理
• 预期产出：按天滚动的日志文件
• 关键要点：使用 TimedRotatingFileHandler 配置

资源索引

• 脚本：见 scripts/init_project.py（项目脚手架生成，参数：--name 项目名 --output 输出目录）
• 参考：见 references/async_best_practices.md（异步编程防阻塞详细规范与常见陷阱）
• 参考：见 references/auth_implementation.md（Bearer Token 完整实现方案与安全建议）
• 参考：见 references/logging_config.md（日志滚动配置详解与生产环境实践）
• 模板：见 assets/templates/（可直接复制使用的代码模板）

注意事项

• Bearer Token 的 SECRET_KEY 必须使用强随机值，禁止硬编码
• 日志 backupCount 设置要合理，避免磁盘空间耗尽
• 生产环境禁止使用 reload=True
• 所有外部 I/O 操作必须使用异步版本
• CPU 密集型任务必须使用 run_in_executor