141 lines
5.4 KiB
Python
141 lines
5.4 KiB
Python
"""第5-6课标准示例：启动本地HTTP服务并自动调用图书接口。"""
|
|
|
|
import json
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from threading import Thread
|
|
from urllib.error import HTTPError
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
class BookApiHandler(BaseHTTPRequestHandler):
|
|
"""处理本课图书查询和新增请求的最小HTTP处理器。"""
|
|
|
|
# 本课不连接数据库，使用类属性保存当前程序运行期间的图书。
|
|
# 程序退出后这份列表会消失，再次运行时会重新从一条图书开始。
|
|
books = [{"id": 1, "title": "Python入门"}]
|
|
|
|
def send_json(
|
|
self,
|
|
status: HTTPStatus,
|
|
data: dict | list,
|
|
extra_headers: dict | None = None,
|
|
) -> None:
|
|
"""把Python对象转换为UTF-8 JSON响应。"""
|
|
|
|
# HTTP连接传输的是字节，因此先生成JSON字符串，再编码成UTF-8字节。
|
|
response_body = json.dumps(data, ensure_ascii=False).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
# Content-Length必须填写字节数量，不能直接计算中文字符串的字符数。
|
|
self.send_header("Content-Length", str(len(response_body)))
|
|
if extra_headers is not None:
|
|
for header_name, header_value in extra_headers.items():
|
|
self.send_header(header_name, header_value)
|
|
self.end_headers()
|
|
self.wfile.write(response_body)
|
|
|
|
def do_GET(self) -> None:
|
|
"""处理GET /books，并拒绝未知路径。"""
|
|
|
|
# 标准库收到GET请求后会自动调用本方法，self.path保存请求目标。
|
|
if self.path == "/books":
|
|
self.send_json(HTTPStatus.OK, {"items": self.books})
|
|
return
|
|
self.send_json(HTTPStatus.NOT_FOUND, {"message": "资源不存在"})
|
|
|
|
def do_POST(self) -> None:
|
|
"""处理POST /books，校验JSON并创建图书。"""
|
|
|
|
if self.path != "/books":
|
|
self.send_json(HTTPStatus.NOT_FOUND, {"message": "资源不存在"})
|
|
return
|
|
|
|
try:
|
|
# Content-Length告诉服务器本次请求体一共有多少字节。
|
|
content_length = int(self.headers.get("Content-Length", "0"))
|
|
request_body = self.rfile.read(content_length).decode("utf-8")
|
|
request_data = json.loads(request_body)
|
|
except (UnicodeDecodeError, json.JSONDecodeError):
|
|
self.send_json(HTTPStatus.BAD_REQUEST, {"message": "JSON格式错误"})
|
|
return
|
|
|
|
title = request_data.get("title")
|
|
if not isinstance(title, str) or not title.strip():
|
|
self.send_json(
|
|
HTTPStatus.UNPROCESSABLE_CONTENT,
|
|
{"message": "书名不能为空"},
|
|
)
|
|
return
|
|
|
|
# 本课用列表长度模拟生成编号；真实项目通常由数据库生成主键。
|
|
book = {"id": len(self.books) + 1, "title": title.strip()}
|
|
self.books.append(book)
|
|
self.send_json(
|
|
HTTPStatus.CREATED,
|
|
book,
|
|
{"Location": f"/books/{book['id']}"},
|
|
)
|
|
|
|
def log_message(self, format_text: str, *args: object) -> None:
|
|
"""关闭标准库默认访问日志，让教学输出保持清晰。"""
|
|
|
|
|
|
def send_request(
|
|
url: str,
|
|
method: str = "GET",
|
|
data: dict | None = None,
|
|
) -> None:
|
|
"""作为客户端发送HTTP请求并输出响应。"""
|
|
|
|
request_body = None
|
|
headers = {"Accept": "application/json"}
|
|
if data is not None:
|
|
request_body = json.dumps(data, ensure_ascii=False).encode("utf-8")
|
|
headers["Content-Type"] = "application/json; charset=utf-8"
|
|
|
|
# Request只负责描述请求；urlopen()执行时才真正连接服务器并发送请求。
|
|
request = Request(url, data=request_body, headers=headers, method=method)
|
|
try:
|
|
with urlopen(request, timeout=3) as response:
|
|
response_text = response.read().decode("utf-8")
|
|
print(f"{method} {request.full_url} -> {response.status}")
|
|
print(f"响应体：{response_text}")
|
|
except HTTPError as error:
|
|
response_text = error.read().decode("utf-8")
|
|
print(f"{method} {request.full_url} -> {error.code}")
|
|
print(f"响应体：{response_text}")
|
|
|
|
|
|
def main() -> None:
|
|
"""在随机本地端口启动服务，完成调用后安全关闭。"""
|
|
|
|
# 127.0.0.1限制为本机访问；端口0表示让操作系统选择可用端口。
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), BookApiHandler)
|
|
# 服务器需要持续等待请求，因此放入后台线程；主线程继续充当客户端。
|
|
server_thread = Thread(target=server.serve_forever, daemon=True)
|
|
server_thread.start()
|
|
host, port = server.server_address
|
|
base_url = f"http://{host}:{port}"
|
|
print(f"本地服务已启动：{base_url}")
|
|
|
|
try:
|
|
send_request(f"{base_url}/books")
|
|
send_request(
|
|
f"{base_url}/books",
|
|
method="POST",
|
|
data={"title": "HTTP接口实践"},
|
|
)
|
|
send_request(f"{base_url}/books")
|
|
send_request(f"{base_url}/missing")
|
|
finally:
|
|
# 即使某次请求失败，也必须释放监听端口并等待后台线程退出。
|
|
server.shutdown()
|
|
server.server_close()
|
|
server_thread.join()
|
|
print("本地服务已关闭。")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|