POST请求后Web Server解析JSON及校验响应实现问题

问题概述 #

当前开发的简易Web Server需实现以下流程：用户输入信息更新JSON文件，通过POST请求发送至服务器；服务器解析JSON中的Asset ID，与硬编码模拟数据库校验后返回响应。开发中遇到以下问题：

1. 如何让Web Server解析POST请求传递的JSON数据？
2. 如何在Web Service中硬编码模拟数据库用于校验？
3. 如何执行JSON中Asset ID与模拟数据库的校验逻辑？
4. 校验完成后如何向用户返回响应？
   此外，POST测试代码出现JSONDecodeError，需一并修复。

问题逐一解决 #

1. 解析POST请求的JSON数据 #

服务器端需先读取请求的内容长度，再读取请求体的字节数据，最后用json.loads()解析为Python字典。同时要处理解析失败的异常，返回错误响应。关键步骤：

• 获取Content-Length头，确定请求体大小
• 用self.rfile.read()读取请求体
• 用json.loads()解析，捕获JSONDecodeError处理非法JSON

2. 硬编码模拟数据库 #

直接在Web Server代码中定义全局字典作为模拟数据库，也可作为类属性。这样服务器启动时即可加载，无需额外文件依赖。

3. Asset ID与模拟数据库的校验逻辑 #

• 从解析后的JSON中提取Asset ID字段
• 将字符串类型的Asset ID转换为整数（匹配数据库的键类型）
• 使用字典的get()方法查询数据库，不存在则返回默认值"Not Found"
• 处理Asset ID缺失、格式错误等异常情况，返回对应错误信息

4. 返回校验结果响应 #

构造包含校验状态、结果的JSON响应，设置正确的Content-Type: application/json头，发送HTTP状态码后，将JSON转为字节写入响应体。

修复POST测试代码的JSONDecodeError #

原测试代码的问题：

• 使用data参数发送请求，默认以表单格式传递，服务器解析为JSON会失败；需改用json参数，自动设置Content-Type为application/json
• 更新JSON文件时未截断旧内容，可能导致文件残留数据引发解析错误
• 未处理文件不存在的情况

修改后完整代码 #

Web Server代码 #

#! /usr/bin/python3 

from http.server import BaseHTTPRequestHandler, HTTPServer
import json

hostName = "localhost"
serverPort = 8080

# 硬编码模拟数据库
sample_database = {
    1: "In Calibration",
    2: "Out of Calibration",
    3: "Not Found"
}

class MyServer(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-type", "text/html")
        self.end_headers()
        self.wfile.write(bytes("<html><head><title>Local Web Server</title></head>", "utf-8"))
        self.wfile.write(bytes("<p>Request: %s</p>" % self.path, "utf-8"))
        self.wfile.write(bytes("<body>", "utf-8"))
        self.wfile.write(bytes("<p>This is a simple web server for asset validation.</p>", "utf-8"))
        self.wfile.write(bytes("</body></html>", "utf-8"))

    def do_POST(self):
        # 解析POST请求的JSON数据
        content_length = int(self.headers['Content-Length'])
        post_data = self.rfile.read(content_length)
        try:
            request_json = json.loads(post_data)
        except json.JSONDecodeError:
            # 处理非法JSON
            self.send_response(400)
            self.send_header("Content-type", "application/json")
            self.end_headers()
            error_resp = {"status": "error", "message": "Invalid JSON data"}
            self.wfile.write(json.dumps(error_resp).encode('utf-8'))
            return

        # 提取并校验Asset ID
        asset_id_str = request_json.get("Asset ID")
        if not asset_id_str:
            self.send_response(400)
            self.send_header("Content-type", "application/json")
            self.end_headers()
            error_resp = {"status": "error", "message": "Asset ID is missing"}
            self.wfile.write(json.dumps(error_resp).encode('utf-8'))
            return

        try:
            asset_id = int(asset_id_str)
        except ValueError:
            self.send_response(400)
            self.send_header("Content-type", "application/json")
            self.end_headers()
            error_resp = {"status": "error", "message": "Asset ID must be an integer"}
            self.wfile.write(json.dumps(error_resp).encode('utf-8'))
            return

        # 查询模拟数据库
        validation_result = sample_database.get(asset_id, "Not Found")

        # 返回响应
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        response = {
            "status": "success",
            "Asset ID": asset_id,
            "validation_result": validation_result,
            "received_data": request_json
        }
        self.wfile.write(json.dumps(response).encode('utf-8'))

if __name__ == "__main__":        
    webServer = HTTPServer((hostName, serverPort), MyServer)
    print("Server started http://%s:%s" % (hostName, serverPort))

    try:
        webServer.serve_forever()
    except KeyboardInterrupt:
        pass

    webServer.server_close()
    print("Server stopped.")

POST测试代码 #

import requests as req
import json

material = input('Please enter the material: ')
asset_id = input('Please enter asset id: ')

# 构造更新数据
update_data = {
    "Date": "8-22-22",
    "Material": material,
    "Asset ID": asset_id
}

# 更新JSON文件
try:
    with open('sample.json', 'r+') as f:
        try:
            existing_data = json.load(f)
        except json.JSONDecodeError:
            existing_data = {}
        existing_data.update(update_data)
        f.seek(0)
        json.dump(existing_data, f, indent=4)
        f.truncate()
except FileNotFoundError:
    # 文件不存在则创建
    with open('sample.json', 'w') as f:
        json.dump(update_data, f, indent=4)

# 读取更新后的文件并发送POST请求
with open('sample.json', 'r') as f:
    send_data = json.load(f)

# 使用json参数发送请求，自动设置正确的Content-Type
resp = req.post('http://localhost:8080/', json=send_data)

# 处理响应
try:
    response_data = resp.json()
    print("Server Response:")
    print(json.dumps(response_data, indent=4))
except json.JSONDecodeError:
    print("Failed to parse response:", resp.text)

内容的提问来源于stack exchange，提问作者Jaheen