Python Web开发基础与框架对比

一、WSGI协议

WSGI（Web Server Gateway Interface）是Python Web应用与服务器之间的标准接口：

def simple_app(environ, start_response):
"""最简单的WSGI应用"""
status = '200 OK'
headers = [('Content-Type', 'text/plain; charset=utf-8')]
start_response(status, headers)
return [b'Hello, World!']

# environ包含请求信息
# PATH_INFO: 请求路径
# REQUEST_METHOD: HTTP方法
# QUERY_STRING: 查询参数
# wsgi.input: 请求体

# 使用内置服务器运行
from wsgiref.simple_server import make_server
server = make_server('localhost', 8000, simple_app)
server.serve_forever()


二、Flask基础

from flask import Flask, request, jsonify, abort

app = Flask(__name__)

# 路由与视图
@app.route('/')
def index():
return 'Hello, World!'

@app.route('/users/', methods=['GET'])
def get_user(user_id):
user = find_user(user_id)
if not user:
abort(404)
return jsonify(user)

@app.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
if not data or 'name' not in data:
return jsonify({'error': '缺少name字段'}), 400

user = save_user(data)
return jsonify(user), 201

# 错误处理
@app.errorhandler(404)
def not_found(error):
return jsonify({'error': '资源不存在'}), 404

@app.errorhandler(500)
def internal_error(error):
return jsonify({'error': '服务器内部错误'}), 500

# 中间件（before/after request）
@app.before_request
def log_request():
app.logger.info(f"{request.method} {request.path}")

@app.after_request
def add_cors_headers(response):
response.headers['Access-Control-Allow-Origin'] = '*'
return response


三、Flask蓝图与项目组织

# blueprints/users.py
from flask import Blueprint, request, jsonify

users_bp = Blueprint('users', __name__, url_prefix='/api/users')

@users_bp.route('/', methods=['GET'])
def list_users():
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 20, type=int)
users = User.query.paginate(page=page, per_page=per_page)
return jsonify({
'users': [u.to_dict() for u in users.items],
'total': users.total,
'page': page,
})

@users_bp.route('/', methods=['PUT'])
def update_user(user_id):
user = User.query.get_or_404(user_id)
data = request.get_json()
user.update(data)
db.session.commit()
return jsonify(user.to_dict())

# app.py
from flask import Flask
from blueprints.users import users_bp
from blueprints.orders import orders_bp

def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(configs[config_name])

# 注册蓝图
app.register_blueprint(users_bp)
app.register_blueprint(orders_bp)

# 初始化扩展
db.init_app(app)
migrate.init_app(app, db)

return app


四、FastAPI基础

from fastapi import FastAPI, HTTPException, Depends, Query, Path
from pydantic import BaseModel, EmailStr
from typing import Optional

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

# 请求/响应模型
class UserCreate(BaseModel):
name: str
email: EmailStr
age: Optional[int] = None

class UserResponse(BaseModel):
id: int
name: str
email: str
age: Optional[int]

class Config:
from_attributes = True

# 路由
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int = Path(..., gt=0)):
user = await db.get_user(user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user

@app.get("/users", response_model=list[UserResponse])
async def list_users(
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
name: Optional[str] = None,
):
return await db.get_users(skip=skip, limit=limit, name=name)

@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
return await db.create_user(user)

# 依赖注入
async def get_current_user(token: str = Depends(oauth2_scheme)):
user = await verify_token(token)
if not user:
raise HTTPException(status_code=401, detail="认证失败")
return user

@app.get("/me", response_model=UserResponse)
async def read_current_user(user: User = Depends(get_current_user)):
return user


五、FastAPI高级特性

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

app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)

@app.middleware("http")
async def add_timing_header(request, call_next):
start = time.perf_counter()
response = await call_next(request)
elapsed = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{elapsed:.4f}"
return response

# 后台任务
from fastapi import BackgroundTasks

async def send_notification(email: str, message: str):
await email_service.send(email, message)

@app.post("/orders")
async def create_order(order: OrderCreate, background_tasks: BackgroundTasks):
result = await process_order(order)
background_tasks.add_task(send_notification, order.email, "订单已创建")
return result

# WebSocket
from fastapi import WebSocket

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"收到: {data}")
except Exception:
pass

# 自动生成API文档
# 访问 /docs 查看Swagger UI
# 访问 /redoc 查看ReDoc


六、Django基础

# models.py
from django.db import models

class User(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField(unique=True)
age = models.IntegerField(null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)

class Meta:
ordering = ['-created_at']

def __str__(self):
return self.name

# views.py
from django.http import JsonResponse
from django.views import View
from django.views.decorators.http import require_http_methods

class UserView(View):
def get(self, request, user_id=None):
if user_id:
user = User.objects.get(pk=user_id)
return JsonResponse({'id': user.id, 'name': user.name})
users = User.objects.all()[:20]
return JsonResponse({'users': list(users.values())})

def post(self, request):
import json
data = json.loads(request.body)
user = User.objects.create(**data)
return JsonResponse({'id': user.id}, status=201)

# Django REST Framework
from rest_framework import serializers, viewsets

class UserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = ['id', 'name', 'email', 'age', 'created_at']

class UserViewSet(viewsets.ModelViewSet):
queryset = User.objects.all()
serializer_class = UserSerializer


七、框架对比

特性 Flask FastAPI Django
类型 微框架 异步框架 全栈框架
性能 中等 高 中等
异步支持 有限 原生 3.1+支持
类型检查 无 Pydantic Serializer
API文档 需扩展 自动生成 需DRF
ORM 需SQLAlchemy 需第三方 内置
管理后台 需扩展 需第三方 内置
学习曲线 低 中 高
适用场景 小型API/原型 高性能API 大型Web应用

选择建议：
- 快速原型/小型服务 -> Flask
- 高性能异步API -> FastAPI
- 大型全功能Web应用 -> Django
- 需要自动API文档 -> FastAPI
- 需要管理后台 -> Django


八、通用Web开发模式

8.1 请求验证

from functools import wraps
from flask import request, jsonify

def validate_json(*required_fields):
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
data = request.get_json()
if not data:
return jsonify({'error': '请求体必须是JSON'}), 400
missing = [field for field in required_fields if field not in data]
if missing:
return jsonify({'error': f'缺少字段: {missing}'}), 400
return f(*args, **kwargs)
return wrapper
return decorator

@app.route('/users', methods=['POST'])
@validate_json('name', 'email')
def create_user():
data = request.get_json()
# ...

8.2 分页

class Paginator:
def __init__(self, query, page=1, per_page=20):
self.query = query
self.page = max(1, page)
self.per_page = min(100, max(1, per_page))
self.total = len(query)

@property
def items(self):
start = (self.page - 1) * self.per_page
end = start + self.per_page
return self.query[start:end]

@property
def pages(self):
return (self.total + self.per_page - 1) // self.per_page

def to_dict(self):
return {
'items': self.items,
'page': self.page,
'per_page': self.per_page,
'total': self.total,
'pages': self.pages,
'has_next': self.page < self.pages,
'has_prev': self.page > 1,
}

8.3 认证中间件

import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"

def create_token(user_id, expires_hours=24):
payload = {
'user_id': user_id,
'exp': datetime.utcnow() + timedelta(hours=expires_hours),
'iat': datetime.utcnow(),
}
return jwt.encode(payload, SECRET_KEY, algorithm='HS256')

def verify_token(token):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
return payload['user_id']
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None


九、部署

# Gunicorn（生产WSGI服务器）
# gunicorn -w 4 -b 0.0.0.0:8000 app:app

# Uvicorn（ASGI服务器，用于FastAPI）
# uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

# Docker部署
"""
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
"""


十、总结

Web开发要点：
1. 理解HTTP协议和WSGI/ASGI规范
2. 选择适合项目规模的框架
3. 使用Pydantic或类似工具验证输入数据
4. 实现统一的错误处理和响应格式
5. 注意安全：CORS、CSRF、SQL注入、XSS防护
6. 使用中间件处理横切关注点（日志、认证、限流）
7. 生产环境使用专业的WSGI/ASGI服务器