1. 为什么需要multipart/form-data上传文件

第一次用Python上传文件时，我踩了个大坑。当时直接用字典传文件参数，结果服务器死活不认。后来抓包对比才发现，原来文件上传需要特殊的编码格式——这就是multipart/form-data的由来。

这种编码方式就像快递打包，每个参数都被独立包装，用boundary作为分隔符。比如上传Excel文件时，实际传输的内容是这样的：

------WebKitFormBoundary7MA4YWxkTrZu0gW Content-Disposition: form-data; name="username" 张三 ------WebKitFormBoundary7MA4YWxkTrZu0gW Content-Disposition: form-data; name="file"; filename="test.xlsx" Content-Type: application/vnd.openxmlformats-officedocument.spreadsheetml.sheet [文件二进制数据]

与普通表单提交不同，multipart/form-data能同时处理文本和二进制数据。我在实际项目中发现，这些场景必须使用它：

• 上传图片/视频等媒体文件
• 提交包含文件的复杂表单
• 需要保留文件原始元数据时

2. 基础方案：使用files参数的标准方法

2.1 单文件上传实战

最基础的用法是通过requests的files参数。还记得我开头提到的坑吗？正确姿势应该是这样的：

import requests url = 'https://example.com/upload' file_path = 'report.xlsx' with open(file_path, 'rb') as f: files = {'document': ('季度报告.xlsx', f, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')} r = requests.post(url, files=files) print(r.status_code)

这里有几个关键点：

1. 文件必须用二进制模式('rb')打开
2. 参数值必须是三元组：(文件名, 文件对象, 文件类型)
3. 字典键'document'需要与接口定义的参数名一致

2.2 混合参数上传技巧

实际项目中，我们经常需要同时上传文件和其他参数。比如上周我做的一个需求就要上传Excel并附带用户信息：

data = { 'user_id': (None, '10086', None), # 格式：(文件名, 值, 类型) 'department': (None, '市场部', None) } files = {'file': ('sales.xlsx', open('sales.xlsx', 'rb'))} response = requests.post(url, files={**data, **files})

注意这里使用了元组格式的混合参数。第一个None表示没有文件名，第三个None让Requests自动判断Content-Type。这种写法在测试人脸识别API时特别实用。

3. 进阶方案：requests_toolbelt高级用法

3.1 MultipartEncoder的优势

当遇到这些情况时，标准方法就不够用了：

• 需要自定义boundary
• 上传超大文件（内存优化）
• 需要精确控制每个字段的编码

这时就该祭出requests_toolbelt了。先安装这个神器：

pip install requests-toolbelt

这是我处理5GB视频上传时的代码：

from requests_toolbelt import MultipartEncoder import requests encoder = MultipartEncoder( fields={ 'title': '产品演示视频', 'description': '2023年最新版', 'video': ('demo.mp4', open('demo.mp4', 'rb'), 'video/mp4') }, boundary='----MyCustomBoundary12345' ) headers = {'Content-Type': encoder.content_type} response = requests.post(url, data=encoder, headers=headers)

MultipartEncoder会自动计算Content-Length，并且支持流式传输，不会一次性加载大文件到内存。

3.2 监控上传进度

上传大文件时，加上进度条用户体验会好很多：

from requests_toolbelt import MultipartEncoderMonitor def callback(monitor): print(f"已上传: {monitor.bytes_read/1024/1024:.2f}MB") encoder = MultipartEncoder(fields={'file': ('bigfile.zip', open('bigfile.zip', 'rb'))}) monitor = MultipartEncoderMonitor(encoder, callback) requests.post(url, data=monitor, headers={'Content-Type': monitor.content_type})

4. 企业级方案：Session管理认证场景

4.1 保持登录状态上传

很多企业系统需要先登录才能上传。用Session对象可以保持cookies：

session = requests.Session() # 先登录 login_data = {'username': 'admin', 'password': '123456'} session.post(login_url, data=login_data) # 再上传 files = {'file': open('data.csv', 'rb')} response = session.post(upload_url, files=files)

4.2 处理CSRF Token

遇到有CSRF防护的系统时，需要先获取token：

# 获取token login_page = session.get(login_url) token = parse_token(login_page.text) # 需要自己解析HTML # 带token上传 files = {'file': ('data.txt', open('data.txt', 'rb'))} data = {'csrf_token': token} response = session.post(upload_url, files=files, data=data)

5. 避坑指南与调试技巧

5.1 常见错误排查

• 错误1：400 Bad Request 可能原因：Content-Type被覆盖。解决方案：不要手动设置Content-Type头

• 错误2：文件损坏 可能原因：文件未以二进制模式打开。确保使用'rb'模式

• 错误3：连接超时 解决方案：调整超时设置

  requests.post(url, files=files, timeout=(3.05, 27))

5.2 抓包分析技巧

用mitmproxy调试时，重点关注：

1. 是否存在boundary字符串
2. 每个part的Content-Disposition是否正确
3. 文件二进制数据是否完整

这是我常用的调试代码片段：

import pprint from http.client import HTTPConnection HTTPConnection.debuglevel = 1 requests.post(url, files=files)

6. 性能优化实践

6.1 内存优化方案

处理大文件时，可以用流式上传避免内存爆炸：

class FileStream: def __init__(self, filename): self.f = open(filename, 'rb') self.size = os.path.getsize(filename) def read(self, size=-1): return self.f.read(size) def __len__(self): return self.size stream = FileStream('huge_file.iso') encoder = MultipartEncoder({'file': ('huge.iso', stream, 'application/octet-stream')}) requests.post(url, data=encoder, headers={'Content-Type': encoder.content_type})

6.2 多文件并行上传

使用线程池加速批量上传：

from concurrent.futures import ThreadPoolExecutor def upload_file(file_path): files = {'file': open(file_path, 'rb')} return requests.post(url, files=files) with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(upload_file, ['a.jpg', 'b.jpg', 'c.jpg']))

7. 内容安全与异常处理

7.1 文件类型校验

防止上传恶意文件的安全措施：

ALLOWED_TYPES = {'image/jpeg', 'image/png'} file = request.files['file'] if file.content_type not in ALLOWED_TYPES: raise ValueError("不支持的文件类型")

7.2 断点续传实现

网络不稳定时的恢复方案：

def resume_upload(file_path, url, chunk_size=1024*1024): file_size = os.path.getsize(file_path) uploaded = 0 while uploaded < file_size: with open(file_path, 'rb') as f: f.seek(uploaded) chunk = f.read(chunk_size) headers = { 'Content-Range': f'bytes {uploaded}-{uploaded+len(chunk)-1}/{file_size}', 'Content-Type': 'application/octet-stream' } resp = requests.put(url, data=chunk, headers=headers) uploaded += len(chunk)