Send Multipart Requests Using AIOHTTP in Python

When working with web applications, you often need to send multipart requests, which allow you to send both text data and files in a single HTTP request.

In this tutorial, you’ll learn how to create and send multipart requests using AIOHTTP in Python.

 

 

Create a Multipart Form Data Payload

To create a multipart form data payload, you’ll use the MultipartWriter class from AIOHTTP.

Here’s how to initialize it:

import aiohttp
from aiohttp import MultipartWriter
multipart = MultipartWriter()

This code creates a new MultipartWriter object that you’ll use to build your multipart payload.

You can add text fields to your multipart payload using the append method:

multipart.append('Hello, World!', {'Content-Type': 'text/plain'})
multipart.append('{"key": "value"}', {'Content-Type': 'application/json'})

This code adds two text fields to the multipart payload: a plain text field and a JSON field.

To add file fields, you can use the append_json method for JSON data or open the file and append its contents:

import json
data = {'name': 'John Doe', 'age': 30}
multipart.append_json(data)
with open('example.txt', 'rb') as f:
    multipart.append(f, {'Content-Type': 'text/plain', 'Content-Disposition': 'attachment; filename="example.txt"'})

This code adds a JSON field and a file field to the multipart payload. The file is read in binary mode and added with appropriate headers.

You can set specific content types for each part of your multipart payload:

part = multipart.append(b'Binary data', {'Content-Type': 'application/octet-stream'})
part.set_content_disposition('attachment', filename='data.bin')

This code adds a binary data part to the payload and sets its content type and disposition.

 

Send Multipart Requests

To send a multipart request, use the ClientSession.post() method:

async with aiohttp.ClientSession() as session:
    async with session.post('https://domain.com/post', data=multipart) as response:
        print(await response.text())

This code sends a POST request with the multipart payload to httpbin.org and prints the response.

You may need to set specific headers for your multipart request:

headers = {
    'User-Agent': 'AIOHTTP Multipart Example',
    'Accept': 'application/json'
}
async with aiohttp.ClientSession() as session:
    async with session.post('https://domain/post', data=multipart, headers=headers) as response:
        print(await response.json())

This code sets custom headers for the request, including a User-Agent and Accept header.

 

Chunked File Uploads

For large files, you can implement streaming uploads to avoid loading the entire file into memory:

async def send_files():
    multipart = MultipartWriter()
    multipart.append('Hello, World!', {'Content-Type': 'text/plain'})
    multipart.append('{"key": "value"}', {'Content-Type': 'application/json'})    
    data = {'name': 'John Doe', 'age': 30}
    multipart.append_json(data)    
    with open('example.txt', 'rb') as f:
        multipart.append(f, {'Content-Type': 'text/plain', 'Content-Disposition': 'attachment; filename="example.txt"'})

    async def file_sender(file_name):
        async with aiofiles.open(file_name, 'rb') as f:
            chunk = await f.read(64 * 1024)
            while chunk:
                yield chunk
                chunk = await f.read(64 * 1024)

    async with aiohttp.ClientSession() as session:
        data = aiohttp.FormData()
        data.add_field('file',
                       file_sender('large_file.zip'),
                       filename='large_file.zip',
                       content_type='application/zip')
        async with session.post('https://httpbin.org/post', data=data) as response:
            print(await response.text())
asyncio.run(send_files())

This code implements a streaming upload for a large file, which is read and sent in chunks.

 

Customize MultipartWriter for Complex Payloads

For complex payloads, you can customize the MultipartWriter:

async def send_multipart_request():
    multipart = MultipartWriter('mixed')
    text_part = multipart.append('Text content', {'Content-Type': 'text/plain'})
    image_part = multipart.append(open('image.jpg', 'rb'), {'Content-Type': 'image/jpeg'})

    # Create a nested multipart
    nested_multipart = MultipartWriter('form-data')
    nested_multipart.append_json({'key': 'value'})
    nested_multipart.append(open('document.pdf', 'rb'), {'Content-Type': 'application/pdf'})

    # Append the nested multipart to the main multipart
    multipart.append(nested_multipart)
    async with aiohttp.ClientSession() as session:
        async with session.post('https://domain.com/post', data=multipart) as response:
            print(await response.text())
asyncio.run(send_multipart_request())

This code creates a complex multipart payload with mixed content types, including nested multipart data, and sends it in a POST request.

It uses asynchronous file handling and networking to manage I/O operations.