第五十一篇：Web 开发基础——HTTP 与 WSGI
前言
Python 在 Web 开发领域有着举足轻重的地位，从最早的 CGI、Django，到 Flask、FastAPI，框架层出不穷。但在深入框架之前，我们需要理解两个核心概念：HTTP 协议和 WSGI 接口规范。本文不急于上手框架，而是先从底层原理出发，让你真正理解 Web 应用是如何工作的。
HTTP 协议基础
什么是 HTTP
┌───────────┐ ┌───────────┐
│ │ HTTP Request │ │
│ Client │ ────────────────────────── → │ Server │
│ (浏览器) │ GET /index.html HTTP/1.1 │ │
│ │ Host: www.example.com │ │
│ │ User-Agent: ... │ │
│ │ ←────────────────────────── │ │
│ │ HTTP Response 200 OK │ │
│ │ Content-Type: text/html │ │
│ │ <html>...</html> │ │
└───────────┘ └───────────┘
HTTP（HyperText Transfer Protocol）是互联网上应用最广泛的应用层协议，定义了客户端和服务端如何通信。
请求方法
| 方法 | 说明 | 安全性 | 幂等性 | 典型场景 |
|---|---|---|---|---|
| GET | 获取资源 | ✓ 安全 | ✓ 幂等 | 查询数据 |
| POST | 提交数据 | ✗ 不安全 | ✗ | 创建资源 |
| PUT | 完整更新 | ✗ 不安全 | ✓ | 替换资源 |
| PATCH | 部分更新 | ✗ 不安全 | ✗ | 修改字段 |
| DELETE | 删除资源 | ✗ 不安全 | ✓ | 删除数据 |
| HEAD | 获取头部 | ✓ 安全 | ✓ | 检查存在性 |
| OPTIONS | 跨域预检 | ✓ 安全 | ✓ | CORS 检查 |
状态码速查
1xx ─── 信息性响应
100 Continue 继续发送请求体
2xx ─── 成功
200 OK 请求成功
201 Created 资源已创建
204 No Content 成功但无内容
3xx ─── 重定向
301 Moved Permanently 永久重定向
302 Found 临时重定向
304 Not Modified 缓存未修改
307 Temporary Redirect 临时重定向（保留方法）
4xx ─── 客户端错误
400 Bad Request 请求语法错误
401 Unauthorized 未认证
403 Forbidden 禁止访问
404 Not Found 资源不存在
405 Method Not Allowed 方法不允许
422 Unprocessable Entity 无法处理实体
5xx ─── 服务器错误
500 Internal Server Error 服务器内部错误
502 Bad Gateway 网关错误
503 Service Unavailable 服务不可用
504 Gateway Timeout 网关超时
请求头与响应头
"""
常用请求头：
- Accept: 客户端接受的 MIME 类型
- Accept-Language: 首选语言
- Authorization: 认证凭据 (Bearer token)
- Cache-Control: 缓存控制
- Content-Type: 请求体格式 (application/json)
- Cookie: 存储在客户端的键值对
- User-Agent: 客户端标识
- Referer: 来源页面 URL
常用响应头：
- Location: 重定向目标 URL
- Set-Cookie: 设置 Cookie
- Content-Type: 响应体格式
- Content-Length: 响应体长度
- Cache-Control: 缓存策略
- Access-Control-Allow-Origin: CORS 允许的来源
- X-Powered-By: 服务器技术栈
"""
HTTP 请求与响应详解
手写一个 HTTP 请求
import http.client
# ========== GET 请求 ==========
conn = http.client.HTTPSConnection("httpbin.org")
conn.request('GET', '/get?name=test&age=25')
response = conn.getresponse()
print(f"状态码: {response.status} {response.reason}")
print(f"Content-Type: {response.getheader('Content-Type')}")
print(f"Body:\n{response.read().decode()}")
conn.close()
# ========== POST JSON 请求 ==========
data = {"username": "admin", "password": "secret"}
import json
conn = http.client.HTTPSConnection("httpbin.org")
conn.request(
'POST',
'/post',
body=json.dumps(data),
headers={'Content-Type': 'application/json'}
)
response = conn.getresponse()
print(f"\nPOST 状态码: {response.status}")
print(f"Body:\n{response.read().decode()}")
conn.close()
# ========== PUT / DELETE 请求 ==========
conn = http.client.HTTPSConnection("httpbin.org")
# PUT - 更新资源
conn.request(
'PUT',
'/put',
body=json.dumps({"id": 1, "title": "Updated Post"}),
headers={'Content-Type': 'application/json'}
)
resp = conn.getresponse()
print(f"\nPUT 状态码: {resp.status}")
# DELETE - 删除资源
conn.request('DELETE', '/delete/123')
resp = conn.getresponse()
print(f"DELETE 状态码: {resp.status}")
conn.close()
curl 快速对照表
# 以下 HTTP 操作可以用 curl 验证
curl https://httpbin.org/get # GET
curl -X POST -H "Content-Type: application/json" -d '{"key":"val"}' https://httpbin.org/post # POST JSON
curl -X PUT -H "Content-Type: application/json" -d '{"key":"updated"}' https://httpbin.org/put # PUT
curl -X DELETE https://httpbin.org/delete/123 # DELETE
curl -I https://www.example.com # 只看 Header
curl -v https://httpbin.org/get # 详细输出（调试用）
curl -L https://httpbin.org/redirect/5 # 跟随重定向
HTML 简介
HTML（HyperText Markup Language）是网页的基础标记语言。虽然 Python Web 教程的重点在后端，但理解 HTML 对开发全栈应用很有帮助。
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>我的第一个网页</title>
<style>
body { font-family: Arial, sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }
.highlight { color: blue; font-weight: bold; }
</style>
</head>
<body>
<!-- 标题 -->
<h1>欢迎学习 Python Web 开发</h1>
<!-- 段落 -->
<p>这是一个示例段落，使用 <span class="highlight">HTML</span> 编写。</p>
<!-- 表单 -->
<form action="/submit" method="POST">
<label>姓名：</label>
<input type="text" name="username" required />
<label>邮箱：</label>
<input type="email" name="email" required />
<label>留言：</label><br/>
<textarea name="message" rows="4"></textarea>
<button type="submit">提交</button>
</form>
<!-- 列表 -->
<ul>
<li>第一章：Hello World</li>
<li>第二章：变量与数据类型</li>
</ul>
<!-- 链接 -->
<a href="https://example.com">跳转到外部网站</a>
</body>
</html>
WSGI——Python Web 的统一接口
为什么需要 WSGI
┌──────────────────────────────────────────────┐
│ Python Web 世界 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌────────┐ │
│ │ Django │ │ Flask │ │FastAPI │ │
│ └────┬─────┘ └────┬─────┘ └───┬────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────┐ │
│ │ WSGI 接口规范 │ │
│ │ │ │
│ └─────────────────────────────────────────┘ │
│ ▲ ▲ ▲ │
│ ┌────┴─────┐ ┌────┴─────┐ ┌───┴────┐ │
│ │ Gunicorn │ │ uWSGI │ │Apache │ │
│ │ Waitress │ │ │ │mod_wsgi│ │
│ └──────────┘ └──────────┘ └────────┘ │
│ 服务器端 │
└──────────────────────────────────────────────┘
WSGI（Web Server Gateway Interface）是 PEP 3333 定义的 Python Web 应用与服务器之间的标准接口。它让框架和服务器可以解耦搭配：
- 任何 WSGI 框架（Django/Flask/FastAPI）可以在任何 WSGI 服务器（Gunicorn/uWSGI/Waitress）上运行
- 你不需要关心服务器的实现细节
WSGI 接口定义
WSGI 非常简单，只有两个角色、三个函数：
服务器调用应用程序：app(environ, start_response)
应用程序回调：start_response(status, headers)
从零手写 WSGI 应用
最简单的 Hello World
def simple_app(environ, start_response):
"""最基本的 WSGI 应用"""
# environ 包含所有 HTTP 请求信息
method = environ.get('REQUEST_METHOD')
path = environ.get('PATH_INFO')
query_string = environ.get('QUERY_STRING', '')
# 根据路径返回不同内容
if path == '/' and method == 'GET':
body = b'Hello, World!'
status = '200 OK'
headers = [('Content-Type', 'text/plain; charset=utf-8')]
elif path == '/hello' and query_string.startswith('name='):
name = query_string.split('=')[1]
body = f'Hello, {name}!'.encode('utf-8')
status = '200 OK'
headers = [('Content-Type', 'text/plain; charset=utf-8')]
elif path not in ('/', '/hello'):
body = b'Not Found'
status = '404 Not Found'
headers = [('Content-Type', 'text/plain; charset=utf-8')]
else:
body = b'Method Not Allowed'
status = '405 Method Not Allowed'
headers = [('Content-Type', 'text/plain; charset=utf-8')]
# 告诉客户端响应头和状态码
start_response(status, headers)
# 返回响应体（必须是字节数组列表）
return [body]
# 测试：使用内置 wsgiref 服务器
from wsgiref.simple_server import make_server
if __name__ == '__main__':
httpd = make_server('127.0.0.1', 8000, simple_app)
print("✓ 服务器运行在 http://127.0.0.1:8000")
print(" 访问: http://127.0.0.1:8000/")
print(" 访问: http://127.0.0.1:8000/hello?name=Python")
print(" 按 Ctrl+C 停止")
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\n服务器已停止")
支持多方法的 WSGI
def multi_method_app(environ, start_response):
"""支持多种 HTTP 方法的 WSGI 应用"""
method = environ.get('REQUEST_METHOD')
path = environ.get('PATH_INFO')
# 路由映射
routes = {
'/api/items': {
'GET': lambda: handle_get_items(),
'POST': lambda: handle_create_item(environ),
},
'/api/items/<id>': {
'GET': lambda: handle_get_item(environ),
'PUT': lambda: handle_update_item(environ),
'DELETE': lambda: handle_delete_item(environ),
},
}
# 简单的路由匹配
if path == '/api/items':
handler_map = routes['/api/items']
elif path.startswith('/api/items/'):
handler_map = routes['/api/items/<id>']
else:
_send_not_found(start_response)
return []
handler = handler_map.get(method)
if not handler:
_send_error(start_response, '405 Method Not Allowed')
return []
response = handler()
return _create_response(response['status'], response['headers'], response['body'])
def handle_get_items():
items = [
{'id': 1, 'name': '键盘', 'price': 299},
{'id': 2, 'name': '鼠标', 'price': 89},
{'id': 3, 'name': '显示器', 'price': 1999},
]
import json
body = json.dumps(items, ensure_ascii=False).encode('utf-8')
return {'status': '200 OK', 'headers': [('Content-Type', 'application/json')], 'body': body}
def handle_create_item(environ):
length = int(environ.get('CONTENT_LENGTH', 0))
body_raw = environ.get('wsgi.input').read(length)
import json
item = json.loads(body_raw.decode())
# 模拟数据库插入
response_body = {'success': True, 'created': item}
import json
body = json.dumps(response_body, ensure_ascii=False).encode('utf-8')
return {'status': '201 Created', 'headers': [('Content-Type', 'application/json')], 'body': body}
def handle_get_item(environ):
item_id = environ['PATH_INFO'].split('/')[-1]
# 模拟查询
items = {
'1': {'id': 1, 'name': '键盘', 'price': 299},
'2': {'id': 2, 'name': '鼠标', 'price': 89},
}
item = items.get(item_id)
if item:
import json
body = json.dumps(item, ensure_ascii=False).encode('utf-8')
return {'status': '200 OK', 'headers': [('Content-Type', 'application/json')], 'body': body}
return {'status': '404 Not Found', 'headers': [], 'body': b'{"error": "Item not found"}'}
def handle_update_item(environ):
length = int(environ.get('CONTENT_LENGTH', 0))
body_raw = environ.get('wsgi.input').read(length)
item = eval(body_raw.decode()) # 演示用，生产请用 json.loads
body = f'{{"success": true, "updated": {item}}}'.encode('utf-8')
return {'status': '200 OK', 'headers': [('Content-Type', 'application/json')], 'body': body}
def handle_delete_item(environ):
item_id = environ['PATH_INFO'].split('/')[-1]
return {'status': '204 No Content', 'headers': [], 'body': b''}
def _send_not_found(start_response):
start_response('404 Not Found', [('Content-Type', 'text/plain')])
return []
def _send_error(start_response, message):
start_response(message, [('Content-Type', 'text/plain')])
return [message.encode()]
def _create_response(status, headers, body):
start_response_func = None
def custom_start_response(status, response_headers):
nonlocal start_response_func
start_response_func = response_headers
return lambda x: None
custom_start_response(status, headers)
return [body]
# 更简洁的实现（推荐写法）
def json_api_app(environ, start_response):
"""JSON API WSGI 应用"""
method = environ.get('REQUEST_METHOD')
path = environ.get('PATH_INFO')
if path == '/api/time':
from datetime import datetime
data = {'time': str(datetime.now()), 'method': method}
_json_response(start_response, 200, data)
return []
elif path == '/api/echo' and method == 'POST':
length = int(environ.get('CONTENT_LENGTH', 0))
raw = environ['wsgi.input'].read(length)
_json_response(start_response, 200, {'echo': raw.decode('utf-8')})
return []
else:
_json_response(start_response, 404, {'error': 'Not Found'})
return []
def _json_response(start_response, status_code, data):
"""辅助函数：发送 JSON 响应"""
import json
status_texts = {
200: 'OK', 201: 'Created', 204: 'No Content',
400: 'Bad Request', 404: 'Not Found', 500: 'Internal Server Error'
}
body = json.dumps(data, ensure_ascii=False).encode('utf-8')
headers = [
('Content-Type', 'application/json; charset=utf-8'),
('Content-Length', str(len(body))),
]
start_response(f'{status_code} {status_texts.get(status_code, "Unknown")}', headers)
return [body]
# 启动 JSON API 服务器
if __name__ == '__main__':
from wsgiref.simple_server import make_server
server = make_server('127.0.0.1', 8080, json_api_app)
print("✓ JSON API 运行在 http://127.0.0.1:8080")
print(" GET http://127.0.0.1:8080/api/time")
print(" POST http://127.0.0.1:8080/api/echo")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n已停止")
装饰器版 WSGI 路由
class Router:
"""简易 WSGI 路由器"""
def __init__(self):
self.routes = {}
def route(self, path, methods='GET'):
"""装饰器：注册路由"""
def decorator(func):
if isinstance(methods, str):
methods = [methods]
for method in methods:
key = (method, path)
self.routes[key] = func
return func
return decorator
def __call__(self, environ, start_response):
method = environ.get('REQUEST_METHOD')
path = environ.get('PATH_INFO', '/')
# 精确匹配
handler = self.routes.get((method, path))
# 参数路由匹配（如 /users/<id>）
if not handler:
for (m, p), h in self.routes.items():
if m != method:
continue
params = self._match_pattern(p, path)
if params:
# 将参数注入 environ
environ['router.params'] = params
handler = h
break
if handler:
response = handler(environ)
status = response.get('status', '200 OK')
headers = response.get('headers', [])
body = response.get('body', b'')
start_response(status, headers)
return [body] if isinstance(body, bytes) else [b'']
else:
start_response('404 Not Found', [('Content-Type', 'text/plain')])
return [b'Not Found']
def _match_pattern(self, pattern, path):
"""匹配 /users/<id> 类型的模式"""
p_parts = pattern.strip('/').split('/')
c_parts = path.strip('/').split('/')
if len(p_parts) != len(c_parts):
return None
params = {}
for i, (pp, cp) in enumerate(zip(p_parts, c_parts)):
if pp.startswith('<') and pp.endswith('>'):
params[pp[1:-1]] = cp
elif pp != cp:
return None
return params
# 使用装饰器构建应用
def app_factory():
router = Router()
@router.route('/', 'GET')
def home(environ):
return {
'status': '200 OK',
'headers': [('Content-Type', 'text/html; charset=utf-8')],
'body': b'<h1>Hello WSGI!</h1>'
}
@router.route('/api/users', 'GET')
def list_users(environ):
import json
users = [
{'id': 1, 'name': '张三', 'email': 'zhangsan@example.com'},
{'id': 2, 'name': '李四', 'email': 'lisi@example.com'},
]
body = json.dumps(users, ensure_ascii=False).encode('utf-8')
return {
'status': '200 OK',
'headers': [('Content-Type', 'application/json')],
'body': body
}
@router.route('/api/users', 'POST')
def create_user(environ):
import json
length = int(environ.get('CONTENT_LENGTH', 0))
data = json.loads(environ['wsgi.input'].read(length).decode())
body = json.dumps({'created': data}).encode('utf-8')
return {
'status': '201 Created',
'headers': [('Content-Type', 'application/json')],
'body': body
}
@router.route('/api/users/<id>', 'GET')
def get_user(environ):
import json
user_id = environ.get('router.params', {}).get('id')
users = {
'1': {'id': 1, 'name': '张三'},
'2': {'id': 2, 'name': '李四'},
}
user = users.get(user_id, {})
body = json.dumps(user).encode('utf-8')
status = '200 OK' if user else '404 Not Found'
return {
'status': status,
'headers': [('Content-Type', 'application/json')],
'body': body
}
return router
# 启动路由器应用
application = app_factory()
if __name__ == '__main__':
from wsgiref.simple_server import make_server
server = make_server('127.0.0.1', 8000, application)
print("✓ WSGI 路由器运行在 http://127.0.0.1:8000")
print(" GET / -> Hello WSGI")
print(" GET /api/users -> 用户列表")
print(" POST /api/users -> 创建用户")
print(" GET /api/users/<id> -> 用户详情")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n服务器已停止")
Cookie 与 Session
def cookie_app(environ, start_response):
"""Cookie 和 Session 示例"""
import base64, hashlib, time
cookies_header = environ.get('HTTP_COOKIE', '')
cookie_dict = _parse_cookies(cookies_header)
path = environ.get('PATH_INFO')
method = environ.get('REQUEST_METHOD')
# === 设置 Cookie ===
if path == '/set-cookie' and method == 'GET':
value = base64.b64encode(b'hello_from_python').decode()
# SHA256 签名防篡改
signature = hashlib.sha256(value.encode()).hexdigest()[:16]
body = f'Set cookie: python_session={value}.{signature}'.encode('utf-8')
headers = [
('Content-Type', 'text/plain; charset=utf-8'),
('Set-Cookie', f'python_session={value}; HttpOnly; Path=/; Max-Age=3600'),
('Set-Cookie', f'token={signature}; Path=/; Max-Age=3600; Secure'),
]
start_response('200 OK', headers)
return [body]
# === 读取 Cookie ===
elif path == '/get-cookie' and method == 'GET':
session_data = cookie_dict.get('python_session', 'None')
token_data = cookie_dict.get('token', 'None')
body = f'Session: {session_data}\nToken: {token_data}'.encode('utf-8')
start_response('200 OK', [('Content-Type', 'text/plain; charset=utf-8')])
return [body]
# === 清除 Cookie ===
elif path == '/clear-cookie' and method == 'GET':
start_response('200 OK', [
('Set-Cookie', 'python_session=; Path=/; Max-Age=0'),
('Set-Cookie', 'token=; Path=/; Max-Age=0'),
('Content-Type', 'text/plain'),
])
return [b'Cookie 已清除']
else:
start_response('404 Not Found', [('Content-Type', 'text/plain')])
return [b'Not Found']
def _parse_cookies(cookie_string):
"""解析 Cookie 字符串为字典"""
if not cookie_string:
return {}
cookies = {}
for part in cookie_string.split(';'):
part = part.strip()
if '=' in part:
key, value = part.split('=', 1)
cookies[key.strip()] = value.strip()
return cookies
if __name__ == '__main__':
from wsgiref.simple_server import make_server
server = make_server('127.0.0.1', 8090, cookie_app)
print("✓ Cookie 示例运行在 http://127.0.0.1:8090")
print(" GET /set-cookie -> 设置 Cookie")
print(" GET /get-cookie -> 读取 Cookie")
print(" GET /clear-cookie -> 清除 Cookie")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n已停止")
中间件链
WSGI 的强大之处在于可以叠加多个中间件，每个中间件拦截请求并修改它：
def logging_middleware(app):
"""日志记录中间件"""
import time
from datetime import datetime
def middleware(environ, start_response):
start_time = time.time()
method = environ.get('REQUEST_METHOD')
path = environ.get('PATH_INFO', '/')
print(f"[{datetime.now().strftime('%H:%M:%S')}] {method} {path}")
# 拦截响应
original_start_response = start_response
response_started = {'started': False, 'status': '', 'headers': []}
def custom_start_response(status, headers):
response_started['started'] = True
response_started['status'] = status
response_started['headers'] = headers
return lambda x: None
try:
result = app(environ, custom_start_response)
duration = time.time() - start_time
status_code = response_started['status'].split()[0]
print(f" → {status_code} ({duration:.3f}s)")
return result
except Exception as e:
duration = time.time() - start_time
print(f" → ERROR ({duration:.3f}s): {e}")
raise
return middleware
def cors_middleware(app):
"""CORS（跨域资源共享）中间件"""
def middleware(environ, start_response):
if environ.get('REQUEST_METHOD') == 'OPTIONS':
start_response('204 No CORS Preflight', [
('Access-Control-Allow-Origin', '*'),
('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS'),
('Access-Control-Allow-Headers', 'Content-Type, Authorization'),
('Access-Control-Max-Age', '86400'),
])
return [b'']
# 拦截正常响应，添加 CORS 头
def capture_start_response(status, headers):
new_headers = list(headers) + [
('Access-Control-Allow-Origin', '*'),
]
return start_response(status, new_headers)
return app(environ, capture_start_response)
return middleware
def cache_middleware(app, max_age=300):
"""简单缓存中间件"""
from functools import lru_cache
import hashlib
import json
cache = {}
def middleware(environ, start_response):
if environ.get('REQUEST_METHOD') != 'GET':
return app(environ, start_response)
path = environ.get('PATH_INFO', '')
cache_key = hashlib.md5(path.encode()).hexdigest()
# 查找缓存
cached = cache.get(cache_key)
if cached and time.time() - cached['timestamp'] < max_age:
start_response(cached['status'], cached['headers'])
return cached['body']
# 执行请求并缓存结果
def capture_start_response(status, headers):
response_started['status'] = status
response_started['headers'] = list(headers)
return lambda x: None
response_started = {'status': '', 'headers': []}
result = app(environ, capture_start_response)
# 仅缓存 2xx 响应
if response_started['status'].startswith('20'):
cache[cache_key] = {
'body': result,
'status': response_started['status'],
'headers': response_started['headers'],
'timestamp': time.time(),
}
return result
import time
return middleware
# 组合多个中间件
application = logging_middleware(cors_middleware(cache_middleware(simple_app)))
if __name__ == '__main__':
from wsgiref.simple_server import make_server
server = make_server('127.0.0.1', 8100, application)
print("✓ 多层中间件 WSGI 运行在 http://127.0.0.1:8100")
print(" 日志中间件 + CORS 中间件 + 缓存中间件")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n已停止")
实战：完整的迷你 Web 框架
"""
mini-web.py — 一个极小的 WSGI Web 框架
包含：路由、请求对象、响应对象、模板渲染、Session
"""
import json
import time
import hashlib
class MiniRequest:
"""封装 HTTP 请求"""
def __init__(self, environ):
self.environ = environ
self.method = environ.get('REQUEST_METHOD')
self.path = environ.get('PATH_INFO', '/')
self.query = environ.get('QUERY_STRING', '')
self.headers = dict(h.split(': ', 1) for h in
filter(lambda x: ': ' in x, environ.get('HTTP_', '').split('\n')))
self.params = self._parse_query()
def _parse_query(self):
"""解析查询参数"""
if not self.query:
return {}
params = {}
for part in self.query.split('&'):
if '=' in part:
k, v = part.split('=', 1)
params[k] = v.replace('+', ' ')
return params
@property
def form_data(self):
"""解析 POST 表单数据"""
length = int(self.environ.get('CONTENT_LENGTH', 0))
if length == 0:
return {}
raw = self.environ['wsgi.input'].read(length).decode('utf-8')
params = {}
for part in raw.split('&'):
if '=' in part:
k, v = part.split('=', 1)
params[k] = v.replace('+', ' ')
return params
@property
def json_data(self):
"""解析 JSON 请求体"""
length = int(self.environ.get('CONTENT_LENGTH', 0))
if length == 0:
return None
raw = self.environ['wsgi.input'].read(length).decode('utf-8')
return json.loads(raw)
class MiniResponse:
"""封装 HTTP 响应"""
def __init__(self, status='200 OK', headers=None, body=b'', content_type='text/plain; charset=utf-8'):
self.status = status
self.headers = headers or []
self.body = body
self.content_type = content_type
@classmethod
def text(cls, text, status='200 OK'):
return cls(status=status, body=text.encode('utf-8'), content_type='text/plain; charset=utf-8')
@classmethod
def html(cls, html, status='200 OK'):
return cls(status=status, body=html.encode('utf-8'), content_type='text/html; charset=utf-8')
@classmethod
def json(cls, data, status='200 OK'):
body = json.dumps(data, ensure_ascii=False).encode('utf-8')
return cls(status=status, body=body, content_type='application/json; charset=utf-8')
@classmethod
def redirect(cls, location, status='302 Found'):
return cls(status=status, headers=[('Location', location)], body=b'Redirecting...')
def _wsgi_handler(self):
def start_response(status, headers):
self.status = status
self.headers = headers
return lambda x: None
# 默认添加 Content-Type
if not any(h[0] == 'Content-Type' for h in self.headers):
self.headers.append(('Content-Type', self.content_type))
start_response(self.status, self.headers)
return [self.body]
class MiniApp:
"""极简 Web 框架"""
def __init__(self):
self.routes = {}
self.before_request_hooks = []
self.after_request_hooks = []
self.error_handlers = {}
def add_route(self, path, methods='GET'):
"""注册路由装饰器"""
def decorator(func):
if isinstance(methods, str):
methods = [methods]
for method in methods:
self.routes[(method.upper(), path)] = func
return func
return decorator
# 便捷路由方法
def get(self, path):
return self.add_route(path, 'GET')
def post(self, path):
return self.add_route(path, 'POST')
def put(self, path):
return self.add_route(path, 'PUT')
def delete(self, path):
return self.add_route(path, 'DELETE')
def error(self, status_code):
"""错误处理器装饰器"""
def decorator(func):
self.error_handlers[status_code] = func
return func
return decorator
def __call__(self, environ, start_response):
request = MiniRequest(environ)
# 执行 before_request 钩子
for hook in self.before_request_hooks:
result = hook(request)
if isinstance(result, MiniResponse):
return result._wsgi_handler()
# 路由匹配
handler = self.routes.get((request.method, request.path))
if not handler:
response = self._handle_404(request)
else:
try:
response = handler(request)
except KeyError as e:
response = self._handle_error(400, request, str(e))
except Exception as e:
response = self._handle_error(500, request, str(e))
# 执行 after_request 钩子
for hook in self.after_request_hooks:
response = hook(request, response)
return response._wsgi_handler()
def _handle_404(self, request):
handler = self.error_handlers.get(404)
if handler:
resp = handler(request)
elif request.path.endswith('.json'):
resp = MiniResponse.json({'error': 'Not Found'}, '404 Not Found')
else:
resp = MiniResponse.html('<h1>404 Not Found</h1>', '404 Not Found')
return resp
def _handle_error(self, code, request, message):
handler = self.error_handlers.get(code)
if handler:
return handler(request)
if request.path.endswith('.json'):
return MiniResponse.json({'error': message}, f'{code} Error')
return MiniResponse.html(f'<h1>{code}</h1><p>{message}</p>', f'{code} Error')
# ===== 实战：博客应用 =====
def create_blog_app():
app = MiniApp()
# 模拟数据
posts = [
{'id': 1, 'title': 'Hello World', 'content': '这是我的第一篇博客文章！', 'views': 100},
{'id': 2, 'title': 'Python 入门', 'content': 'Python 是一门简单易学的语言...', 'views': 200},
{'id': 3, 'title': 'Web 开发指南', 'content': '从零开始学习 Web 开发...', 'views': 150},
]
next_id = 4
comments = {1: ['写得好！', '学到了'], 2: ['非常有用']}
# ===== 钩子函数 =====
@app.before_request
def log_request(req):
print(f"[{req.method}] {req.path}")
@app.after_request
def add_security_headers(req, resp):
resp.headers.append(('X-Frame-Options', 'DENY'))
resp.headers.append(('X-Content-Type-Options', 'nosniff'))
return resp
# ===== 错误处理 =====
@app.error(500)
def handle_500(req):
return MiniResponse.json({'error': '服务器内部错误'}, '500 Error')
# ===== 页面路由 =====
@app.get('/')
def index(req):
title = '首页 - 我的博客'
html = f'''
<div style="max-width:800px;margin:0 auto;padding:20px;font-family:sans-serif;">
<h1 style="color:#654ACB;">{title}</h1>
<nav style="margin:20px 0;">
<a href="/posts" style="margin-right:15px;">全部文章</a>
<a href="/about">关于我</a>
</nav>
<hr/>
<h2>最新文章</h2>
{"".join(f'''
<div style="margin:15px 0;padding:15px;border:1px solid #eee;border-radius:8px;">
<h3><a href="/posts/{p['id']}" style="color:#654ACB;text-decoration:none;">{p['title']}</a></h3>
<p style="color:#666;">{p['content'][:50]}...</p>
<small style="color:#999;">浏览: {p['views']}</small>
</div>
''' for p in posts)}
</div>
'''
return MiniResponse.html(html)
@app.get('/posts')
def list_posts(req):
html = f'''<div style="max-width:800px;margin:0 auto;padding:20px;">
<h1>全部文章</h1>
{"".join(f'<p><a href="/posts/{p["id"]}">{p["title"]}</a> (浏览: {p["views"]})</p>' for p in posts)}
</div>'''
return MiniResponse.html(html)
@app.get('/posts/<id>')
def view_post(req):
pid = req.path.split('/')[-1]
post = next((p for p in posts if str(p['id']) == pid), None)
if not post:
return MiniResponse.text('文章不存在', '404 Not Found')
post['views'] += 1
comments_list = comments.get(post['id'], [])
html = f'''<div style="max-width:800px;margin:0 auto;padding:20px;">
<a href="/">← 返回首页</a>
<h1>{post['title']}</h1>
<p>{post['content']}</p>
<small>浏览量: {post['views']}</small>
<hr/>
<h3>评论 ({len(comments_list)})</h3>
<ul>
{"".join(f'<li>{c}</li>' for c in comments_list)}
</ul>
<h3>发表评论</h3>
<form method="POST" action="/posts/{pid}/comments">
<textarea name="comment" rows="3" cols="50"></textarea><br/>
<button type="submit">提交评论</button>
</form>
</div>'''
return MiniResponse.html(html)
@app.post('/posts/<id>/comments')
def add_comment(req):
pid = req.path.split('/')[2]
comment = req.form_data.get('comment', '')
if comment:
comments.setdefault(int(pid), []).append(comment)
return MiniResponse.redirect(f'/posts/{pid}')
@app.get('/api/posts')
def api_list_posts(req):
return MiniResponse.json(posts)
@app.get('/api/posts/<id>')
def api_get_post(req):
pid = req.path.split('/')[-1]
post = next((p for p in posts if str(p['id']) == pid), None)
if post:
return MiniResponse.json(post)
return MiniResponse.json({'error': 'Not Found'}, '404 Not Found')
@app.post('/api/posts')
def api_create_post(req):
global next_id
data = req.json_data
if data and 'title' in data:
new_post = {
'id': next_id,
'title': data['title'],
'content': data.get('content', ''),
'views': 0,
}
posts.append(new_post)
next_id += 1
return MiniResponse.json(new_post, '201 Created')
return MiniResponse.json({'error': '缺少 title'}, '400 Bad Request')
@app.delete('/api/posts/<id>')
def api_delete_post(req):
global posts
pid = req.path.split('/')[-1]
posts = [p for p in posts if str(p['id']) != pid]
return MiniResponse.json({'deleted': pid})
@app.get('/about')
def about(req):
return MiniResponse.html('''<div style="max-width:600px;margin:0 auto;padding:20px;">
<h1>关于我</h1>
<p>这是一个用纯 WSGI 编写的迷你博客系统，无任何第三方依赖。</p>
</div>''')
return app
if __name__ == '__main__':
from wsgiref.simple_server import make_server
blog_app = create_blog_app()
server = make_server('127.0.0.1', 8088, blog_app)
print("✓ 迷你博客系统运行在 http://127.0.0.1:8088")
print("\n可用的 API:")
print(" GET / -> 首页")
print(" GET /posts -> 文章列表")
print(" GET /posts/<id> -> 文章详情")
print(" POST /posts/<id>/comments -> 发表评论")
print(" GET /api/posts -> JSON 文章列表")
print(" GET /api/posts/<id> -> JSON 文章详情")
print(" POST /api/posts -> 创建文章")
print(" DELETE /api/posts/<id> -> 删除文章")
print("\n按 Ctrl+C 停止")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n服务器已停止")
小结
| 知识点 | 说明 |
|---|---|
| HTTP 协议 | 请求/响应模型，基于 TCP |
| 请求方法 | GET/POST/PUT/PATCH/DELETE |
| 状态码 | 2xx 成功 / 3xx 重定向 / 4xx 客户端错误 / 5xx 服务端错误 |
| Headers | 传输元数据的键值对 |
| HTML 基础 | 页面结构和样式标记 |
| WSGI 接口 | app(environ, start_response) 两参形式 |
| environ 字典 | 包含 HTTP 请求的所有信息 |
| start_response | 回调函数，设置响应状态码和头 |
| 中间件 | 层层包装的函数，可拦截和修改请求/响应 |
| mini-web 框架 | 用 WSGI 实现的迷你 MVC 框架原型 |
核心理解：所有 Python Web 框架的本质都是一个 WSGI 应用（或者 ASGI，异步版本）。理解了 WSGI，你就理解了 Web 框架的根本——它们只是在 WSGI 之上添加了路由、模板、ORM 等便利功能。当你需要排查框架级问题时，回归 WSGI 视角往往能更快定位问题。
练习题
在 WSGI 应用中实现一个简单的静态文件服务器，能够返回
.html、.css、.js文件。当文件不存在时返回 404。提示：使用os.path和open()读取文件。实现一个 WSGI 中间件，统计每个请求的处理时间，并将耗时超过 1 秒的请求记录到日志中。尝试用它包裹你前面的博客应用进行测试。
扩展 mini-web 框架，添加模板引擎支持。实现一个
render_template(name, **context)方法，支持{% %}控制和{variable}变量替换（无需完整 Jinja2 能力，实现基本功能即可）。