Python 现代开发学习笔记

(五)解析正文并形成明确的响应

1. 先判断格式,再转换字段

继续在 Request 中加入两个方法:

# 续写同一文件:web_project/src/webbook/http.py
    # 先按 UTF-8 解码字节,再核对 Content-Type;解析失败通过 Problem 转成 HTTP 错误。
    def text(self):
        try:
            return self.body.decode("utf-8")
        except UnicodeDecodeError as error:
            raise Problem(400, "请求体不是UTF-8文本") from error

    def expect_type(self, expected):
        actual = self.headers.get("content-type", "").split(";", 1)[0].strip().lower()
        if actual != expected:
            raise Problem(415, f"本入口需要 {expected}")

text 将本次正文按 UTF-8 解码;expect_type 取 Content-Type 的媒体类型部分,不把后面的 charset 参数当作格式名字的一部分。正文无法解码与媒体类型不符,分别以明确状态交回。

随后在同一个 http.py 中加入表单与 JSON 的读取方法:

# 续写同一文件:web_project/src/webbook/http.py
# 表单和 JSON 分别解析,再把单值字段取出来;重复提交同名字段时返回明确错误。
def one_value(values, key, default=""):
    items = values.get(key, [default])
    if len(items) != 1:
        raise Problem(422, f"{key} 重复出现")
    return items[0]


def read_form(request):
    request.expect_type("application/x-www-form-urlencoded")
    return parse_qs(request.text(), keep_blank_values=True)


def read_json(request):
    request.expect_type("application/json")
    try:
        value = json.loads(request.text())
    except ValueError as error:
        raise Problem(400, "JSON文本结构错误") from error
    if not isinstance(value, dict):
        raise Problem(422, "书籍数据应为JSON对象")
    return value

one_value 取单值字段,并拒绝重复的同名值。JSON 文本结构错误返回 400;JSON 能解析但得到列表而非书籍对象时返回 422。解析得到一个 Python 对象与这个对象符合书籍字段要求是两个步骤,下一节还会继续核对具体字段。

2. 响应先编码再计算长度

# 续写同一文件:web_project/src/webbook/http.py
# 文本先编码成字节,Content-Length 按字节数计算;WSGI 响应返回字节块列表。
@dataclass
class Response:
    status: int
    body: bytes = b""
    headers: list = field(default_factory=list)

    @classmethod
    def text(cls, text, status=200, kind="text/html; charset=utf-8", headers=None):
        return cls(
            status, text.encode("utf-8"), [("Content-Type", kind), *(headers or [])]
        )

    @classmethod
    def json(cls, value, status=200, headers=None):
        return cls.text(
            json.dumps(value, ensure_ascii=False, allow_nan=False),
            status,
            "application/json; charset=utf-8",
            headers,
        )

    @classmethod
    def redirect(cls, location):
        return cls(303, headers=[("Location", location)])

    def send(self, start_response):
        headers = [
            *self.headers,
            ("Content-Length", str(len(self.body))),
            ("Cache-Control", "no-store"),
            ("X-Content-Type-Options", "nosniff"),
        ]
        start_response(f"{self.status} {HTTPStatus(self.status).phrase}", headers)
        return [self.body]

Response.text 将文本编码为字节,Response.json 先序列化数据再走相同的编码过程;redirect 交回 303 与 Location。send 最后计算正文的字节数,把状态和头交给服务器,再返回正文。

HTTPStatus 根据整数提供状态的英文短语。Cache-Control 的 no-store 告诉客户端不缓存这些动态响应,nosniff 让浏览器按声明的内容类型处理。默认列表通过 default_factory 为每个响应分别建立,避免不同请求共享同一份头字段列表。

笔记正文与全部题解 · Ctrl / ⌘ K