Python Web 框架：FastAPI、Flask 与 Django 深度对比与实战

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List

app = FastAPI(title="Task API", version="1.0.0")

class Task(BaseModel):
    id: int
    title: str
    completed: bool = False
    
    class Config:
        json_schema_extra = {
            "example": {
                "id": 1,
                "title": "Learn FastAPI",
                "completed": False
            }
        }

# 内存存储（实际用数据库）
tasks_db: dict[int, Task] = {}

@app.get("/tasks", response_model=List[Task])
async def list_tasks():
    return list(tasks_db.values())

@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
async def create_task(task: Task):
    if task.id in tasks_db:
        raise HTTPException(status_code=400, detail="Task already exists")
    tasks_db[task.id] = task
    return task

@app.get("/tasks/{task_id}", response_model=Task)
async def get_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Task not found")
    return tasks_db[task_id]

@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: int, task: Task):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Task not found")
    tasks_db[task_id] = task
    return task

@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(task_id: int):
    if task_id not in tasks_db:
        raise HTTPException(status_code=404, detail="Task not found")
    del tasks_db[task_id]

pip install fastapi uvicorn
uvicorn main:app --reload
# 自动文档: http://localhost:8000/docs (Swagger)
#              http://localhost:8000/redoc (ReDoc)

from fastapi import Depends, HTTPException
from typing import Annotated

# 依赖函数
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 可复用的分页依赖
def pagination_params(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100)
) -> PaginationParams:
    return PaginationParams(page=page, page_size=page_size)

# 认证依赖
def get_current_user(token: str = Header(...)) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return User(id=payload["sub"], name=payload["name"])
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

# 使用依赖
@app.get("/users/me")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_user)]
):
    return current_user

# 依赖嵌套
async def require_admin(
    user: Annotated[User, Depends(get_current_user)]
) -> User:
    if not user.is_admin:
        raise HTTPException(status_code=403, detail="Admin required")
    return user

from fastapi import BackgroundTasks
import asyncio

async def send_email(email: str, message: str):
    await asyncio.sleep(2)  # 模拟发送
    print(f"Email sent to {email}: {message}")

@app.post("/signup")
async def signup(
    user: UserCreate,
    background_tasks: BackgroundTasks
):
    # 同步响应（不等待邮件发送）
    background_tasks.add_task(send_email, user.email, "Welcome!")
    return {"message": "User registered"}

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start
        response.headers["X-Response-Time"] = f"{duration:.3f}s"
        return response

# 注册
app.add_middleware(TimingMiddleware)

# 另一个：CORS
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

class AppException(Exception):
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code

@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"success": False, "message": exc.message}
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "message": "Validation failed",
            "errors": exc.errors()
        }
    )

# 统一响应模型
class APIResponse(BaseModel):
    success: bool
    data: Any = None
    message: str | None = None

from sqlmodel import SQLModel, Field, create_engine, Session, select
from typing import Optional

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    email: str = Field(unique=True, index=True)
    is_active: bool = True

# 创建引擎
engine = create_engine("sqlite:///./app.db", echo=True)
SQLModel.metadata.create_all(engine)

# CRUD 操作
with Session(engine) as session:
    # Create
    user = User(name="Alice", email="alice@example.com")
    session.add(user)
    session.commit()
    
    # Read
    statement = select(User).where(User.name == "Alice")
    alice = session.exec(statement).first()
    
    # Update
    alice.is_active = False
    session.add(alice)
    session.commit()
    
    # Delete
    session.delete(alice)
    session.commit()

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# 使用 asyncpg（PostgreSQL）
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_async_db():
    async with async_session() as session:
        yield session

@app.get("/users")
async def list_users(db: AsyncSession = Depends(get_async_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

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

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta | None = None):
    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)

# OAuth2 密码流
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=401,
        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
    except JWTError:
        raise credentials_exception
    
    user = get_user(username)  # 从数据库获取
    if user is None:
        raise credentials_exception
    return user

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    token = create_access_token(data={"sub": user.username})
    return {"access_token": token, "token_type": "bearer"}

@app.get("/protected")
async def protected_route(current_user: User = Depends(get_current_user)):
    return {"message": f"Hello, {current_user.username}!"}

# Build stage
FROM python:3.12-slim as builder

WORKDIR /app
RUN pip install --no-cache-dir poetry
COPY pyproject.toml poetry.lock ./
RUN poetry config virtualenvs.create false \
    && poetry install --no-dev

# Production stage
FROM python:3.12-slim

WORKDIR /app
COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
COPY ./app ./app

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# 多 worker 模式（推荐）
gunicorn app.main:app \
    --workers 4 \
    --worker-class uvicorn.workers.UvicornWorker \
    --bind 0.0.0.0:8000 \
    --access-logfile - \
    --error-logfile -

# 或使用 uvicorn 原生（单进程）
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4

# project/
# ├── app/
# │   ├── __init__.py          # create_app() 工厂
# │   ├── extensions.py        # db, migrate, mail
# │   ├── api/
# │   │   ├── __init__.py
# │   │   └── users.py         # Blueprint 路由
# │   ├── models/
# │   │   └── user.py
# │   └── utils/
# │       └── errors.py        # 统一异常响应
# ├── tests/
# │   ├── conftest.py          # pytest fixture
# │   └── test_users.py        # 集成测试
# ├── config.py                # 多环境配置
# └── wsgi.py                  # 入口

# app/__init__.py
from flask import Flask
from .extensions import db
from .api.users import users_bp
from .utils.errors import register_error_handlers

def create_app(config_name="production"):
    app = Flask(__name__)
    app.config.from_object(f"config.{config_name}Config")

    db.init_app(app)
    app.register_blueprint(users_bp, url_prefix="/api/v1")
    register_error_handlers(app)

    return app

# app/api/users.py
from flask import Blueprint, jsonify, request
from app.models.user import User
from app.extensions import db

users_bp = Blueprint("users", __name__)

@users_bp.route("/users", methods=["GET"])
def list_users():
    users = User.query.all()
    return jsonify([u.to_dict() for u in users])

@users_bp.route("/users/<int:user_id>", methods=["GET"])
def get_user(user_id):
    user = User.query.get(user_id)
    if not user:
        return jsonify({"error": "Not found"}), 404
    return jsonify(user.to_dict())

@users_bp.route("/users", methods=["POST"])
def create_user():
    data = request.get_json()
    if not data or "username" not in data or "email" not in data:
        return jsonify({"error": "Invalid payload"}), 400
    user = User(username=data["username"], email=data["email"])
    db.session.add(user)
    db.session.commit()
    return jsonify(user.to_dict()), 201

# app/utils/errors.py
from flask import jsonify, current_app
from werkzeug.exceptions import HTTPException
from app.extensions import db

def register_error_handlers(app):
    @app.errorhandler(400)
    def bad_request(e):
        return jsonify({"success": False, "error": "Bad request"}), 400

    @app.errorhandler(404)
    def not_found(e):
        return jsonify({"success": False, "error": "Resource not found"}), 404

    @app.errorhandler(500)
    def internal_error(e):
        db.session.rollback()
        return jsonify({"success": False, "error": "Internal server error"}), 500

    @app.errorhandler(HTTPException)
    def handle_http_exception(e):
        return jsonify({"success": False, "error": e.description}), e.code

    @app.errorhandler(Exception)
    def handle_unexpected(e):
        current_app.logger.exception("Unhandled exception")
        return jsonify({"success": False, "error": "An unexpected error occurred"}), 500

# tests/conftest.py
import pytest
from app import create_app
from app.extensions import db

@pytest.fixture
def app():
    app = create_app("testing")
    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()

@pytest.fixture
def client(app):
    return app.test_client()

# tests/test_users.py
def test_create_user(client):
    resp = client.post("/api/v1/users", json={
        "username": "alice",
        "email": "alice@example.com"
    })
    assert resp.status_code == 201
    data = resp.get_json()
    assert data["username"] == "alice"
    assert data["email"] == "alice@example.com"

def test_get_user_not_found(client):
    resp = client.get("/api/v1/users/999")
    assert resp.status_code == 404
    data = resp.get_json()
    assert data["error"] == "Resource not found"

def test_create_user_missing_field(client):
    resp = client.post("/api/v1/users", json={"username": "bob"})
    assert resp.status_code == 400
    data = resp.get_json()
    assert "Invalid payload" in data["error"]

# myapp/views.py
from django.http import JsonResponse
from django.views import View

async def async_hello(request):
    return JsonResponse({"message": "Hello from async Django"})

class AsyncUserListView(View):
    async def get(self, request):
        from django.contrib.auth.models import User
        users = await User.objects.all().aall()
        data = [{"id": u.id, "username": u.username} for u in users]
        return JsonResponse({"users": data})

# myapp/views.py
from myapp.models import Product
from django.http import JsonResponse

async def get_products(request):
    count = await Product.objects.filter(active=True).acount()
    products = []
    async for product in Product.objects.filter(active=True).aiterator():
        products.append({
            "id": product.id,
            "name": product.name,
            "price": str(product.price)
        })
    return JsonResponse({"count": count, "products": products})

from asgiref.sync import sync_to_async
import time

def heavy_computation(n):
    time.sleep(n)
    return n * 2

async def async_compute(request, n):
    result = await sync_to_async(heavy_computation, thread_sensitive=False)(n)
    return JsonResponse({"result": result})

# myapp/middleware.py
import time

class AsyncTimingMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    async def __call__(self, request):
        start = time.perf_counter()
        response = await self.get_response(request)
        duration = time.perf_counter() - start
        response["X-Response-Time"] = f"{duration:.3f}s"
        return response