Python 编程教程 / 18 - Web 开发
第 18 章：Web 开发
掌握 FastAPI、Flask、Django 等主流 Web 框架的核心概念。
18.1 Python Web 框架对比
| 框架 | 类型 | 特点 | 适用场景 |
|---|---|---|---|
| FastAPI | 异步 | 类型驱动、自动文档、高性能 | API 服务、微服务 |
| Flask | 同步 | 轻量灵活、扩展丰富 | 小型应用、原型 |
| Django | 同步 | 全功能、ORM、Admin | 大型应用、CMS |
18.2 FastAPI
18.2.1 基本示例
from fastapi import FastAPI
app = FastAPI(title="My API", version="1.0.0")
@app.get("/")
async def root():
return {"message": "Hello, World!"}
@app.get("/items/{item_id}")
async def get_item(item_id: int, q: str | None = None):
return {"item_id": item_id, "q": q}
$ pip install fastapi uvicorn
$ uvicorn main:app --reload
# 访问 http://localhost:8000/docs 查看自动文档
18.2.2 请求体与验证
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class Item(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
price: float = Field(..., gt=0)
description: str | None = None
tags: list[str] = []
@app.post("/items/", status_code=201)
async def create_item(item: Item):
return {"id": 1, **item.model_dump()}
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
return {"item_id": item_id, **item.model_dump()}
18.2.3 路由组织
from fastapi import APIRouter
router = APIRouter(prefix="/api/users", tags=["users"])
@router.get("/")
async def list_users():
return [{"id": 1, "name": "Alice"}]
@router.get("/{user_id}")
async def get_user(user_id: int):
return {"id": user_id, "name": "Alice"}
# main.py
app = FastAPI()
app.include_router(router)
18.2.4 依赖注入
from fastapi import Depends, FastAPI, HTTPException, Header
app = FastAPI()
async def get_token_header(authorization: str = Header()):
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid token")
return authorization[7:]
async def get_current_user(token: str = Depends(get_token_header)):
# 验证 token 并返回用户
return {"id": 1, "name": "Alice"}
@app.get("/me")
async def read_me(user: dict = Depends(get_current_user)):
return user
18.2.5 中间件
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import time
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.middleware("http")
async def add_process_time_header(request, call_next):
start = time.perf_counter()
response = await call_next(request)
process_time = time.perf_counter() - start
response.headers["X-Process-Time"] = str(process_time)
return response
18.3 Flask
18.3.1 基本示例
from flask import Flask, jsonify, request
app = Flask(__name__)
@app.route("/")
def index():
return jsonify(message="Hello, World!")
@app.route("/items/<int:item_id>")
def get_item(item_id):
q = request.args.get("q")
return jsonify(item_id=item_id, q=q)
@app.route("/items", methods=["POST"])
def create_item():
data = request.get_json()
return jsonify(data), 201
if __name__ == "__main__":
app.run(debug=True)
$ pip install flask
$ flask --app main run
18.3.2 Blueprint（蓝图）
from flask import Blueprint, jsonify
users_bp = Blueprint("users", __name__, url_prefix="/api/users")
@users_bp.route("/")
def list_users():
return jsonify(users=[{"id": 1, "name": "Alice"}])
# app.py
from flask import Flask
app = Flask(__name__)
app.register_blueprint(users_bp)
18.4 Django
18.4.1 项目结构
mysite/
├── manage.py
├── mysite/
│ ├── __init__.py
│ ├── settings.py
│ ├── urls.py
│ └── wsgi.py
└── users/
├── __init__.py
├── models.py
├── views.py
├── urls.py
└── serializers.py
18.4.2 模型
from django.db import models
class User(models.Model):
name = models.CharField(max_length=100)
email = models.EmailField(unique=True)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ["-created_at"]
def __str__(self):
return self.name
18.4.3 视图
from django.http import JsonResponse
from django.views import View
class UserView(View):
def get(self, request):
users = list(User.objects.values("id", "name", "email"))
return JsonResponse({"users": users})
def post(self, request):
data = json.loads(request.body)
user = User.objects.create(**data)
return JsonResponse({"id": user.id}, status=201)
18.5 REST API 设计
| 方法 | 路径 | 动作 | 状态码 |
|---|---|---|---|
| GET | /items | 获取列表 | 200 |
| GET | /items/{id} | 获取单个 | 200 / 404 |
| POST | /items | 创建 | 201 |
| PUT | /items/{id} | 全量更新 | 200 / 404 |
| PATCH | /items/{id} | 部分更新 | 200 / 404 |
| DELETE | /items/{id} | 删除 | 204 / 404 |
18.6 模板渲染
Jinja2（Flask/FastAPI）
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/")
async def index(request: Request):
return templates.TemplateResponse("index.html", {
"request": request,
"title": "首页",
"items": ["Python", "FastAPI", "Docker"],
})
<!-- templates/index.html -->
<h1>{{ title }}</h1>
<ul>
{% for item in items %}
<li>{{ item }}</li>
{% endfor %}
</ul>
18.7 注意事项
🔴 注意：
- 不要在生产环境使用 Flask/FastAPI 的开发服务器
- 始终验证和过滤用户输入
- 使用
gunicorn或uvicorn部署 - 敏感配置使用环境变量，不要硬编码
💡 提示：
- API 服务首选 FastAPI（自动文档、类型验证）
- 全功能 Web 应用首选 Django
- 轻量应用或微服务用 Flask
- 使用 Pydantic 做数据验证
📌 业务场景：
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
app = FastAPI()
class UserCreate(BaseModel):
name: str
email: str
class UserResponse(BaseModel):
id: int
name: str
email: str
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = db.query(User).filter(User.email == user.email).first()
if db_user:
raise HTTPException(status_code=400, detail="邮箱已注册")
new_user = User(**user.model_dump())
db.add(new_user)
db.commit()
db.refresh(new_user)
return new_user