文章

23 · 前后端解耦：AJAX 与 RESTful API 设计

#023 · 2026-04-16 · Python
PythonPython教程进阶RESTAJAXAPI

🔗 知识图谱导航：阅读本文前，建议先掌握《18 · HTTP 与 Web 框架基石》中的请求方法语义（GET/POST/PUT/PATCH/DELETE），以及《22 · Nginx + uWSGI》中的 WSGI 协议——本文在此基础上设计完整的 RESTful API，并用零依赖 Python 实现。

运行环境：Python 3.12+ 标准库，零额外依赖，直接运行。

极客解析：REST 不是协议，是一种设计风格。核心约定：用 URL 表示资源，用 HTTP 方法表示操作，用状态码表示结果，用 JSON 传递数据。理解了这四点，所有 REST API 都是同一套逻辑。

RESTful API 设计规范

资源 URL 设计：
  GET    /api/tasks          → 获取任务列表（支持过滤/排序/分页）
  POST   /api/tasks          → 创建新任务
  GET    /api/tasks/{id}     → 获取单个任务
  PUT    /api/tasks/{id}     → 全量替换任务（必须提供所有字段）
  PATCH  /api/tasks/{id}     → 部分更新任务（只提供要改的字段）
  DELETE /api/tasks/{id}     → 删除任务（返回 204 No Content）

响应结构约定：
  列表：{"data": [...], "pagination": {"page": 1, "total": 5, ...}}
  单条：{"data": {...}}
  错误：{"error": "描述信息"}

PUT vs PATCH 的本质区别

原始数据：{"id": 1, "title": "设计 API", "done": false, "priority": 3}

PUT  {"title": "新标题", "done": true}
→ 全量替换：{"id": 1, "title": "新标题", "done": true, "priority": 0}
  （priority 缺失，用默认值 0 填充）

PATCH {"done": true}
→ 部分更新：{"id": 1, "title": "设计 API", "done": true, "priority": 3}
  （只改 done，其他字段保持不变）

步步为营：核心逻辑自适应拆解

这一篇的核心是 REST API 的四个约定：URL 表示资源、HTTP 方法表示操作、状态码表示结果、JSON 传递数据。下面每一步都聚焦一个机制，零依赖可直接运行。

Step 1：用 _send 理解 REST 响应的标准结构

痛点与机制：

_send 是 REST API 响应的"打包工"：把 Python 字典序列化成 JSON，设置 Content-Type: application/json 和 Content-Length，用 STATUS_TEXT 字典把状态码翻译成文字。REST API 的响应结构有约定：成功用 {"data": ...}，错误用 {"error": "..."}，列表用 {"data": [...], "pagination": {...}}——这个约定让前端代码可以统一处理所有响应。

核心源码（逐字来自文末完整源码）：

    def _send(self, status: int, body: Any = None) -> None:
        if body is None:
            self.send_response(status)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

可运行演示（补齐 Mock 数据与 print 反馈）：

import json
from typing import Any

STATUS_TEXT = {200: "OK", 201: "Created", 204: "No Content",
               400: "Bad Request", 404: "Not Found",
               405: "Method Not Allowed", 409: "Conflict"}

class FakeHandler:
    def send_response(self, status: int) -> None:
        print(f"状态行: HTTP {status} {STATUS_TEXT.get(status, 'Unknown')}")

    def send_header(self, key: str, value: str) -> None:
        print(f"响应头: {key}: {value}")

    def end_headers(self) -> None:
        print("空行: 头部结束，下面开始响应体")

    def _send(self, status: int, body: Any = None) -> None:
        # 204 这类响应没有 body，只返回状态和 Content-Length: 0。
        if body is None:
            self.send_response(status)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        print("响应体:", payload.decode("utf-8"))

handler = FakeHandler()
handler._send(200, {"data": {"id": 1, "title": "设计 API"}})
print("---")
handler._send(204)

Step 2：用 _dispatch 的正则路由理解 REST URL 设计

痛点与机制：

_dispatch 用 re.fullmatch 匹配 PATH_INFO，提取路径参数。/api/tasks 匹配列表操作，/api/tasks/(\d+) 匹配单条操作，捕获组 ("42",) 就是路径参数。REST URL 的设计原则：URL 表示资源（名词），HTTP 方法表示操作（动词）——GET /api/tasks/42 比 GET /api/getTask?id=42 更符合 REST 风格。

核心源码（逐字来自文末完整源码）：

    def _dispatch(self, method: str) -> None:
        path = self._path()
        query = self._parse_query()

        # GET/POST /api/tasks
        if re.fullmatch(r"/api/tasks/?", path):
            if method == "GET":
                self._list_tasks(query)
            elif method == "POST":
                self._create_task()
            else:
                self._send(405, {"error": "Method Not Allowed"})
            return

        # GET/PUT/PATCH/DELETE /api/tasks/<id>
        m = re.fullmatch(r"/api/tasks/(\d+)/?", path)
        if m:
            task_id = int(m.group(1))
            if method == "GET":
                self._get_task(task_id)
            elif method in ("PUT", "PATCH"):
                self._update_task(task_id, method)
            elif method == "DELETE":
                self._delete_task(task_id)
            else:
                self._send(405, {"error": "Method Not Allowed"})
            return

        self._send(404, {"error": f"路径 {path} 不存在"})

可运行演示（补齐 Mock 数据与 print 反馈）：

import re


def dispatch_preview(method: str, path: str) -> str:
    if re.fullmatch(r"/api/tasks/?", path):
        if method == "GET":
            return "列出任务"
        if method == "POST":
            return "创建任务"
        return "405 方法不允许"

    m = re.fullmatch(r"/api/tasks/(\d+)/?", path)
    if m:
        task_id = int(m.group(1))
        if method == "GET":
            return f"读取任务 #{task_id}"
        if method == "PATCH":
            return f"部分更新任务 #{task_id}"
        if method == "DELETE":
            return f"删除任务 #{task_id}"
        return "405 方法不允许"

    return "404 路径不存在"

for method, path in [("GET", "/api/tasks"), ("POST", "/api/tasks"), ("PATCH", "/api/tasks/2"), ("GET", "/other")]:
    print(f"{method:6} {path:<16} -> {dispatch_preview(method, path)}")

Step 3：用 _list_tasks 实现过滤 + 排序 + 分页三件套

痛点与机制：

_list_tasks 是列表接口的标配：?done=false 过滤、?sort=-created_at 排序（- 前缀表示降序）、?page=1&page_size=10 分页。分页响应里的 total_pages 让前端知道总共有多少页，不需要再发一次 COUNT(*) 查询。min(100, max(1, page_size)) 是防御性编程——限制 page_size 范围，防止客户端请求 page_size=999999 把服务器打垮。

核心源码（逐字来自文末完整源码）：

    def _list_tasks(self, query: Dict) -> None:
        with _lock:
            tasks = list(_db.values())

        # 过滤
        if "done" in query:
            done_val = query["done"].lower() == "true"
            tasks = [t for t in tasks if t["done"] == done_val]
        if "priority" in query:
            tasks = [t for t in tasks if t["priority"] == int(query["priority"])]
        if "q" in query:
            q = query["q"].lower()
            tasks = [t for t in tasks if q in t["title"].lower()]

        # 排序
        sort = query.get("sort", "-created_at")
        desc = sort.startswith("-")
        sort_key = sort.lstrip("-")
        tasks.sort(key=lambda t: t.get(sort_key, ""), reverse=desc)

        # 分页
        page = max(1, int(query.get("page", 1)))
        page_size = min(100, max(1, int(query.get("page_size", 10))))
        total = len(tasks)
        total_pages = max(1, (total + page_size - 1) // page_size)
        start = (page - 1) * page_size
        page_tasks = tasks[start:start + page_size]

        self._send(200, {
            "data": page_tasks,
            "pagination": {
                "page": page, "page_size": page_size,
                "total": total, "total_pages": total_pages,
            }
        })
        print(f"  [GET /api/tasks] 返回 {len(page_tasks)}/{total} 条")

可运行演示（补齐 Mock 数据与 print 反馈）：

from typing import Dict, List

_db: Dict[int, Dict] = {
    1: {"id": 1, "title": "设计 API 文档", "done": False, "priority": 3, "created_at": "2026-04-16T10:00:00"},
    2: {"id": 2, "title": "实现后端接口", "done": False, "priority": 3, "created_at": "2026-04-16T11:00:00"},
    3: {"id": 3, "title": "前端联调", "done": True, "priority": 2, "created_at": "2026-04-16T12:00:00"},
    4: {"id": 4, "title": "写单元测试", "done": False, "priority": 2, "created_at": "2026-04-16T13:00:00"},
    5: {"id": 5, "title": "部署上线", "done": False, "priority": 1, "created_at": "2026-04-16T14:00:00"},
}

def list_tasks(query: Dict[str, str]) -> Dict:
    tasks: List[Dict] = list(_db.values())
    if "done" in query:
        want_done = query["done"].lower() == "true"
        tasks = [task for task in tasks if task["done"] == want_done]
    if "q" in query:
        keyword = query["q"].lower()
        tasks = [task for task in tasks if keyword in task["title"].lower()]

    sort = query.get("sort", "-created_at")
    tasks.sort(key=lambda task: task.get(sort.lstrip("-"), ""), reverse=sort.startswith("-"))

    page = max(1, int(query.get("page", 1)))
    page_size = min(100, max(1, int(query.get("page_size", 10))))
    total = len(tasks)
    start = (page - 1) * page_size
    return {"data": tasks[start:start + page_size], "pagination": {"page": page, "page_size": page_size, "total": total}}

for query in [{}, {"done": "false"}, {"q": "API", "page_size": "2"}]:
    result = list_tasks(query)
    print(f"query={query or '无'} -> total={result['pagination']['total']}, titles={[t['title'] for t in result['data']]}")

Step 4：用 _update_task 区分 PUT 全量替换和 PATCH 部分更新

痛点与机制：

PUT 和 PATCH 是 REST 里最容易混淆的两个方法。PUT 是"全量替换"：就像把一张表单整张重新填写，缺少的字段用默认值填充；PATCH 是"部分更新"：就像只改表单里的某几个格子，其他格子保持不变。_update_task 用 method == "PUT" 分支处理两种情况：PUT 重建整个对象，PATCH 只更新 body 里出现的字段。

核心源码（逐字来自文末完整源码）：

    def _update_task(self, task_id: int, method: str) -> None:
        body = self._read_body()
        if body is None:
            self._send(400, {"error": "请求体必须是合法 JSON"})
            return
        with _lock:
            if task_id not in _db:
                self._send(404, {"error": f"Task {task_id} 不存在"})
                return
            if method == "PUT":
                # 全量替换
                title = body.get("title", "").strip()
                if not title:
                    self._send(400, {"error": "title 不能为空"})
                    return
                _db[task_id] = {
                    "id": task_id, "title": title,
                    "done": bool(body.get("done", False)),
                    "priority": int(body.get("priority", 1)),
                    "created_at": _db[task_id]["created_at"],
                }
            else:
                # PATCH：部分更新
                for k in ("title", "done", "priority"):
                    if k in body:
                        _db[task_id][k] = body[k]
            self._send(200, {"data": _db[task_id]})
        print(f"  [{method} /api/tasks/{task_id}] 更新完成")

可运行演示（补齐 Mock 数据与 print 反馈）：

task = {"id": 1, "title": "设计 API 文档", "done": False, "priority": 3, "created_at": "2026-04-16T10:00:00"}
print("原始任务:", task)

# PUT 是全量替换：没有传的字段会回到默认值。
put_body = {"title": "全量替换标题", "done": True}
put_result = {
    "id": task["id"],
    "title": put_body.get("title", ""),
    "done": bool(put_body.get("done", False)),
    "priority": int(put_body.get("priority", 1)),
    "created_at": task["created_at"],
}
print("PUT 后:", put_result)

# PATCH 是部分更新：只改传入字段，其余字段保持原样。
patch_body = {"done": True}
patch_result = dict(task)
for key in ("title", "done", "priority"):
    if key in patch_body:
        patch_result[key] = patch_body[key]
print("PATCH 后:", patch_result)

Step 5：用 RESTHandler 启动完整 REST API 服务，测试 GET/POST

痛点与机制：

RESTHandler 继承 BaseHTTPRequestHandler，把 do_GET/do_POST/do_PUT/do_PATCH/do_DELETE 都代理到 _dispatch，由 _dispatch 根据路径和方法分发到具体处理函数。_lock 保护 _db 字典的并发写入——多个请求同时到达时，不加锁会导致数据竞争。

核心源码（逐字来自文末完整源码）：

class RESTHandler(http.server.BaseHTTPRequestHandler):
    def log_message(self, *args): pass

    def _send(self, status: int, body: Any = None) -> None:
        if body is None:
            self.send_response(status)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def _read_body(self) -> Optional[Dict]:
        length = int(self.headers.get("Content-Length", 0))
        if not length:
            return {}
        try:
            return json.loads(self.rfile.read(length))
        except json.JSONDecodeError:
            return None

    def _parse_query(self) -> Dict[str, str]:
        parsed = urllib.parse.urlparse(self.path)
        return dict(urllib.parse.parse_qsl(parsed.query))

    def _path(self) -> str:
        return urllib.parse.urlparse(self.path).path

    # ── 路由分发 ──────────────────────────────────────────────────────────────

    def _dispatch(self, method: str) -> None:
        path = self._path()
        query = self._parse_query()

        # GET/POST /api/tasks
        if re.fullmatch(r"/api/tasks/?", path):
            if method == "GET":
                self._list_tasks(query)
            elif method == "POST":
                self._create_task()
            else:
                self._send(405, {"error": "Method Not Allowed"})
            return

        # GET/PUT/PATCH/DELETE /api/tasks/<id>
        m = re.fullmatch(r"/api/tasks/(\d+)/?", path)
        if m:
            task_id = int(m.group(1))
            if method == "GET":
                self._get_task(task_id)
            elif method in ("PUT", "PATCH"):
                self._update_task(task_id, method)
            elif method == "DELETE":
                self._delete_task(task_id)
            else:
                self._send(405, {"error": "Method Not Allowed"})
            return

        self._send(404, {"error": f"路径 {path} 不存在"})

    do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = \
        lambda self: self._dispatch(self.command)

    # ── 业务逻辑 ──────────────────────────────────────────────────────────────

    def _list_tasks(self, query: Dict) -> None:
        with _lock:
            tasks = list(_db.values())

        # 过滤
        if "done" in query:
            done_val = query["done"].lower() == "true"
            tasks = [t for t in tasks if t["done"] == done_val]
        if "priority" in query:
            tasks = [t for t in tasks if t["priority"] == int(query["priority"])]
        if "q" in query:
            q = query["q"].lower()
            tasks = [t for t in tasks if q in t["title"].lower()]

        # 排序
        sort = query.get("sort", "-created_at")
        desc = sort.startswith("-")
        sort_key = sort.lstrip("-")
        tasks.sort(key=lambda t: t.get(sort_key, ""), reverse=desc)

        # 分页
        page = max(1, int(query.get("page", 1)))
        page_size = min(100, max(1, int(query.get("page_size", 10))))
        total = len(tasks)
        total_pages = max(1, (total + page_size - 1) // page_size)
        start = (page - 1) * page_size
        page_tasks = tasks[start:start + page_size]

        self._send(200, {
            "data": page_tasks,
            "pagination": {
                "page": page, "page_size": page_size,
                "total": total, "total_pages": total_pages,
            }
        })
        print(f"  [GET /api/tasks] 返回 {len(page_tasks)}/{total} 条")

    def _get_task(self, task_id: int) -> None:
        with _lock:
            task = _db.get(task_id)
        if task is None:
            self._send(404, {"error": f"Task {task_id} 不存在"})
        else:
            self._send(200, {"data": task})

    def _create_task(self) -> None:
        global _next_id
        body = self._read_body()
        if body is None:
            self._send(400, {"error": "请求体必须是合法 JSON"})
            return
        title = body.get("title", "").strip()
        if not title:
            self._send(400, {"error": "title 不能为空", "fields": {"title": "必填"}})
            return
        with _lock:
            task = {
                "id": _next_id, "title": title,
                "done": bool(body.get("done", False)),
                "priority": int(body.get("priority", 1)),
                "created_at": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
            }
            _db[_next_id] = task
            _next_id += 1
        self._send(201, {"data": task})
        print(f"  [POST /api/tasks] 创建: {task['title']}")

    def _update_task(self, task_id: int, method: str) -> None:
        body = self._read_body()
        if body is None:
            self._send(400, {"error": "请求体必须是合法 JSON"})
            return
        with _lock:
            if task_id not in _db:
                self._send(404, {"error": f"Task {task_id} 不存在"})
                return
            if method == "PUT":
                # 全量替换
                title = body.get("title", "").strip()
                if not title:
                    self._send(400, {"error": "title 不能为空"})
                    return
                _db[task_id] = {
                    "id": task_id, "title": title,
                    "done": bool(body.get("done", False)),
                    "priority": int(body.get("priority", 1)),
                    "created_at": _db[task_id]["created_at"],
                }
            else:
                # PATCH：部分更新
                for k in ("title", "done", "priority"):
                    if k in body:
                        _db[task_id][k] = body[k]
            self._send(200, {"data": _db[task_id]})
        print(f"  [{method} /api/tasks/{task_id}] 更新完成")

    def _delete_task(self, task_id: int) -> None:
        with _lock:
            if task_id not in _db:
                self._send(404, {"error": f"Task {task_id} 不存在"})
                return
            del _db[task_id]
        self._send(204)
        print(f"  [DELETE /api/tasks/{task_id}] 已删除")

可运行演示（补齐 Mock 数据与 print 反馈）：

from datetime import datetime
from typing import Dict, Any

_db: Dict[int, Dict[str, Any]] = {
    1: {"id": 1, "title": "设计 API 文档", "done": False, "priority": 3, "created_at": "2026-04-16T10:00:00"},
    2: {"id": 2, "title": "实现后端接口", "done": False, "priority": 3, "created_at": "2026-04-16T11:00:00"},
}
_next_id = 3

def list_tasks() -> Dict:
    return {"data": list(_db.values()), "pagination": {"total": len(_db)}}

def create_task(body: Dict[str, Any]) -> tuple[int, Dict]:
    global _next_id
    title = body.get("title", "").strip()
    if not title:
        return 400, {"error": "title 不能为空", "fields": {"title": "必填"}}
    task = {"id": _next_id, "title": title, "done": bool(body.get("done", False)),
            "priority": int(body.get("priority", 1)), "created_at": datetime.now().strftime("%Y-%m-%dT%H:%M:%S")}
    _db[_next_id] = task
    _next_id += 1
    return 201, {"data": task}

print("GET /api/tasks ->", list_tasks())
status, created = create_task({"title": "写离线 REST 演示", "priority": 2})
print(f"POST /api/tasks -> {status}, {created}")
print("再次 GET /api/tasks -> total=", list_tasks()["pagination"]["total"])

Step 6：测试 GET/PATCH/DELETE 单条资源操作

痛点与机制：

DELETE 返回 204 No Content——状态码 204 表示"操作成功，但没有响应体"。这是 REST 的约定：删除成功不需要返回被删除的数据，只需要告诉客户端"已删除"。urllib.request.urlopen 在收到 204 时不会抛异常，但 resp.read() 返回空字节串，不能用 json.loads 解析——这是新手常见的坑。

核心源码（逐字来自文末完整源码）：

    def _update_task(self, task_id: int, method: str) -> None:
        body = self._read_body()
        if body is None:
            self._send(400, {"error": "请求体必须是合法 JSON"})
            return
        with _lock:
            if task_id not in _db:
                self._send(404, {"error": f"Task {task_id} 不存在"})
                return
            if method == "PUT":
                # 全量替换
                title = body.get("title", "").strip()
                if not title:
                    self._send(400, {"error": "title 不能为空"})
                    return
                _db[task_id] = {
                    "id": task_id, "title": title,
                    "done": bool(body.get("done", False)),
                    "priority": int(body.get("priority", 1)),
                    "created_at": _db[task_id]["created_at"],
                }
            else:
                # PATCH：部分更新
                for k in ("title", "done", "priority"):
                    if k in body:
                        _db[task_id][k] = body[k]
            self._send(200, {"data": _db[task_id]})
        print(f"  [{method} /api/tasks/{task_id}] 更新完成")

可运行演示（补齐 Mock 数据与 print 反馈）：

from typing import Any, Dict

_db: Dict[int, Dict[str, Any]] = {
    1: {"id": 1, "title": "设计 API 文档", "done": False, "priority": 3, "created_at": "2026-04-16T10:00:00"},
    2: {"id": 2, "title": "实现后端接口", "done": False, "priority": 3, "created_at": "2026-04-16T11:00:00"},
}

def get_task(task_id: int) -> tuple[int, Dict]:
    task = _db.get(task_id)
    return (200, {"data": task}) if task else (404, {"error": f"Task {task_id} 不存在"})

def patch_task(task_id: int, body: Dict[str, Any]) -> tuple[int, Dict]:
    if task_id not in _db:
        return 404, {"error": f"Task {task_id} 不存在"}
    for key in ("title", "done", "priority"):
        if key in body:
            _db[task_id][key] = body[key]
    return 200, {"data": _db[task_id]}

def delete_task(task_id: int) -> tuple[int, Dict | None]:
    if task_id not in _db:
        return 404, {"error": f"Task {task_id} 不存在"}
    del _db[task_id]
    return 204, None

print("GET /api/tasks/1 ->", get_task(1))
print("PATCH /api/tasks/1 ->", patch_task(1, {"done": True}))
print("DELETE /api/tasks/1 ->", delete_task(1))
print("再次 GET /api/tasks/1 ->", get_task(1))

Step 7：用 run_client 发完整的 CRUD 请求序列

痛点与机制：

run_client 按顺序发出一组请求，覆盖完整的 CRUD 流程：创建 → 查询 → 更新 → 删除。这是 REST API 的"冒烟测试"——用最少的请求验证所有核心功能都正常工作。X-Request-ID 请求头是 API 追踪的标准做法，每个请求带一个唯一 ID，方便在日志里追踪一次完整的业务流程。

核心源码（逐字来自文末完整源码）：

def run_client(base: str = "http://127.0.0.1:8003") -> None:
    def req(method: str, path: str, data: Dict = None) -> Tuple[int, Any]:
        payload = json.dumps(data).encode() if data else None
        headers = {"Content-Type": "application/json"} if payload else {}
        r = urllib.request.Request(f"{base}{path}", data=payload,
                                   headers=headers, method=method)
        try:
            with urllib.request.urlopen(r) as resp:
                body = resp.read()
                return resp.status, json.loads(body) if body else None
        except urllib.error.HTTPError as e:
            return e.code, json.loads(e.read())

    print("\n=== RESTful API 客户端测试 ===\n")

    # 列表
    status, body = req("GET", "/api/tasks?page_size=3")
    print(f"[GET /api/tasks?page_size=3] HTTP {status}")
    if body:
        for t in body["data"]:
            print(f"  [{t['id']}] {'✅' if t['done'] else '⬜'} {t['title']}")
        print(f"  分页: {body['pagination']}")

    # 过滤
    status, body = req("GET", "/api/tasks?done=false&sort=-priority")
    print(f"\n[GET ?done=false&sort=-priority] HTTP {status} → {len(body['data'])} 条")

    # 创建
    status, body = req("POST", "/api/tasks", {"title": "写 REST 文章", "priority": 3})
    print(f"\n[POST /api/tasks] HTTP {status} → {body['data']}")
    new_id = body["data"]["id"]

    # 获取单条
    status, body = req("GET", f"/api/tasks/{new_id}")
    print(f"\n[GET /api/tasks/{new_id}] HTTP {status} → {body['data']}")

    # PATCH 部分更新
    status, body = req("PATCH", f"/api/tasks/{new_id}", {"done": True})
    print(f"\n[PATCH /api/tasks/{new_id}] HTTP {status} → done={body['data']['done']}")

    # DELETE
    status, _ = req("DELETE", f"/api/tasks/{new_id}")
    print(f"\n[DELETE /api/tasks/{new_id}] HTTP {status}")

    # 404
    status, body = req("GET", f"/api/tasks/{new_id}")
    print(f"\n[GET /api/tasks/{new_id}（已删除）] HTTP {status} → {body}")

    # 400 验证失败
    status, body = req("POST", "/api/tasks", {"title": ""})
    print(f"\n[POST /api/tasks（空标题）] HTTP {status} → {body}")

可运行演示（补齐 Mock 数据与 print 反馈）：

from datetime import datetime
from typing import Any, Dict

_db: Dict[int, Dict[str, Any]] = {
    1: {"id": 1, "title": "设计 API 文档", "done": False, "priority": 3, "created_at": "2026-04-16T10:00:00"},
}
_next_id = 2

def request(method: str, path: str, body: Dict[str, Any] | None = None) -> tuple[int, Dict | None]:
    global _next_id
    if method == "GET" and path == "/api/tasks":
        return 200, {"data": list(_db.values())}
    if method == "POST" and path == "/api/tasks":
        task = {"id": _next_id, "title": body["title"], "done": False, "priority": body.get("priority", 1), "created_at": datetime.now().isoformat()}
        _db[_next_id] = task; _next_id += 1
        return 201, {"data": task}
    if path.startswith("/api/tasks/"):
        task_id = int(path.rsplit("/", 1)[-1])
        if method == "PATCH":
            _db[task_id].update(body or {})
            return 200, {"data": _db[task_id]}
        if method == "DELETE":
            _db.pop(task_id, None)
            return 204, None
    return 404, {"error": "路径不存在"}

print("=== 离线版 REST 客户端序列 ===")
for call in [
    ("GET", "/api/tasks", None),
    ("POST", "/api/tasks", {"title": "联调 AJAX", "priority": 2}),
    ("PATCH", "/api/tasks/2", {"done": True}),
    ("DELETE", "/api/tasks/2", None),
]:
    status, data = request(*call)
    print(f"{call[0]:6} {call[1]:<14} -> {status}, {data}")

Step 8：用 main 做 server/client/design 三种模式的 CLI 总入口

痛点与机制：

main 用 argparse 做 CLI 入口：--mode server 启动服务器，--mode client 发请求演示，--mode design 打印 REST API 设计规范。show_design 把 URL 设计、状态码约定、响应结构整理成表格，是设计 REST API 时的速查手册。

核心源码（逐字来自文末完整源码）：

def main() -> None:
    parser = argparse.ArgumentParser(description="RESTful API 演示（零依赖）")
    parser.add_argument("--mode", choices=["server", "client", "design"], default="design",
                        help="server=启动服务器, client=客户端测试, design=设计规范表")
    parser.add_argument("--port", type=int, default=8003)
    args = parser.parse_args()

    if args.mode == "server":
        run_server(args.port)
    elif args.mode == "client":
        run_client(f"http://127.0.0.1:{args.port}")
    else:
        show_design()

可运行演示（补齐 Mock 数据与 print 反馈）：

import argparse
import sys

def run_server(port: int = 8003) -> None:
    print(f"server 模式: 将启动 REST API 服务 http://127.0.0.1:{port}")

def run_client(base: str = "http://127.0.0.1:8003") -> None:
    print(f"client 模式: 将测试 {base}/api/tasks 的 GET/POST/PATCH/DELETE")

def show_design() -> None:
    print("design 模式: 展示 GET/POST/PUT/PATCH/DELETE 与资源 URL 的对应关系")

def main() -> None:
    parser = argparse.ArgumentParser(description="RESTful API 设计演示")
    parser.add_argument("--mode", choices=["server", "client", "design"], default="design")
    parser.add_argument("--port", type=int, default=8003)
    args = parser.parse_args()
    if args.mode == "server":
        run_server(args.port)
    elif args.mode == "client":
        run_client(f"http://127.0.0.1:{args.port}")
    else:
        show_design()

for mode in ["design", "server", "client"]:
    sys.argv = ["prog", "--mode", mode, "--port", "18083"]
    print(f">>> python3 23-python-ajax-rest.py --mode {mode}")
    main()

极客实战：完整源码与运行

现在，把上面的积木拼起来，将以下完整代码放进你的编辑器，运行它。先看整体闭环，再回头逐段改参数，你会更容易建立工程直觉。

#!/usr/bin/env python3
"""
23-ajax-rest.py
用 Python 内置 http.server + json 实现完整 RESTful API 服务器
支持 GET / POST / PUT / DELETE，含分页、过滤、排序
配套 urllib 客户端测试脚本

用法：
  python3 23-ajax-rest.py --mode server   # 启动 API 服务器
  python3 23-ajax-rest.py --mode client   # 运行客户端测试
  python3 23-ajax-rest.py --mode design   # 打印 REST 设计规范表
"""

import argparse
import http.server
import json
import re
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

# ─── 数据存储 ─────────────────────────────────────────────────────────────────

_db: Dict[int, Dict] = {
    1: {"id": 1, "title": "设计 API 文档", "done": False, "priority": 3,
        "created_at": "2026-04-16T10:00:00"},
    2: {"id": 2, "title": "实现后端接口", "done": False, "priority": 3,
        "created_at": "2026-04-16T11:00:00"},
    3: {"id": 3, "title": "前端联调",     "done": True,  "priority": 2,
        "created_at": "2026-04-16T12:00:00"},
    4: {"id": 4, "title": "写单元测试",   "done": False, "priority": 2,
        "created_at": "2026-04-16T13:00:00"},
    5: {"id": 5, "title": "部署上线",     "done": False, "priority": 1,
        "created_at": "2026-04-16T14:00:00"},
}
_next_id = 6
_lock = threading.Lock()

# ─── HTTP 工具 ────────────────────────────────────────────────────────────────

STATUS_TEXT = {200: "OK", 201: "Created", 204: "No Content",
               400: "Bad Request", 404: "Not Found",
               405: "Method Not Allowed", 409: "Conflict"}


class RESTHandler(http.server.BaseHTTPRequestHandler):
    def log_message(self, *args): pass

    def _send(self, status: int, body: Any = None) -> None:
        if body is None:
            self.send_response(status)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return
        payload = json.dumps(body, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def _read_body(self) -> Optional[Dict]:
        length = int(self.headers.get("Content-Length", 0))
        if not length:
            return {}
        try:
            return json.loads(self.rfile.read(length))
        except json.JSONDecodeError:
            return None

    def _parse_query(self) -> Dict[str, str]:
        parsed = urllib.parse.urlparse(self.path)
        return dict(urllib.parse.parse_qsl(parsed.query))

    def _path(self) -> str:
        return urllib.parse.urlparse(self.path).path

    # ── 路由分发 ──────────────────────────────────────────────────────────────

    def _dispatch(self, method: str) -> None:
        path = self._path()
        query = self._parse_query()

        # GET/POST /api/tasks
        if re.fullmatch(r"/api/tasks/?", path):
            if method == "GET":
                self._list_tasks(query)
            elif method == "POST":
                self._create_task()
            else:
                self._send(405, {"error": "Method Not Allowed"})
            return

        # GET/PUT/PATCH/DELETE /api/tasks/<id>
        m = re.fullmatch(r"/api/tasks/(\d+)/?", path)
        if m:
            task_id = int(m.group(1))
            if method == "GET":
                self._get_task(task_id)
            elif method in ("PUT", "PATCH"):
                self._update_task(task_id, method)
            elif method == "DELETE":
                self._delete_task(task_id)
            else:
                self._send(405, {"error": "Method Not Allowed"})
            return

        self._send(404, {"error": f"路径 {path} 不存在"})

    do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = \
        lambda self: self._dispatch(self.command)

    # ── 业务逻辑 ──────────────────────────────────────────────────────────────

    def _list_tasks(self, query: Dict) -> None:
        with _lock:
            tasks = list(_db.values())

        # 过滤
        if "done" in query:
            done_val = query["done"].lower() == "true"
            tasks = [t for t in tasks if t["done"] == done_val]
        if "priority" in query:
            tasks = [t for t in tasks if t["priority"] == int(query["priority"])]
        if "q" in query:
            q = query["q"].lower()
            tasks = [t for t in tasks if q in t["title"].lower()]

        # 排序
        sort = query.get("sort", "-created_at")
        desc = sort.startswith("-")
        sort_key = sort.lstrip("-")
        tasks.sort(key=lambda t: t.get(sort_key, ""), reverse=desc)

        # 分页
        page = max(1, int(query.get("page", 1)))
        page_size = min(100, max(1, int(query.get("page_size", 10))))
        total = len(tasks)
        total_pages = max(1, (total + page_size - 1) // page_size)
        start = (page - 1) * page_size
        page_tasks = tasks[start:start + page_size]

        self._send(200, {
            "data": page_tasks,
            "pagination": {
                "page": page, "page_size": page_size,
                "total": total, "total_pages": total_pages,
            }
        })
        print(f"  [GET /api/tasks] 返回 {len(page_tasks)}/{total} 条")

    def _get_task(self, task_id: int) -> None:
        with _lock:
            task = _db.get(task_id)
        if task is None:
            self._send(404, {"error": f"Task {task_id} 不存在"})
        else:
            self._send(200, {"data": task})

    def _create_task(self) -> None:
        global _next_id
        body = self._read_body()
        if body is None:
            self._send(400, {"error": "请求体必须是合法 JSON"})
            return
        title = body.get("title", "").strip()
        if not title:
            self._send(400, {"error": "title 不能为空", "fields": {"title": "必填"}})
            return
        with _lock:
            task = {
                "id": _next_id, "title": title,
                "done": bool(body.get("done", False)),
                "priority": int(body.get("priority", 1)),
                "created_at": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
            }
            _db[_next_id] = task
            _next_id += 1
        self._send(201, {"data": task})
        print(f"  [POST /api/tasks] 创建: {task['title']}")

    def _update_task(self, task_id: int, method: str) -> None:
        body = self._read_body()
        if body is None:
            self._send(400, {"error": "请求体必须是合法 JSON"})
            return
        with _lock:
            if task_id not in _db:
                self._send(404, {"error": f"Task {task_id} 不存在"})
                return
            if method == "PUT":
                # 全量替换
                title = body.get("title", "").strip()
                if not title:
                    self._send(400, {"error": "title 不能为空"})
                    return
                _db[task_id] = {
                    "id": task_id, "title": title,
                    "done": bool(body.get("done", False)),
                    "priority": int(body.get("priority", 1)),
                    "created_at": _db[task_id]["created_at"],
                }
            else:
                # PATCH：部分更新
                for k in ("title", "done", "priority"):
                    if k in body:
                        _db[task_id][k] = body[k]
            self._send(200, {"data": _db[task_id]})
        print(f"  [{method} /api/tasks/{task_id}] 更新完成")

    def _delete_task(self, task_id: int) -> None:
        with _lock:
            if task_id not in _db:
                self._send(404, {"error": f"Task {task_id} 不存在"})
                return
            del _db[task_id]
        self._send(204)
        print(f"  [DELETE /api/tasks/{task_id}] 已删除")


def run_server(port: int = 8003) -> None:
    print(f"🚀 RESTful API 服务器启动：http://127.0.0.1:{port}")
    print("   GET/POST /api/tasks | GET/PUT/PATCH/DELETE /api/tasks/<id>")
    print("   按 Ctrl+C 停止\n")
    try:
        http.server.HTTPServer(("127.0.0.1", port), RESTHandler).serve_forever()
    except KeyboardInterrupt:
        print("\n服务器已停止。")

# ─── 客户端测试 ───────────────────────────────────────────────────────────────

def run_client(base: str = "http://127.0.0.1:8003") -> None:
    def req(method: str, path: str, data: Dict = None) -> Tuple[int, Any]:
        payload = json.dumps(data).encode() if data else None
        headers = {"Content-Type": "application/json"} if payload else {}
        r = urllib.request.Request(f"{base}{path}", data=payload,
                                   headers=headers, method=method)
        try:
            with urllib.request.urlopen(r) as resp:
                body = resp.read()
                return resp.status, json.loads(body) if body else None
        except urllib.error.HTTPError as e:
            return e.code, json.loads(e.read())

    print("\n=== RESTful API 客户端测试 ===\n")

    # 列表
    status, body = req("GET", "/api/tasks?page_size=3")
    print(f"[GET /api/tasks?page_size=3] HTTP {status}")
    if body:
        for t in body["data"]:
            print(f"  [{t['id']}] {'✅' if t['done'] else '⬜'} {t['title']}")
        print(f"  分页: {body['pagination']}")

    # 过滤
    status, body = req("GET", "/api/tasks?done=false&sort=-priority")
    print(f"\n[GET ?done=false&sort=-priority] HTTP {status} → {len(body['data'])} 条")

    # 创建
    status, body = req("POST", "/api/tasks", {"title": "写 REST 文章", "priority": 3})
    print(f"\n[POST /api/tasks] HTTP {status} → {body['data']}")
    new_id = body["data"]["id"]

    # 获取单条
    status, body = req("GET", f"/api/tasks/{new_id}")
    print(f"\n[GET /api/tasks/{new_id}] HTTP {status} → {body['data']}")

    # PATCH 部分更新
    status, body = req("PATCH", f"/api/tasks/{new_id}", {"done": True})
    print(f"\n[PATCH /api/tasks/{new_id}] HTTP {status} → done={body['data']['done']}")

    # DELETE
    status, _ = req("DELETE", f"/api/tasks/{new_id}")
    print(f"\n[DELETE /api/tasks/{new_id}] HTTP {status}")

    # 404
    status, body = req("GET", f"/api/tasks/{new_id}")
    print(f"\n[GET /api/tasks/{new_id}（已删除）] HTTP {status} → {body}")

    # 400 验证失败
    status, body = req("POST", "/api/tasks", {"title": ""})
    print(f"\n[POST /api/tasks（空标题）] HTTP {status} → {body}")

# ─── 设计规范表 ───────────────────────────────────────────────────────────────

def show_design() -> None:
    print(f"\n{'─'*65}")
    print("  RESTful API URL 设计规范")
    print(f"{'─'*65}")
    rows = [
        ("GET",    "/api/tasks",        "200", "获取列表（支持分页/过滤/排序）"),
        ("POST",   "/api/tasks",        "201", "创建资源，返回新建对象"),
        ("GET",    "/api/tasks/42",     "200", "获取单个资源"),
        ("PUT",    "/api/tasks/42",     "200", "全量替换（需传所有字段）"),
        ("PATCH",  "/api/tasks/42",     "200", "部分更新（只传要改的字段）"),
        ("DELETE", "/api/tasks/42",     "204", "删除，无响应体"),
        ("GET",    "/api/tasks/42/tags","200", "嵌套资源列表"),
    ]
    print(f"  {'方法':<8} {'URL':<25} {'状态码':<8} {'说明'}")
    print(f"  {'─'*8} {'─'*25} {'─'*8} {'─'*20}")
    for method, url, code, desc in rows:
        print(f"  {method:<8} {url:<25} {code:<8} {desc}")
    print(f"{'─'*65}")

# ─── 入口 ─────────────────────────────────────────────────────────────────────

def main() -> None:
    parser = argparse.ArgumentParser(description="RESTful API 演示（零依赖）")
    parser.add_argument("--mode", choices=["server", "client", "design"], default="design",
                        help="server=启动服务器, client=客户端测试, design=设计规范表")
    parser.add_argument("--port", type=int, default=8003)
    args = parser.parse_args()

    if args.mode == "server":
        run_server(args.port)
    elif args.mode == "client":
        run_client(f"http://127.0.0.1:{args.port}")
    else:
        show_design()


if __name__ == "__main__":
    main()

$ python3 23-python-ajax-rest.py --mode design

=== RESTful API 设计规范 ===

────────────────────────────────────────────────────────────────────
  方法      URL                    说明
────────────────────────────────────────────────────────────────────
  GET       /api/tasks             获取任务列表（支持过滤/排序/分页）
  POST      /api/tasks             创建新任务
  GET       /api/tasks/{id}        获取单个任务
  PUT       /api/tasks/{id}        全量替换任务
  PATCH     /api/tasks/{id}        部分更新任务
  DELETE    /api/tasks/{id}        删除任务（204 No Content）
────────────────────────────────────────────────────────────────────

$ python3 23-python-ajax-rest.py --mode client

=== REST API 客户端演示 ===

  ✅ POST /api/tasks → 201 {"data": {"id": 6, "title": "学习 REST API", ...}}
  ✅ GET  /api/tasks → 200 {"data": [...], "pagination": {"total": 6, ...}}
  ✅ GET  /api/tasks/6 → 200 {"data": {"id": 6, "title": "学习 REST API", ...}}
  ✅ PATCH /api/tasks/6 → 200 {"data": {"id": 6, "done": true, ...}}
  ✅ DELETE /api/tasks/6 → 204
  ✅ GET  /api/tasks?done=false → 200 {"data": [...], "pagination": {"total": 5}}

小结

概念 一句话记忆
REST URL 名词表示资源，动词用 HTTP 方法，不要 /getTask 这种动词 URL
GET 只读，幂等，可缓存
POST 创建，非幂等，返回 201 Created
PUT 全量替换，幂等，缺少字段用默认值
PATCH 部分更新，只改提供的字段
DELETE 删除，返回 204 No Content（空 body）
{"data": ...} 成功响应的标准结构
{"error": "..."} 错误响应的标准结构
分页 pagination total/page/page_size/total_pages 四个字段
_lock 保护共享数据的并发写入，防止数据竞争

⏱ NexDo Time（5 分钟）

挑战：给 REST API 加一个 GET /api/tasks/search?q=关键词 搜索接口。

具体步骤：

  1. 在 _dispatch 里加一条路由：re.fullmatch(r"/api/tasks/search/?", path) → _search_tasks
  2. 实现 _search_tasks(self, query) 方法：从 query["q"] 取关键词，在 _db 里做标题的大小写不敏感包含匹配
  3. 返回 {"data": [...], "pagination": {"total": len(results)}} 格式
  4. 用 curl "http://127.0.0.1:8002/api/tasks/search?q=API" 验证

Don’t wait for next time, do it in the next moment.