如何使用email.message模块解析HTTP POST请求？

自Python 3.11起，标准库cgi被弃用，官方推荐使用WSGI替代，其中FieldStorage类的功能可由email.message模块或第三方multipart库替代，用于处理POST/PUT请求的multipart数据。但直接使用email.parser解析HTTP POST请求时容易出现问题，比如以下代码会在EmailBytesParser.parse()处挂起：

import wsgiref.simple_server
import email.parser

import logging

_EmailBytesParser = email.parser.BytesParser()

def main_wsgi(environ, start_response):
    if environ['REQUEST_METHOD'] == 'GET':
        start_response('200 OK', [('Content-Type', 'text/html')])
        yield '<form method="POST" enctype="multipart/form-data"><label for="username">username:</label><input name="username" value="user123" /><br /><label for="avatar">avatar: </label><input name="avatar" type="file"><br /><input type="submit" value="register" />'.encode()
    else:
        assert environ['REQUEST_METHOD'] == 'POST'
        assert environ['CONTENT_TYPE'].startswith('multipart/form-data')
        start_response('200 OK', [('Content-Type', 'text/plain')])
        logging.debug("About to start parsing form data...")
        # body = environ['wsgi.input'].read(int(environ['CONTENT_LENGTH']))
        body = _EmailBytesParser.parse(environ['wsgi.input'], headersonly=True)
        logging.debug("Form data parsed!")
        yield repr(body).encode()

if __name__ == '__main__':
    logging.basicConfig()
    logging.getLogger().setLevel(logging.DEBUG)
    wsgiref.simple_server.make_server('', 8000, main_wsgi).serve_forever()

代码存在的问题 #

• 输入流读取未限制长度：BytesParser.parse()默认会持续读取输入流直到EOF，但WSGI的wsgi.input是一个可迭代对象，若不指定读取长度，它会一直等待数据，导致挂起。必须先通过CONTENT_LENGTH读取固定长度的请求体内容。
• 错误使用headersonly=True：该参数用于只解析邮件头部，而HTTP multipart请求的内容是完整的multipart消息体，并非邮件头部，此参数完全不适用。
• 未传递multipart边界符：email模块解析multipart内容需要知道boundary值，这个值包含在请求的CONTENT_TYPE头中，必须提取后传给解析器。

修正后的代码 #

import wsgiref.simple_server
import email
import email.policy
import logging
import re

def extract_boundary(content_type):
    # 从Content-Type中提取boundary参数
    match = re.search(r'boundary=([^;]+)', content_type)
    if not match:
        raise ValueError("No boundary found in Content-Type")
    return match.group(1).strip('"')

def main_wsgi(environ, start_response):
    if environ['REQUEST_METHOD'] == 'GET':
        start_response('200 OK', [('Content-Type', 'text/html')])
        yield '<form method="POST" enctype="multipart/form-data"><label for="username">username:</label><input name="username" value="user123" /><br /><label for="avatar">avatar: </label><input name="avatar" type="file"><br /><input type="submit" value="register" />'.encode()
    else:
        assert environ['REQUEST_METHOD'] == 'POST'
        content_type = environ['CONTENT_TYPE']
        assert content_type.startswith('multipart/form-data')
        
        start_response('200 OK', [('Content-Type', 'text/plain')])
        logging.debug("About to start parsing form data...")
        
        # 1. 读取完整的请求体
        content_length = int(environ.get('CONTENT_LENGTH', 0))
        body_data = environ['wsgi.input'].read(content_length)
        
        # 2. 提取boundary
        boundary = extract_boundary(content_type)
        
        # 3. 使用email模块解析multipart数据
        msg = email.message_from_bytes(
            body_data,
            policy=email.policy.default,
            _class=email.message.EmailMessage
        )
        
        # 4. 处理解析后的字段
        result = []
        for part in msg.walk():
            if part.is_multipart():
                continue
            name = part.get_param('name', header='Content-Disposition')
            if name:
                if part.get_content_maintype() == 'text':
                    value = part.get_content()
                    result.append(f"{name}: {value}")
                else:
                    # 处理文件上传
                    filename = part.get_filename()
                    file_content = part.get_content()
                    result.append(f"{name} (file: {filename}, size: {len(file_content)} bytes)")
        
        logging.debug("Form data parsed!")
        yield '\n'.join(result).encode()

if __name__ == '__main__':
    logging.basicConfig()
    logging.getLogger().setLevel(logging.DEBUG)
    wsgiref.simple_server.make_server('', 8000, main_wsgi).serve_forever()

说明 #

• 先通过CONTENT_LENGTH读取完整的请求体，避免输入流阻塞。
• 从Content-Type头中提取boundary，确保email模块能正确分割multipart部分。
• 使用email.message_from_bytes结合默认策略解析数据，遍历每个part获取表单字段和文件内容。

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