HTTP 服务项目实战

本章将综合运用前面学习的知识，构建一个完整的 HTTP API 服务。

项目介绍

我们将构建一个待办事项 API 服务，支持：

创建、读取、更新、删除待办事项（CRUD）
JSON 数据交互
持久化存储
错误处理

项目结构

todo_api/
├── app.py           # 主应用入口
├── models.py        # 数据模型
├── storage.py       # 数据存储
├── handlers.py      # 请求处理
├── data/
│   └── todos.json   # 数据文件
└── requirements.txt # 依赖

第一步：使用标准库创建简单 HTTP 服务

Python 标准库自带 HTTP 服务器，无需安装任何依赖。

最简单的 HTTP 服务器

# simple_server.py
from http.server import HTTPServer, BaseHTTPRequestHandler

class SimpleHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        # 设置响应状态码
        self.send_response(200)
        # 设置响应头
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.end_headers()
        # 发送响应体
        self.wfile.write("Hello, World! 你好世界！".encode("utf-8"))

# 启动服务器
server = HTTPServer(("localhost", 8000), SimpleHandler)
print("服务器启动: http://localhost:8000")
server.serve_forever()

运行后访问 http://localhost:8000 查看效果。

处理不同路径

# path_server.py
from http.server import HTTPServer, BaseHTTPRequestHandler
import json

class PathHandler(BaseHTTPRequestHandler):

    def do_GET(self):
        if self.path == "/":
            self.send_json({"message": "欢迎访问API"})
        elif self.path == "/hello":
            self.send_json({"message": "Hello, World!"})
        elif self.path == "/time":
            from datetime import datetime
            self.send_json({"time": datetime.now().isoformat()})
        else:
            self.send_error_json(404, "路径不存在")

    def send_json(self, data, status=200):
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.end_headers()
        self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8"))

    def send_error_json(self, status, message):
        self.send_json({"error": message}, status)

    # 禁用默认日志
    def log_message(self, format, *args):
        print(f"[{self.command}] {self.path}")


server = HTTPServer(("localhost", 8000), PathHandler)
print("服务器启动: http://localhost:8000")
server.serve_forever()

第二步：构建待办事项数据模型

# models.py
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Optional
import uuid


@dataclass
class Todo:
    """待办事项数据模型"""
    title: str
    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    completed: bool = False
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: Optional[str] = None

    def to_dict(self):
        """转换为字典"""
        return asdict(self)

    def complete(self):
        """标记为完成"""
        self.completed = True
        self.updated_at = datetime.now().isoformat()

    def update(self, title: str = None, completed: bool = None):
        """更新待办事项"""
        if title is not None:
            self.title = title
        if completed is not None:
            self.completed = completed
        self.updated_at = datetime.now().isoformat()

    @classmethod
    def from_dict(cls, data: dict):
        """从字典创建实例"""
        return cls(
            id=data.get("id", str(uuid.uuid4())[:8]),
            title=data["title"],
            completed=data.get("completed", False),
            created_at=data.get("created_at", datetime.now().isoformat()),
            updated_at=data.get("updated_at")
        )

第三步：实现数据存储层

# storage.py
import json
from pathlib import Path
from typing import List, Optional
from models import Todo


class TodoStorage:
    """待办事项存储类"""

    def __init__(self, data_file: str = "data/todos.json"):
        self.data_file = Path(data_file)
        self.data_file.parent.mkdir(parents=True, exist_ok=True)
        self.todos: List[Todo] = self.load()

    def load(self) -> List[Todo]:
        """从文件加载数据"""
        try:
            if self.data_file.exists():
                with open(self.data_file, "r", encoding="utf-8") as f:
                    data = json.load(f)
                    return [Todo.from_dict(item) for item in data]
        except (json.JSONDecodeError, KeyError) as e:
            print(f"加载数据失败: {e}")
        return []

    def save(self):
        """保存数据到文件"""
        try:
            with open(self.data_file, "w", encoding="utf-8") as f:
                data = [todo.to_dict() for todo in self.todos]
                json.dump(data, f, ensure_ascii=False, indent=2)
        except Exception as e:
            print(f"保存数据失败: {e}")

    def get_all(self) -> List[Todo]:
        """获取所有待办事项"""
        return self.todos

    def get_by_id(self, todo_id: str) -> Optional[Todo]:
        """根据ID获取待办事项"""
        for todo in self.todos:
            if todo.id == todo_id:
                return todo
        return None

    def create(self, title: str) -> Todo:
        """创建待办事项"""
        todo = Todo(title=title)
        self.todos.append(todo)
        self.save()
        return todo

    def update(self, todo_id: str, title: str = None, completed: bool = None) -> Optional[Todo]:
        """更新待办事项"""
        todo = self.get_by_id(todo_id)
        if todo:
            todo.update(title=title, completed=completed)
            self.save()
        return todo

    def delete(self, todo_id: str) -> bool:
        """删除待办事项"""
        todo = self.get_by_id(todo_id)
        if todo:
            self.todos.remove(todo)
            self.save()
            return True
        return False

    def clear_completed(self) -> int:
        """清除已完成的待办事项"""
        original_count = len(self.todos)
        self.todos = [t for t in self.todos if not t.completed]
        self.save()
        return original_count - len(self.todos)

第四步：实现 HTTP 请求处理

# handlers.py
from http.server import BaseHTTPRequestHandler
import json
from urllib.parse import urlparse, parse_qs
from storage import TodoStorage


class TodoHandler(BaseHTTPRequestHandler):
    """待办事项 API 处理器"""

    # 类级别的存储实例（所有请求共享）
    storage = TodoStorage()

    def do_GET(self):
        """处理 GET 请求"""
        parsed = urlparse(self.path)
        path = parsed.path

        if path == "/api/todos":
            # 获取所有待办事项
            todos = [t.to_dict() for t in self.storage.get_all()]
            self.send_json({"todos": todos, "count": len(todos)})

        elif path.startswith("/api/todos/"):
            # 获取单个待办事项
            todo_id = path.split("/")[-1]
            todo = self.storage.get_by_id(todo_id)
            if todo:
                self.send_json(todo.to_dict())
            else:
                self.send_error_json(404, f"待办事项 {todo_id} 不存在")

        elif path == "/":
            # API 首页
            self.send_json({
                "message": "待办事项 API",
                "version": "1.0",
                "endpoints": {
                    "GET /api/todos": "获取所有待办事项",
                    "GET /api/todos/:id": "获取单个待办事项",
                    "POST /api/todos": "创建待办事项",
                    "PUT /api/todos/:id": "更新待办事项",
                    "DELETE /api/todos/:id": "删除待办事项"
                }
            })

        else:
            self.send_error_json(404, "路径不存在")

    def do_POST(self):
        """处理 POST 请求"""
        if self.path == "/api/todos":
            try:
                data = self.get_json_body()
                if not data.get("title"):
                    self.send_error_json(400, "title 字段必填")
                    return

                todo = self.storage.create(data["title"])
                self.send_json(todo.to_dict(), 201)

            except json.JSONDecodeError:
                self.send_error_json(400, "无效的 JSON 数据")
        else:
            self.send_error_json(404, "路径不存在")

    def do_PUT(self):
        """处理 PUT 请求"""
        if self.path.startswith("/api/todos/"):
            todo_id = self.path.split("/")[-1]

            try:
                data = self.get_json_body()
                todo = self.storage.update(
                    todo_id,
                    title=data.get("title"),
                    completed=data.get("completed")
                )

                if todo:
                    self.send_json(todo.to_dict())
                else:
                    self.send_error_json(404, f"待办事项 {todo_id} 不存在")

            except json.JSONDecodeError:
                self.send_error_json(400, "无效的 JSON 数据")
        else:
            self.send_error_json(404, "路径不存在")

    def do_DELETE(self):
        """处理 DELETE 请求"""
        if self.path.startswith("/api/todos/"):
            todo_id = self.path.split("/")[-1]

            if self.storage.delete(todo_id):
                self.send_json({"message": f"待办事项 {todo_id} 已删除"})
            else:
                self.send_error_json(404, f"待办事项 {todo_id} 不存在")
        else:
            self.send_error_json(404, "路径不存在")

    def get_json_body(self):
        """获取请求体的 JSON 数据"""
        content_length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(content_length)
        return json.loads(body.decode("utf-8"))

    def send_json(self, data, status=200):
        """发送 JSON 响应"""
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Access-Control-Allow-Origin", "*")  # CORS
        self.end_headers()
        self.wfile.write(json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8"))

    def send_error_json(self, status, message):
        """发送错误响应"""
        self.send_json({"error": message, "status": status}, status)

    def do_OPTIONS(self):
        """处理预检请求（CORS）"""
        self.send_response(200)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def log_message(self, format, *args):
        """自定义日志格式"""
        print(f"[{self.command}] {self.path} - {args[1] if len(args) > 1 else ''}")

第五步：主应用入口

# app.py
from http.server import HTTPServer
from handlers import TodoHandler


def run_server(host="localhost", port=8000):
    """启动 HTTP 服务器"""
    server_address = (host, port)
    httpd = HTTPServer(server_address, TodoHandler)

    print(f"""
╔══════════════════════════════════════════════════════╗
║           待办事项 API 服务已启动                    ║
╠══════════════════════════════════════════════════════╣
║  地址: http://{host}:{port}                        ║
║                                                      ║
║  API 端点:                                           ║
║    GET    /api/todos      - 获取所有待办事项        ║
║    GET    /api/todos/:id  - 获取单个待办事项        ║
║    POST   /api/todos      - 创建待办事项            ║
║    PUT    /api/todos/:id  - 更新待办事项            ║
║    DELETE /api/todos/:id  - 删除待办事项            ║
║                                                      ║
║  按 Ctrl+C 停止服务器                               ║
╚══════════════════════════════════════════════════════╝
    """)

    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n服务器已停止")
        httpd.shutdown()


if __name__ == "__main__":
    run_server()

第六步：测试 API

使用 curl 测试

# 1. 创建待办事项
curl -X POST http://localhost:8000/api/todos \
  -H "Content-Type: application/json" \
  -d '{"title": "学习 Python"}'

# 2. 获取所有待办事项
curl http://localhost:8000/api/todos

# 3. 获取单个待办事项
curl http://localhost:8000/api/todos/abc123

# 4. 更新待办事项
curl -X PUT http://localhost:8000/api/todos/abc123 \
  -H "Content-Type: application/json" \
  -d '{"completed": true}'

# 5. 删除待办事项
curl -X DELETE http://localhost:8000/api/todos/abc123

使用 Python 测试

# test_api.py
import requests

BASE_URL = "http://localhost:8000/api"

def test_api():
    # 1. 创建待办事项
    print("=== 创建待办事项 ===")
    resp = requests.post(f"{BASE_URL}/todos", json={"title": "学习 Python"})
    print(resp.json())
    todo_id = resp.json()["id"]

    # 2. 获取所有待办事项
    print("\n=== 获取所有待办事项 ===")
    resp = requests.get(f"{BASE_URL}/todos")
    print(resp.json())

    # 3. 更新待办事项
    print("\n=== 更新待办事项 ===")
    resp = requests.put(f"{BASE_URL}/todos/{todo_id}", json={"completed": True})
    print(resp.json())

    # 4. 获取单个待办事项
    print("\n=== 获取单个待办事项 ===")
    resp = requests.get(f"{BASE_URL}/todos/{todo_id}")
    print(resp.json())

    # 5. 删除待办事项
    print("\n=== 删除待办事项 ===")
    resp = requests.delete(f"{BASE_URL}/todos/{todo_id}")
    print(resp.json())


if __name__ == "__main__":
    test_api()

进阶：使用 Flask 框架

Flask 是一个轻量级 Web 框架，开发效率更高。

安装 Flask

pip install flask

Flask 版本实现

# flask_app.py
from flask import Flask, request, jsonify
from storage import TodoStorage

app = Flask(__name__)
storage = TodoStorage()


@app.route("/")
def index():
    """API 首页"""
    return jsonify({
        "message": "待办事项 API (Flask 版)",
        "version": "1.0"
    })


@app.route("/api/todos", methods=["GET"])
def get_todos():
    """获取所有待办事项"""
    todos = [t.to_dict() for t in storage.get_all()]
    return jsonify({"todos": todos, "count": len(todos)})


@app.route("/api/todos/<todo_id>", methods=["GET"])
def get_todo(todo_id):
    """获取单个待办事项"""
    todo = storage.get_by_id(todo_id)
    if todo:
        return jsonify(todo.to_dict())
    return jsonify({"error": "待办事项不存在"}), 404


@app.route("/api/todos", methods=["POST"])
def create_todo():
    """创建待办事项"""
    data = request.get_json()
    if not data or not data.get("title"):
        return jsonify({"error": "title 字段必填"}), 400

    todo = storage.create(data["title"])
    return jsonify(todo.to_dict()), 201


@app.route("/api/todos/<todo_id>", methods=["PUT"])
def update_todo(todo_id):
    """更新待办事项"""
    data = request.get_json()
    todo = storage.update(
        todo_id,
        title=data.get("title"),
        completed=data.get("completed")
    )

    if todo:
        return jsonify(todo.to_dict())
    return jsonify({"error": "待办事项不存在"}), 404


@app.route("/api/todos/<todo_id>", methods=["DELETE"])
def delete_todo(todo_id):
    """删除待办事项"""
    if storage.delete(todo_id):
        return jsonify({"message": "删除成功"})
    return jsonify({"error": "待办事项不存在"}), 404


@app.errorhandler(404)
def not_found(error):
    return jsonify({"error": "路径不存在"}), 404


@app.errorhandler(500)
def internal_error(error):
    return jsonify({"error": "服务器内部错误"}), 500


if __name__ == "__main__":
    print("Flask 服务器启动: http://localhost:5000")
    app.run(debug=True, port=5000)

完整项目代码

将所有文件组合在一起：

# 完整的单文件版本 todo_api.py

from http.server import HTTPServer, BaseHTTPRequestHandler
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import List, Optional
from pathlib import Path
import json
import uuid


# ==================== 数据模型 ====================

@dataclass
class Todo:
    title: str
    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    completed: bool = False
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    updated_at: Optional[str] = None

    def to_dict(self):
        return asdict(self)

    def update(self, title=None, completed=None):
        if title is not None:
            self.title = title
        if completed is not None:
            self.completed = completed
        self.updated_at = datetime.now().isoformat()

    @classmethod
    def from_dict(cls, data):
        return cls(
            id=data.get("id", str(uuid.uuid4())[:8]),
            title=data["title"],
            completed=data.get("completed", False),
            created_at=data.get("created_at", datetime.now().isoformat()),
            updated_at=data.get("updated_at")
        )


# ==================== 数据存储 ====================

class TodoStorage:
    def __init__(self, data_file="todos.json"):
        self.data_file = Path(data_file)
        self.todos: List[Todo] = self.load()

    def load(self):
        try:
            if self.data_file.exists():
                with open(self.data_file, "r", encoding="utf-8") as f:
                    return [Todo.from_dict(item) for item in json.load(f)]
        except Exception as e:
            print(f"加载失败: {e}")
        return []

    def save(self):
        with open(self.data_file, "w", encoding="utf-8") as f:
            json.dump([t.to_dict() for t in self.todos], f, ensure_ascii=False, indent=2)

    def get_all(self):
        return self.todos

    def get_by_id(self, todo_id):
        return next((t for t in self.todos if t.id == todo_id), None)

    def create(self, title):
        todo = Todo(title=title)
        self.todos.append(todo)
        self.save()
        return todo

    def update(self, todo_id, title=None, completed=None):
        todo = self.get_by_id(todo_id)
        if todo:
            todo.update(title=title, completed=completed)
            self.save()
        return todo

    def delete(self, todo_id):
        todo = self.get_by_id(todo_id)
        if todo:
            self.todos.remove(todo)
            self.save()
            return True
        return False


# ==================== HTTP 处理器 ====================

class TodoHandler(BaseHTTPRequestHandler):
    storage = TodoStorage()

    def do_GET(self):
        if self.path == "/api/todos":
            todos = [t.to_dict() for t in self.storage.get_all()]
            self.send_json({"todos": todos, "count": len(todos)})
        elif self.path.startswith("/api/todos/"):
            todo_id = self.path.split("/")[-1]
            todo = self.storage.get_by_id(todo_id)
            if todo:
                self.send_json(todo.to_dict())
            else:
                self.send_error_json(404, "待办事项不存在")
        elif self.path == "/":
            self.send_json({"message": "待办事项 API", "version": "1.0"})
        else:
            self.send_error_json(404, "路径不存在")

    def do_POST(self):
        if self.path == "/api/todos":
            try:
                data = self.get_json_body()
                if not data.get("title"):
                    self.send_error_json(400, "title 必填")
                    return
                todo = self.storage.create(data["title"])
                self.send_json(todo.to_dict(), 201)
            except Exception as e:
                self.send_error_json(400, str(e))
        else:
            self.send_error_json(404, "路径不存在")

    def do_PUT(self):
        if self.path.startswith("/api/todos/"):
            todo_id = self.path.split("/")[-1]
            try:
                data = self.get_json_body()
                todo = self.storage.update(todo_id, data.get("title"), data.get("completed"))
                if todo:
                    self.send_json(todo.to_dict())
                else:
                    self.send_error_json(404, "待办事项不存在")
            except Exception as e:
                self.send_error_json(400, str(e))
        else:
            self.send_error_json(404, "路径不存在")

    def do_DELETE(self):
        if self.path.startswith("/api/todos/"):
            todo_id = self.path.split("/")[-1]
            if self.storage.delete(todo_id):
                self.send_json({"message": "删除成功"})
            else:
                self.send_error_json(404, "待办事项不存在")
        else:
            self.send_error_json(404, "路径不存在")

    def get_json_body(self):
        length = int(self.headers.get("Content-Length", 0))
        return json.loads(self.rfile.read(length).decode("utf-8"))

    def send_json(self, data, status=200):
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.end_headers()
        self.wfile.write(json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8"))

    def send_error_json(self, status, message):
        self.send_json({"error": message}, status)

    def log_message(self, format, *args):
        print(f"[{self.command}] {self.path}")


# ==================== 启动服务器 ====================

if __name__ == "__main__":
    server = HTTPServer(("localhost", 8000), TodoHandler)
    print("服务器启动: http://localhost:8000")
    print("API: http://localhost:8000/api/todos")
    print("按 Ctrl+C 停止")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n服务器已停止")

项目扩展思路

添加用户认证
- 实现用户注册/登录
- 使用 JWT Token 认证
- 待办事项关联用户
使用数据库
- SQLite 本地数据库
- MySQL/PostgreSQL 生产环境
添加前端页面
- HTML + JavaScript 前端
- Vue/React 单页应用
部署上线
- 使用 Gunicorn/uWSGI
- Docker 容器化
- 部署到云服务器

小结

通过这个项目，我们实践了：

Python 基础语法：变量、函数、类
数据类型：字典、列表、dataclass
面向对象：类的设计、封装
文件操作：JSON 读写、持久化
异常处理：try-except、错误响应
模块化：代码组织、分层设计
HTTP 服务：请求处理、RESTful API

恭喜你完成了 Python 入门教程！继续练习，尝试更多项目！