Web开发入门：HTTP协议、WSGI与路由概念

Web开发是Python最热门的应用方向之一。在学习Flask、Django等框架之前，理解HTTP协议、WSGI规范和路由概念至关重要。本文将从底层原理开始，带你理解Web开发的核心机制。

HTTP协议基础

HTTP（超文本传输协议）是Web的基础，采用客户端-服务器模型：

请求与响应

# HTTP请求的结构
"""
GET /api/users?page=1 HTTP/1.1
Host: www.example.com
User-Agent: Mozilla/5.0
Accept: application/json
Authorization: Bearer token123

{"name": "value"}
"""

# HTTP响应的结构
"""
HTTP/1.1 200 OK
Content-Type: application/json
Content-Length: 123
Set-Cookie: session=abc123

{"users": [{"id": 1, "name": "张三"}]}
"""

HTTP方法

方法	用途	幂等性	安全性
GET	获取资源	是	是
POST	创建资源	否	否
PUT	更新资源（全量）	是	否
PATCH	更新资源（部分）	否	否
DELETE	删除资源	是	否
HEAD	获取响应头	是	是
OPTIONS	查询支持的方法	是	是

状态码

# 常见HTTP状态码
STATUS_CODES = {
    # 2xx 成功
    200: "OK",
    201: "Created",
    204: "No Content",
    
    # 3xx 重定向
    301: "Moved Permanently",
    302: "Found",
    304: "Not Modified",
    
    # 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"
}

WSGI规范

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

WSGI应用结构

# 最简单的WSGI应用
def simple_app(environ, start_response):
    """
    environ: 包含所有请求信息的字典
    start_response: 发送HTTP响应头的回调函数
    """
    # 获取请求信息
    path = environ.get("PATH_INFO", "/")
    method = environ.get("REQUEST_METHOD", "GET")
    query_string = environ.get("QUERY_STRING", "")
    
    # 构建响应
    status = "200 OK"
    headers = [("Content-Type", "text/plain; charset=utf-8")]
    body = f"请求路径: {path}\n请求方法: {method}\n查询参数: {query_string}"
    
    # 发送响应
    start_response(status, headers)
    return [body.encode("utf-8")]

# 使用wsgiref运行（开发用）
if __name__ == "__main__":
    from wsgiref.simple_server import make_server
    
    server = make_server("127.0.0.1", 8000, simple_app)
    print("服务器运行在 http://127.0.0.1:8000")
    server.serve_forever()

解析请求

from urllib.parse import parse_qs, urlparse

def parse_request_app(environ, start_response):
    """解析请求的WSGI应用"""
    
    # 解析URL
    url = urlparse(environ.get("PATH_INFO", "/"))
    path = url.path
    
    # 解析查询参数
    query_params = parse_qs(environ.get("QUERY_STRING", ""))
    
    # 读取请求体
    content_length = int(environ.get("CONTENT_LENGTH", 0))
    request_body = environ["wsgi.input"].read(content_length) if content_length else b""
    
    # 获取请求头
    headers = {
        k[5:].lower().replace("_", "-"): v
        for k, v in environ.items()
        if k.startswith("HTTP_")
    }
    
    # 构建响应
    import json
    response_data = {
        "path": path,
        "method": environ.get("REQUEST_METHOD"),
        "query": query_params,
        "headers": headers,
        "body": request_body.decode("utf-8") if request_body else None
    }
    
    status = "200 OK"
    response_headers = [("Content-Type", "application/json")]
    body = json.dumps(response_data, ensure_ascii=False, indent=2)
    
    start_response(status, response_headers)
    return [body.encode("utf-8")]

路由概念

路由是将URL映射到处理函数的机制：

简单路由器实现

import re
from urllib.parse import parse_qs

class Router:
    """简单的路由器实现"""
    
    def __init__(self):
        self.routes = []
    
    def add_route(self, method, path, handler):
        """添加路由"""
        # 将路径模式转换为正则表达式
        pattern = re.sub(r"\{(\w+)\}", r"(?P<\1>[^/]+)", path)
        pattern = f"^{pattern}$"
        self.routes.append({
            "method": method.upper(),
            "pattern": re.compile(pattern),
            "handler": handler
        })
    
    def get(self, path, handler):
        self.add_route("GET", path, handler)
    
    def post(self, path, handler):
        self.add_route("POST", path, handler)
    
    def match(self, method, path):
        """匹配路由"""
        for route in self.routes:
            if route["method"] == method.upper():
                match = route["pattern"].match(path)
                if match:
                    return route["handler"], match.groupdict()
        return None, {}

# 使用路由器
router = Router()

def index_handler(params):
    return {"message": "欢迎访问首页"}

def user_handler(params):
    return {"user_id": params.get("id")}

def create_user_handler(data):
    return {"created": True, "user": data}

router.get("/", index_handler)
router.get("/users/{id}", user_handler)
router.post("/users", create_user_handler)

完整的WSGI路由应用

from wsgiref.simple_server import make_server
from urllib.parse import parse_qs, urlparse
import json
import re

class SimpleWebApp:
    """简单的Web应用框架"""
    
    def __init__(self):
        self.routes = []
        self.before_request_hooks = []
        self.after_request_hooks = []
    
    def route(self, path, methods=None):
        """路由装饰器"""
        if methods is None:
            methods = ["GET"]
        
        def decorator(handler):
            self.routes.append({
                "path": path,
                "methods": [m.upper() for m in methods],
                "handler": handler
            })
            return handler
        return decorator
    
    def before_request(self, hook):
        """请求前钩子"""
        self.before_request_hooks.append(hook)
        return hook
    
    def __call__(self, environ, start_response):
        """WSGI接口"""
        # 解析请求
        method = environ["REQUEST_METHOD"]
        path = environ.get("PATH_INFO", "/")
        
        # 匹配路由
        handler, params = self._match_route(method, path)
        
        if handler is None:
            return self._response(start_response, 404, {"error": "Not Found"})
        
        # 执行请求前钩子
        for hook in self.before_request_hooks:
            hook(environ)
        
        # 读取请求体
        content_length = int(environ.get("CONTENT_LENGTH", 0))
        body = environ["wsgi.input"].read(content_length) if content_length else b""
        
        # 解析请求数据
        request_data = {
            "method": method,
            "path": path,
            "params": params,
            "query": parse_qs(environ.get("QUERY_STRING", "")),
            "body": body.decode("utf-8") if body else None,
            "headers": {k[5:].lower(): v for k, v in environ.items() if k.startswith("HTTP_")}
        }
        
        # 调用处理函数
        try:
            result = handler(request_data, **params)
            return self._response(start_response, 200, result)
        except Exception as e:
            return self._response(start_response, 500, {"error": str(e)})
    
    def _match_route(self, method, path):
        for route in self.routes:
            if method in route["methods"]:
                pattern = re.sub(r"\{(\w+)\}", r"(?P<\1>[^/]+)", route["path"])
                match = re.match(f"^{pattern}$", path)
                if match:
                    return route["handler"], match.groupdict()
        return None, {}
    
    def _response(self, start_response, status_code, data):
        status = f"{status_code} {'OK' if status_code == 200 else 'Error'}"
        headers = [("Content-Type", "application/json")]
        body = json.dumps(data, ensure_ascii=False)
        start_response(status, headers)
        return [body.encode("utf-8")]
    
    def run(self, host="127.0.0.1", port=8000):
        server = make_server(host, port, self)
        print(f"服务器运行在 http://{host}:{port}")
        server.serve_forever()

# 使用示例
app = SimpleWebApp()

@app.route("/")
def index(request):
    return {"message": "欢迎访问"}

@app.route("/users/{user_id}")
def get_user(request, user_id):
    return {"user_id": user_id, "name": "张三"}

@app.route("/users", methods=["POST"])
def create_user(request):
    data = json.loads(request["body"])
    return {"created": True, "user": data}

if __name__ == "__main__":
    app.run()

RESTful API设计

# RESTful API设计原则
"""
资源导向:
  GET    /api/users        - 获取用户列表
  POST   /api/users        - 创建用户
  GET    /api/users/1      - 获取用户1
  PUT    /api/users/1      - 更新用户1
  DELETE /api/users/1      - 删除用户1

嵌套资源:
  GET    /api/users/1/posts - 获取用户1的文章
  POST   /api/users/1/posts - 为用户1创建文章
"""

# 使用Flask的RESTful示例
"""
from flask import Flask, jsonify, request

app = Flask(__name__)
users = []

@app.route("/api/users", methods=["GET"])
def get_users():
    return jsonify(users)

@app.route("/api/users", methods=["POST"])
def create_user():
    user = request.json
    user["id"] = len(users) + 1
    users.append(user)
    return jsonify(user), 201

@app.route("/api/users/<int:user_id>", methods=["GET"])
def get_user(user_id):
    user = next((u for u in users if u["id"] == user_id), None)
    if user:
        return jsonify(user)
    return jsonify({"error": "Not found"}), 404
"""

中间件概念

# 中间件是处理请求和响应的组件
class LoggingMiddleware:
    """日志中间件"""
    
    def __init__(self, app):
        self.app = app
    
    def __call__(self, environ, start_response):
        import time
        start_time = time.time()
        
        def custom_start_response(status, headers, exc_info=None):
            duration = time.time() - start_time
            print(f"{environ['REQUEST_METHOD']} {environ['PATH_INFO']} - {status} ({duration:.3f}s)")
            return start_response(status, headers, exc_info)
        
        return self.app(environ, custom_start_response)

class CORSMiddleware:
    """CORS中间件"""
    
    def __init__(self, app):
        self.app = app
    
    def __call__(self, environ, start_response):
        def custom_start_response(status, headers, exc_info=None):
            headers.append(("Access-Control-Allow-Origin", "*"))
            headers.append(("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE"))
            headers.append(("Access-Control-Allow-Headers", "Content-Type, Authorization"))
            return start_response(status, headers, exc_info)
        
        return self.app(environ, custom_start_response)

# 组合中间件
# app = CORSMiddleware(LoggingMiddleware(simple_app))

总结

理解HTTP协议、WSGI规范和路由概念是掌握Python Web开发的基础。虽然实际开发中我们使用Flask、Django等框架，但理解底层原理能帮助你更好地调试、优化和扩展应用。下一步建议学习Flask框架，将这些概念应用到实际项目中。