Python解析multipart boundary：aiohttp与requests文件上传详解

Content-Type: multipart/form-data; boundary=----WebKitFormBoundary7MA4YWxkTrZu0gW

import requests

# 目标URL（测试用：httpbin.org可返回提交的数据）
url = 'http://httpbin.org/post'

# 构造文件上传数据：requests会自动构造multipart/form-data请求
files = {
    # 第一个参数为字段名称，元组中依次为：(文件名, 文件对象, MIME类型)
    'file':('test.txt', open('test.txt', 'rb'), 'text/plain')
}

# 发送POST请求
response = requests.post(url, files=files)

# 打印服务器返回内容
print(response.text)

pip install requests-toolbelt

from requests_toolbelt.multipart.encoder import MultipartEncoder
import requests

def send_formdata_manual():
    url = 'http://httpbin.org/post'
    # 自定义 boundary 字符串
    boundary = '----WebKitFormBoundary7MA4YWxkTrZu0gW'
    # 使用 MultipartEncoder 构造 multipart 数据，同时指定 boundary
    encoder = MultipartEncoder(
        fields={
            'field1': 'value1',
            'file': ('test.txt', open('test.txt', 'rb'), 'text/plain')
        },
        boundary=boundary
    )
    # 设置 Content-Type 头，包含自定义的 boundary
    headers = {'Content-Type': encoder.content_type}
    # 发送 POST 请求
    response = requests.post(url, data=encoder, headers=headers)
    print("手动设置 boundary 的响应：", response.text)

send_formdata_manual()

import requests

# 目标URL
url = 'http://httpbin.org/post'

# 自定义boundary字符串
boundary = '----WebKitFormBoundary7MA4YWxkTrZu0gW'

# 构造请求体各部分数据，注意各部分之间以boundary分隔
data_lines = []
# 添加第一个字段：普通文本字段
data_lines.append('--' + boundary)
data_lines.append('Content-Disposition: form-data; name="field1"')
data_lines.append('')  # 空行分隔头与内容
data_lines.append('value1')

# 添加第二个字段：文件字段
data_lines.append('--' + boundary)
data_lines.append('Content-Disposition: form-data; name="file"; filename="test.txt"')
data_lines.append('Content-Type: text/plain')
data_lines.append('')
# 读取文件内容（确保当前目录下有test.txt文件）
with open('test.txt', 'r', encoding='utf-8') as f:
    data_lines.append(f.read())

# 结束标志：加上结尾的boundary标记
data_lines.append('--' + boundary + '--')

# 将各部分用CRLF连接
body = '\r\n'.join(data_lines)

# 构造请求头，指明Content-Type及boundary
headers = {
    'Content-Type': 'multipart/form-data; boundary=' + boundary
}

# 发送POST请求，此处需要将body转换为字节串
response = requests.post(url, data=body.encode('utf-8'), headers=headers)
print(response.text)

import aiohttp
import asyncio

async def main():
    url = 'http://httpbin.org/post'
    # 使用aiohttp提供的FormData构造表单数据
    form = aiohttp.FormData()
    form.add_field('field1', 'value1')
    # 添加文件字段，注意以二进制方式打开文件
    form.add_field('file',
                   open('test.txt', 'rb'),
                   filename='test.txt',
                   content_type='text/plain')
    
    # 使用异步上下文管理器发送请求
    async with aiohttp.ClientSession() as session:
        async with session.post(url, data=form) as resp:
            print(await resp.text())

# 运行异步任务
asyncio.run(main())

import aiohttp
import asyncio

async def send_formdata_manual():
    # 自定义 boundary 字符串（注意确保不会与数据内容冲突）
    boundary = '----WebKitFormBoundary7MA4YWxkTrZu0gW'
    # 创建 MultipartWriter 对象，手动指定 boundary
    mp_writer = aiohttp.MultipartWriter(boundary=boundary)
    
    # 添加普通字段
    part1 = mp_writer.append('value1')
    part1.set_content_disposition('form-data', name='field1')
    
    # 添加文件字段
    with open('test.txt', 'rb') as f:
        part2 = mp_writer.append(f.read(), {'Content-Type': 'text/plain'})
        part2.set_content_disposition('form-data', name='file', filename='test.txt')
    
    # 发送 POST 请求
    async with aiohttp.ClientSession() as session:
        async with session.post('http://httpbin.org/post', data=mp_writer) as resp:
            result = await resp.text()
            print("手动设置 boundary 的响应：", result)

# 运行异步任务
asyncio.run(send_formdata_manual())

import aiohttp
import asyncio

async def main():
    url = 'http://httpbin.org/post'
    
    # 自定义boundary
    boundary = '----WebKitFormBoundary7MA4YWxkTrZu0gW'
    parts = []
    # 添加普通文本字段
    parts.append('--' + boundary)
    parts.append('Content-Disposition: form-data; name="field1"')
    parts.append('')
    parts.append('value1')
    
    # 添加文件字段
    parts.append('--' + boundary)
    parts.append('Content-Disposition: form-data; name="file"; filename="test.txt"')
    parts.append('Content-Type: text/plain')
    parts.append('')
    with open('test.txt', 'r', encoding='utf-8') as f:
        parts.append(f.read())
    
    # 结束标记
    parts.append('--' + boundary + '--')
    
    # 构造完整请求体
    body = '\r\n'.join(parts)
    headers = {
        'Content-Type': 'multipart/form-data; boundary=' + boundary
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.post(url, data=body.encode('utf-8'), headers=headers) as resp:
            print(await resp.text())

asyncio.run(main())