AWS Lambda（Python）解析multipart/form-data字段与文件报错求助

你的代码存在几个关键问题，导致multipart/form-data解析失败，以下是问题分析和修正后的完整代码：

问题点梳理 #

• 缺失BytesIO导入：代码中使用BytesIO但未在头部导入，会触发NameError
• 变量名不匹配：解析得到的表单数据存在form_values和form_files变量中，但后续代码错误使用了form和files
• 二进制文件错误解码：图片是二进制数据，调用.decode()会损坏文件内容，直接读取二进制即可
• 未传递Content-Type头：parse_form_data需要从请求头中获取multipart的boundary参数，否则无法正确拆分表单字段和文件
• Cognito组判断逻辑错误：Cognito的cognito:groups是数组类型，不能直接与字符串"Admin"做相等比较

修正后的完整代码 #

import json
import boto3
import logging
from werkzeug.formparser import parse_form_data
from werkzeug.datastructures import FileStorage
from werkzeug.wrappers import Request
import base64
import jwt
import os
import uuid
from io import BytesIO  # 新增导入BytesIO

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

def handler(event, context):

    headers = {
        "Access-Control-Allow-Headers": "*",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "*"
    }

    body = event["body"]
    isBase64Encoded = bool(event["isBase64Encoded"])
    # 获取请求头中的Content-Type，用于解析multipart
    content_type = event.get("headers", {}).get("Content-Type", event.get("headers", {}).get("content-type", ""))
    
    if isBase64Encoded:
        body = base64.b64decode(body)
    else:
        body = body.encode('utf-8')

    # Parse multipart/form data (includes an uploaded image)
    # 传入content_type参数，让parse_form_data获取boundary
    form_data = parse_form_data(
        {"CONTENT_TYPE": content_type}, 
        stream_factory=lambda: BytesIO(body)
    )
    
    form_values = form_data[0]
    form_files = form_data[1]

    # End of Parse multipart/form 

    # 使用正确的变量名form_values获取字段
    hotel_name = form_values.get('hotelName')
    hotel_rating = form_values.get('hotelRating')
    hotel_city = form_values.get('hotelCity')
    hotel_price = form_values.get('hotelPrice')
    file_name = form_values.get('fileName')
    user_id = form_values.get('userId')
    id_token = form_values.get('idToken')
    
    logger.info(f"Hotel Name: {hotel_name}")
    
    # 直接读取二进制文件，不做decode
    file = form_files.get('fileData').read()

    token = jwt.decode(id_token, verify=False)
    groups = token.get("cognito:groups", [])

    # 判断Admin是否在用户组列表中
    if "Admin" not in groups:
        return {
            "statusCode": 401,
            "headers": headers,
            "body": json.dumps({
                "Error": "You are not a member of the Admin group"
            })
        }

    bucket_name = os.environ.get("bucketName")
    region = os.environ.get("AWS_REGION")

    s3_client = boto3.client("s3", region_name=region)
    dynamoDb = boto3.resource("dynamodb", region_name=region)
    table = dynamoDb.Table("Hotels")

    try:
        s3_client.put_object(
            Bucket=bucket_name,
            Key=file_name,
            Body=file
        )

        hotel = {
            "UserId": user_id,
            "Id": str(uuid.uuid4()),
            "Name": hotel_name,
            "CityName": hotel_city,
            "Price": int(hotel_price),
            "Rating": int(hotel_rating),
            "FileName": file_name
        }

        table.put_item(Item=hotel)
        logger.info("Hotel data saved successfully")

    except Exception as e:
        logger.error(f"Upload error: {str(e)}")
        return {
            "statusCode": 500,
            "headers": headers,
            "body": json.dumps({
                "Error": "Uploading the hotel photo failed",
                "Details": str(e)
            })
        }

    return {
        'statusCode': 200,
        'headers': headers,
        'body': json.dumps('Hotel uploaded successfully')
    }

额外说明 #

1. Layer依赖验证：确保你的Lambda Layer包含了werkzeug及其所有依赖（如markupsafe等），可以通过在本地创建虚拟环境安装后打包来生成Layer
2. API Gateway配置：确保API Gateway的集成请求中，没有将请求体转换为JSON，要保持原始multipart/form-data格式传递给Lambda
3. JWT安全建议：verify=False会关闭JWT签名验证，生产环境中建议配置正确的公钥进行验证，避免伪造Token

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