Python Web开发基础与框架对比 Python Web开发基础与框架对比一、WSGI协议WSGIWeb Server Gateway Interface是Python Web应用与服务器之间的标准接口def simple_app(environ, start_response):最简单的WSGI应用status 200 OKheaders [(Content-Type, text/plain; charsetutf-8)]start_response(status, headers)return [bHello, World!]# environ包含请求信息# PATH_INFO: 请求路径# REQUEST_METHOD: HTTP方法# QUERY_STRING: 查询参数# wsgi.input: 请求体# 使用内置服务器运行from wsgiref.simple_server import make_serverserver make_server(localhost, 8000, simple_app)server.serve_forever()二、Flask基础from flask import Flask, request, jsonify, abortapp 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字段}), 400user save_user(data)return jsonify(user), 201# 错误处理app.errorhandler(404)def not_found(error):return jsonify({error: 资源不存在}), 404app.errorhandler(500)def internal_error(error):return jsonify({error: 服务器内部错误}), 500# 中间件before/after requestapp.before_requestdef log_request():app.logger.info(f{request.method} {request.path})app.after_requestdef add_cors_headers(response):response.headers[Access-Control-Allow-Origin] *return response三、Flask蓝图与项目组织# blueprints/users.pyfrom flask import Blueprint, request, jsonifyusers_bp Blueprint(users, __name__, url_prefix/api/users)users_bp.route(/, methods[GET])def list_users():page request.args.get(page, 1, typeint)per_page request.args.get(per_page, 20, typeint)users User.query.paginate(pagepage, per_pageper_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.pyfrom flask import Flaskfrom blueprints.users import users_bpfrom blueprints.orders import orders_bpdef create_app(config_namedefault):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, Pathfrom pydantic import BaseModel, EmailStrfrom typing import Optionalapp FastAPI(titleMy API, version1.0.0)# 请求/响应模型class UserCreate(BaseModel):name: stremail: EmailStrage: Optional[int] Noneclass UserResponse(BaseModel):id: intname: stremail: strage: Optional[int]class Config:from_attributes True# 路由app.get(/users/{user_id}, response_modelUserResponse)async def get_user(user_id: int Path(..., gt0)):user await db.get_user(user_id)if not user:raise HTTPException(status_code404, detail用户不存在)return userapp.get(/users, response_modellist[UserResponse])async def list_users(skip: int Query(0, ge0),limit: int Query(20, ge1, le100),name: Optional[str] None,):return await db.get_users(skipskip, limitlimit, namename)app.post(/users, response_modelUserResponse, status_code201)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_code401, detail认证失败)return userapp.get(/me, response_modelUserResponse)async def read_current_user(user: User Depends(get_current_user)):return user五、FastAPI高级特性# 中间件from fastapi.middleware.cors import CORSMiddlewareimport timeapp.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() - startresponse.headers[X-Process-Time] f{elapsed:.4f}return response# 后台任务from fastapi import BackgroundTasksasync 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# WebSocketfrom fastapi import WebSocketapp.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.pyfrom django.db import modelsclass User(models.Model):name models.CharField(max_length100)email models.EmailField(uniqueTrue)age models.IntegerField(nullTrue, blankTrue)created_at models.DateTimeField(auto_now_addTrue)class Meta:ordering [-created_at]def __str__(self):return self.name# views.pyfrom django.http import JsonResponsefrom django.views import Viewfrom django.views.decorators.http import require_http_methodsclass UserView(View):def get(self, request, user_idNone):if user_id:user User.objects.get(pkuser_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 jsondata json.loads(request.body)user User.objects.create(**data)return JsonResponse({id: user.id}, status201)# Django REST Frameworkfrom rest_framework import serializers, viewsetsclass UserSerializer(serializers.ModelSerializer):class Meta:model Userfields [id, name, email, age, created_at]class UserViewSet(viewsets.ModelViewSet):queryset User.objects.all()serializer_class UserSerializer七、框架对比特性 Flask FastAPI Django类型 微框架 异步框架 全栈框架性能 中等 高 中等异步支持 有限 原生 3.1支持类型检查 无 Pydantic SerializerAPI文档 需扩展 自动生成 需DRFORM 需SQLAlchemy 需第三方 内置管理后台 需扩展 需第三方 内置学习曲线 低 中 高适用场景 小型API/原型 高性能API 大型Web应用选择建议- 快速原型/小型服务 - Flask- 高性能异步API - FastAPI- 大型全功能Web应用 - Django- 需要自动API文档 - FastAPI- 需要管理后台 - Django八、通用Web开发模式8.1 请求验证from functools import wrapsfrom flask import request, jsonifydef validate_json(*required_fields):def decorator(f):wraps(f)def wrapper(*args, **kwargs):data request.get_json()if not data:return jsonify({error: 请求体必须是JSON}), 400missing [field for field in required_fields if field not in data]if missing:return jsonify({error: f缺少字段: {missing}}), 400return f(*args, **kwargs)return wrapperreturn decoratorapp.route(/users, methods[POST])validate_json(name, email)def create_user():data request.get_json()# ...8.2 分页class Paginator:def __init__(self, query, page1, per_page20):self.query queryself.page max(1, page)self.per_page min(100, max(1, per_page))self.total len(query)propertydef items(self):start (self.page - 1) * self.per_pageend start self.per_pagereturn self.query[start:end]propertydef pages(self):return (self.total self.per_page - 1) // self.per_pagedef 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 jwtfrom datetime import datetime, timedeltaSECRET_KEY your-secret-keydef create_token(user_id, expires_hours24):payload {user_id: user_id,exp: datetime.utcnow() timedelta(hoursexpires_hours),iat: datetime.utcnow(),}return jwt.encode(payload, SECRET_KEY, algorithmHS256)def verify_token(token):try:payload jwt.decode(token, SECRET_KEY, algorithms[HS256])return payload[user_id]except jwt.ExpiredSignatureError:return Noneexcept jwt.InvalidTokenError:return None九、部署# Gunicorn生产WSGI服务器# gunicorn -w 4 -b 0.0.0.0:8000 app:app# UvicornASGI服务器用于FastAPI# uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4# Docker部署FROM python:3.11-slimWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtCOPY . .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服务器