1. Web 开发基础
1-1. Web 框架概述
Python 主要 Web 框架:
框架 | 特点 | 适用场景 |
---|---|---|
Flask | 轻量级、灵活 | 小型应用、API 服务 |
Django | 全栈、功能丰富 | 大型应用、企业级开发 |
FastAPI | 高性能、异步 | API 服务、微服务 |
Tornado | 异步、高性能 | 实时应用、长连接 |
Sanic | 异步、快速 | 高性能 API 服务 |
1-2. HTTP 基础
python
from http.server import HTTPServer, BaseHTTPRequestHandler
class SimpleHandler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header('Content-type', 'text/html')
self.end_headers()
self.wfile.write(b'Hello, World!')
def run_server():
server = HTTPServer(('localhost', 8000), SimpleHandler)
server.serve_forever()
1-3. WSGI 应用
python
def simple_app(environ, start_response):
status = '200 OK'
headers = [('Content-type', 'text/plain')]
start_response(status, headers)
return [b'Hello, World!']
# 使用 WSGI 服务器运行
from wsgiref.simple_server import make_server
server = make_server('localhost', 8000, simple_app)
server.serve_forever()
2. Flask 框架
2-1. 基本应用
python
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/')
def hello():
return 'Hello, World!'
@app.route('/api/users', methods=['GET'])
def get_users():
return jsonify({'users': ['Alice', 'Bob']})
@app.route('/api/users', methods=['POST'])
def create_user():
data = request.get_json()
return jsonify({'message': 'User created', 'data': data}), 201
if __name__ == '__main__':
app.run(debug=True)
2-2. 路由和视图
python
@app.route('/user/<username>')
def show_user(username):
return f'User: {username}'
@app.route('/post/<int:post_id>')
def show_post(post_id):
return f'Post: {post_id}'
@app.route('/path/<path:subpath>')
def show_subpath(subpath):
return f'Subpath: {subpath}'
2-3. 模板渲染
python
from flask import render_template
@app.route('/hello/<name>')
def hello(name):
return render_template('hello.html', name=name)
# templates/hello.html
"""
<!DOCTYPE html>
<html>
<head>
<title>Hello</title>
</head>
<body>
<h1>Hello, {{ name }}!</h1>
</body>
</html>
"""
3. Django 框架
3-1. 项目结构
myproject/
manage.py
myproject/
__init__.py
settings.py
urls.py
wsgi.py
myapp/
__init__.py
models.py
views.py
urls.py
templates/
static/
3-2. 模型定义
python
from django.db import models
class User(models.Model):
username = models.CharField(max_length=100)
email = models.EmailField(unique=True)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return self.username
class Post(models.Model):
title = models.CharField(max_length=200)
content = models.TextField()
author = models.ForeignKey(User, on_delete=models.CASCADE)
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
ordering = ['-created_at']
3-3. 视图和URL配置
python
# views.py
from django.shortcuts import render, get_object_or_404
from django.http import JsonResponse
from .models import Post
def post_list(request):
posts = Post.objects.all()
return render(request, 'blog/post_list.html', {'posts': posts})
def post_detail(request, pk):
post = get_object_or_404(Post, pk=pk)
return render(request, 'blog/post_detail.html', {'post': post})
# urls.py
from django.urls import path
from . import views
urlpatterns = [
path('', views.post_list, name='post_list'),
path('post/<int:pk>/', views.post_detail, name='post_detail'),
]
4. FastAPI 框架
4-1. 基本应用
python
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.get("/")
async def read_root():
return {"Hello": "World"}
@app.post("/items/")
async def create_item(item: Item):
return item
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
4-2. 依赖注入
python
from fastapi import Depends, HTTPException
from typing import List
async def get_db():
db = "fake_db"
try:
yield db
finally:
db = None
@app.get("/items/")
async def read_items(db: str = Depends(get_db)):
return {"db": db}
4-3. 中间件
python
from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
5. Web 开发最佳实践
5-1. RESTful API 设计
python
# 使用 Flask-RESTful
from flask_restful import Api, Resource
api = Api(app)
class UserResource(Resource):
def get(self, user_id):
return {'user': 'get user'}
def put(self, user_id):
return {'user': 'update user'}
def delete(self, user_id):
return {'user': 'delete user'}
api.add_resource(UserResource, '/users/<int:user_id>')
5-2. 认证和授权
python
# 使用 Flask-JWT
from flask_jwt_extended import JWTManager, jwt_required, create_access_token
app.config['JWT_SECRET_KEY'] = 'super-secret'
jwt = JWTManager(app)
@app.route('/login', methods=['POST'])
def login():
username = request.json.get('username')
password = request.json.get('password')
if username == 'admin' and password == 'password':
access_token = create_access_token(identity=username)
return jsonify(access_token=access_token)
return jsonify({"msg": "Bad username or password"}), 401
@app.route('/protected', methods=['GET'])
@jwt_required()
def protected():
return jsonify(logged_in_as=current_user), 200
5-3. 错误处理
python
# Flask 错误处理
@app.errorhandler(404)
def not_found_error(error):
return jsonify({'error': 'Not found'}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({'error': 'Internal server error'}), 500
# FastAPI 错误处理
from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}
6. 性能优化
6-1. 缓存
python
# 使用 Flask-Caching
from flask_caching import Cache
cache = Cache(app, config={'CACHE_TYPE': 'simple'})
@app.route('/expensive-operation')
@cache.cached(timeout=300)
def expensive_operation():
# 耗时操作
return result
6-2. 异步处理
python
# 使用 Celery
from celery import Celery
celery = Celery('tasks', broker='redis://localhost:6379/0')
@celery.task
def process_data(data):
# 异步处理数据
return result
@app.route('/process', methods=['POST'])
def process():
data = request.get_json()
task = process_data.delay(data)
return jsonify({'task_id': task.id}), 202
6-3. 数据库优化
python
# Django 查询优化
from django.db.models import Prefetch
# 使用 select_related 和 prefetch_related
posts = Post.objects.select_related('author').prefetch_related('comments')
# 使用 annotate 进行聚合
from django.db.models import Count
posts = Post.objects.annotate(comment_count=Count('comments'))
7. 测试和部署
7-1. 单元测试
python
# Flask 测试
import unittest
from flask import url_for
class TestApp(unittest.TestCase):
def setUp(self):
self.app = app.test_client()
def test_home_page(self):
response = self.app.get('/')
self.assertEqual(response.status_code, 200)
def test_create_user(self):
response = self.app.post('/api/users', json={'name': 'Alice'})
self.assertEqual(response.status_code, 201)
7-2. 部署配置
python
# Gunicorn 配置
# gunicorn_config.py
bind = "0.0.0.0:8000"
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
timeout = 120
7-3. 监控和日志
python
# 日志配置
import logging
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler('app.log', maxBytes=10000, backupCount=3)
handler.setLevel(logging.INFO)
app.logger.addHandler(handler)
@app.route('/')
def index():
app.logger.info('访问首页')
return 'Hello, World!'
TIP
Web 开发最佳实践:
- 遵循 RESTful API 设计原则
- 实现适当的错误处理和日志记录
- 使用缓存优化性能
- 实现安全的认证和授权
- 编写单元测试和集成测试
- 使用适当的部署策略
- 监控应用性能和错误