如何在Python版AWS Lambda中获取文件上传表单的POST数据

你的当前代码仅实现了返回表单的逻辑，未处理POST提交后的multipart/form-data解析。要通过Lambda Function URL获取上传的二进制文件，需按以下方式修改代码：

1. 区分GET/POST请求 #

Lambda Function URL的请求方法可从event['requestContext']['http']['method']获取，我们需要：

• GET请求：返回文件上传表单
• POST请求：解析上传的文件数据

2. 解析Multipart表单数据 #

提交multipart/form-data时，Function URL会将请求体以Base64编码形式存入event['body']，且event['isBase64Encoded']为True。需完成以下操作：

• 解码Base64得到二进制请求体
• 从请求头的Content-Type中提取分隔符（boundary）
• 使用Python标准库解析multipart数据

修改后的完整代码 #

import json
import os
import logging
from xml.etree import ElementTree as ET
import cgi
import base64

logger = logging.getLogger()
logger.setLevel(logging.INFO)

def lambda_handler(event, context):
    # 获取请求方法
    http_method = event['requestContext']['http']['method']
    
    if http_method == 'GET':
        # 返回文件上传表单
        div = ET.Element('div')
        p = ET.SubElement(div, 'p')
        p.text = '文件上传测试'
        form = ET.SubElement(p, 'form', attrib={'method': 'POST', 'enctype': 'multipart/form-data', 'action': '/'})
        ET.SubElement(form, 'input', attrib={'type': 'file', 'name': 'filename'})
        ET.SubElement(form, 'input', attrib={'type': 'submit', 'value': '上传文件'})
        return {
            'statusCode': 200,
            'headers': {'Content-Type': 'text/html'},
            'body': ET.tostring(div, encoding='utf8', method='html').decode('utf8')
        }
    
    elif http_method == 'POST':
        try:
            # 获取请求头信息
            headers = event['headers']
            content_type = headers.get('content-type') or headers.get('Content-Type')
            
            if not content_type or 'multipart/form-data' not in content_type:
                return {
                    'statusCode': 400,
                    'body': '请使用multipart/form-data格式上传文件'
                }
            
            # 提取multipart分隔符
            boundary = cgi.parse_header(content_type)[1]['boundary']
            # 解码Base64请求体
            body_bytes = base64.b64decode(event['body']) if event['isBase64Encoded'] else event['body'].encode('utf-8')
            
            # 解析表单数据
            form_data = cgi.FieldStorage(
                fp=body_bytes,
                headers={'content-type': content_type},
                environ={'REQUEST_METHOD': 'POST'}
            )
            
            # 获取上传的文件对象
            file_item = form_data.get('filename')
            if not file_item:
                return {
                    'statusCode': 400,
                    'body': '未检测到上传的文件'
                }
            
            # 提取文件名和二进制内容
            filename = file_item.filename
            file_content = file_item.file.read()
            
            # 示例：将文件保存到Lambda临时目录/tmp
            save_path = f'/tmp/{filename}'
            with open(save_path, 'wb') as f:
                f.write(file_content)
            
            logger.info(f'文件 {filename} 已保存至 {save_path}')
            logger.info(f'/tmp目录当前内容: {os.listdir("/tmp")}')
            
            # 返回上传成功提示
            return {
                'statusCode': 200,
                'headers': {'Content-Type': 'text/html'},
                'body': f'<p>文件 {filename} 上传成功！文件大小：{len(file_content)} 字节</p>'
            }
        
        except Exception as e:
            logger.error(f'文件上传处理失败: {str(e)}')
            return {
                'statusCode': 500,
                'body': f'上传失败：{str(e)}'
            }
    
    # 处理非GET/POST请求
    return {
        'statusCode': 405,
        'body': '仅支持GET和POST请求'
    }

关键说明 #

• 使用Python标准库cgi解析multipart数据，无需额外安装依赖，适配Lambda运行环境
• 需严格判断event['isBase64Encoded']，确保请求体解码正确
• Lambda的/tmp目录提供512MB临时存储空间，若需持久化文件，可添加逻辑上传至S3
• 注意Lambda执行时间限制（最长15分钟）和Function URL的payload上限（最大6MB），避免大文件上传超时或失败

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