WSGI应用接收JSON Payload触发AssertionError，寻求技术指导

搭建基于Python + WSGI的应用，用于接收云端JSON格式Payload并最终发送邮件，环境搭配NGINX。测试时用curl发送JSON数据触发AssertionError，核心是JSON数据捕获失败。

代码片段 #

from wsgiref.simple_server import make_server

import cgi
import subprocess
from email.message import EmailMessage
import json

def sendEmail(from_addr, to_addrs, msg_subject, msg_body):
    msg = EmailMessage()
    msg.set_content(msg_body)
    msg['From'] = from_addr
    msg['To'] = to_addrs
    msg['Subject'] = msg_subject
    sendmail_location = "/usr/sbin/sendmail"
    subprocess.run([sendmail_location, "-t", "-oi"], input=msg.as_bytes())

def is_post_request(environ):
    if environ['REQUEST_METHOD'].upper() != 'POST':
        return False
    content_type = environ.get('CONTENT_TYPE', 'application/x-www-form-urlencoded')
    print("I am printing here " + content_type)
    return (content_type.startswith('application/x-www-form-urlencoded' or content_type.startswith('multipart/form-data' or content_type.startswith('application/json'))))

def get_post_form(environ):
    print ("Printing environ" + str(environ))
    assert is_post_request(environ)
    input = environ['wsgi.input']
    post_form = environ.get('wsgi.post_form')
    if (post_form is not None
        and post_form[0] is input):
        return post_form[2]
    # This must be done to avoid a bug in cgi.FieldStorage
    environ.setdefault('QUERY_STRING', '')
    fs = cgi.FieldStorage(fp=input,
                          environ=environ,
                          keep_blank_values=1)
    new_input = InputProcessed()
    post_form = (new_input, input, fs)
    environ['wsgi.post_form'] = post_form
    environ['wsgi.input'] = new_input
    return fs

class InputProcessed(object):
    def read(self, *args):
        raise EOFError('The wsgi.input stream has already been consumed')
    readline = readlines = __iter__ = read


def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/html')])
    user = get_post_form(environ).getvalue('user')
    password = get_post_form(environ).getvalue('password')
    output = 'user is: '+user+' and password is: '+password
    return [output.encode()]

测试命令 #

curl --header "Content-Type: application/json" --request POST --data '{"user":"abc","password":"xyz"}' http://localhost:8580

错误信息 #

Printing environ{'QUERY_STRING': '', 'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': 'application/json', 'CONTENT_LENGTH': '31', 'REQUEST_URI': '/', 'PATH_INFO': '/', 'DOCUMENT_ROOT': '/usr/share/nginx/html', 'SERVER_PROTOCOL': 'HTTP/1.1', 'REQUEST_SCHEME': 'http', 'REMOTE_ADDR': '127.0.0.1', 'REMOTE_PORT': '43534', 'SERVER_PORT': '8580', 'SERVER_NAME': 'p0001.emea.net', 'HTTP_HOST': 'localhost:8580', 'HTTP_USER_AGENT': 'curl/7.61.1', 'HTTP_ACCEPT': '*/*', 'HTTP_CONTENT_TYPE': 'application/json', 'HTTP_CONTENT_LENGTH': '31', 'wsgi.input': <uwsgi._Input object at 0x7f34812cfb28>, 'wsgi.file_wrapper': <built-in function uwsgi_sendfile>, 'wsgi.version': (1, 0), 'wsgi.errors': <_io.TextIOWrapper name=2 mode='w' encoding='UTF-8'>, 'wsgi.run_once': False, 'wsgi.multithread': False, 'wsgi.multiprocess': True, 'wsgi.url_scheme': 'http', 'uwsgi.version': b'2.0.26', 'uwsgi.node': b'p0001.emea.net'}
I am printing here application/json
Traceback (most recent call last):
  File "./wsgi.py", line 61, in application
    user = get_post_form(environ).getvalue('user')
  File "./wsgi.py", line 36, in get_post_form
    assert is_post_request(environ)
AssertionError
[pid: 379691|app: 0|req: 1/1] 127.0.0.1 () {36 vars in 463 bytes} [Wed Jul 10 10:04:26 2024] POST / => generated 0 bytes in 1 msecs (HTTP/1.1 200) 1 headers in 44 bytes (0 switches on core 0)

1. 修复is_post_request函数逻辑错误 #

原函数中return语句的逻辑嵌套错误导致application/json类型无法被识别，正确写法需对三种Content-Type分别做判断：

def is_post_request(environ):
    if environ['REQUEST_METHOD'].upper() != 'POST':
        return False
    content_type = environ.get('CONTENT_TYPE', 'application/x-www-form-urlencoded')
    print("I am printing here " + content_type)
    return (content_type.startswith('application/x-www-form-urlencoded') 
            or content_type.startswith('multipart/form-data') 
            or content_type.startswith('application/json'))

2. 新增JSON数据解析逻辑 #

cgi.FieldStorage无法处理JSON格式请求，需单独读取请求体并解析。修改原get_post_form函数为get_post_data，分支处理不同类型数据：

def get_post_data(environ):
    print ("Printing environ" + str(environ))
    assert is_post_request(environ)
    
    content_type = environ.get('CONTENT_TYPE', '')
    input_stream = environ['wsgi.input']
    
    # 检查是否已处理过请求体
    post_data = environ.get('wsgi.post_data')
    if post_data is not None:
        return post_data
    
    if content_type.startswith('application/json'):
        # 解析JSON数据
        content_length = int(environ.get('CONTENT_LENGTH', 0))
        json_data = json.loads(input_stream.read(content_length))
        environ['wsgi.post_data'] = json_data
        return json_data
    else:
        # 处理表单数据
        environ.setdefault('QUERY_STRING', '')
        fs = cgi.FieldStorage(fp=input_stream,
                              environ=environ,
                              keep_blank_values=1)
        environ['wsgi.post_data'] = fs
        # 标记输入流已处理
        environ['wsgi.input'] = InputProcessed()
        return fs

3. 修改application函数适配多数据类型 #

根据返回的数据类型（字典或FieldStorage）获取字段值，并预留邮件调用逻辑：

def application(environ, start_response):
    start_response('200 OK', [('Content-Type', 'text/html')])
    post_data = get_post_data(environ)
    
    if isinstance(post_data, dict):
        # JSON格式数据
        user = post_data.get('user')
        password = post_data.get('password')
    else:
        # 表单格式数据
        user = post_data.getvalue('user')
        password = post_data.getvalue('password')
    
    # 根据payload中的method字段调用邮件发送函数
    if post_data.get('method') == 'email':
        # 替换为实际的发件人、收件人信息
        sendEmail('sender@example.com', 'recipient@example.com', 'Test Notification', f'User: {user}\nPassword: {password}')
    
    output = f'user is: {user} and password is: {password}'
    return [output.encode()]

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