如何在Python Serverless函数中从请求体获取CSV文件
从Multipart请求体中提取CSV文件内容
我来帮你搞定这个问题!你的请求体是标准的multipart/form-data格式,得先解析这个格式才能拿到CSV内容。下面是具体的实现思路和代码:
步骤分解
- 提取分隔符(Boundary):请求体第一行开头的
----------------------------xxxx就是表单字段的分隔符,我们需要先提取它,记得去掉开头的--。 - 拆分请求体:用分隔符把整个body拆分成各个表单部分,过滤掉无效的空块和结尾的
--标记。 - 定位文件字段:遍历拆分后的部分,找到包含
name="file"的表单项,这就是我们要的CSV文件部分。 - 提取CSV文本:从这个表单项里剥离掉头部的
Content-Disposition和Content-Type信息,剩下的就是原始的CSV内容。 - 解析CSV(可选):如果需要把CSV转换成可操作的数据结构,可以用Python内置的
csv模块来处理。
完整代码实现
import csv from io import StringIO def data_file_prepare(event, context): # 1. 获取请求体和编码状态 body = event.get('body', '') is_base64 = event.get('isBase64Encoded', False) # 兼容base64编码的情况(你的场景是False,留着适配其他情况) if is_base64: import base64 body = base64.b64decode(body).decode('utf-8') # 2. 提取boundary分隔符 lines = body.split('\r\n') boundary = lines[0].strip('--') # 去掉开头的-- # 3. 分割成各个表单部分 parts = body.split(f'--{boundary}') # 过滤空内容和结尾的结束块 valid_parts = [part.strip() for part in parts if part.strip() and not part.strip().endswith('--')] csv_content = None for part in valid_parts: # 找到file对应的表单部分 if 'name="file"' in part: # 拆分头部和实际内容:以第一个空行(\r\n\r\n)为界 header_content_split = part.split('\r\n\r\n', 1) if len(header_content_split) == 2: _, content = header_content_split csv_content = content.strip() break if not csv_content: return {"statusCode": 400, "body": "请求中未找到CSV文件"} # 可选:把CSV文本解析成列表形式,方便后续处理 csv_reader = csv.reader(StringIO(csv_content)) csv_data = list(csv_reader) # 根据你的需求返回结果,这里示例返回成功信息 return { "statusCode": 200, "body": f"CSV内容提取成功,共{len(csv_data)}行数据" }
代码说明
- 兼容了base64编码的场景,让函数更通用;
- 手动解析multipart格式,不需要额外依赖,适合Lambda这类无服务器环境;
- 如果只需要原始的CSV文本,直接用
csv_content即可,解析成列表是可选操作。
内容的提问来源于stack exchange,提问作者Jonah Geladze
相关产品推荐
相关产品推荐

