CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-requests-toolbelt

A utility belt for advanced users of python-requests

—

Pending
Overview
Eval results
Files

multipart.mddocs/

Multipart Processing

Streaming multipart/form-data encoding and decoding capabilities for efficient handling of file uploads, form submissions, and multipart responses. The implementation supports large files without loading everything into memory and provides progress monitoring.

Capabilities

Multipart Encoding

Creates streaming multipart/form-data content for HTTP requests with support for files, strings, and mixed content types.

class MultipartEncoder:
    """
    Generic interface for creating multipart/form-data request bodies.
    
    Parameters:
    - fields: dict or list of tuples containing form fields
    - boundary: str, custom boundary string (auto-generated if None)
    - encoding: str, character encoding (default: 'utf-8')
    """
    def __init__(self, fields, boundary=None, encoding='utf-8'): ...
    
    # Instance attributes
    boundary_value: str  # The raw boundary string
    boundary: str        # Formatted boundary with --
    encoding: str        # Character encoding
    fields: any         # Original fields provided
    finished: bool      # Whether encoding is complete
    parts: list         # Pre-computed parts
    
    @property
    def content_type(self) -> str:
        """Returns the Content-Type header value including boundary."""
    
    @property
    def len(self) -> int:
        """Returns the total length of the multipart content."""
    
    def read(self, size: int = -1) -> bytes:
        """
        Read and return up to size bytes of multipart data.
        
        Parameters:
        - size: int, number of bytes to read (-1 for all)
        
        Returns:
        bytes: multipart data chunk
        """
    
    def to_string(self) -> bytes:
        """
        Returns the entire multipart content as bytes.
        
        Returns:
        bytes: complete multipart data
        """
    
    def __repr__(self) -> str:
        """Return string representation of the encoder."""

Usage Examples

import requests
from requests_toolbelt import MultipartEncoder

# Simple form fields
encoder = MultipartEncoder({
    'username': 'john_doe',
    'email': 'john@example.com'
})

response = requests.post(
    'https://httpbin.org/post',
    data=encoder,
    headers={'Content-Type': encoder.content_type}
)

# File upload with additional fields
encoder = MultipartEncoder({
    'field': 'value',
    'file': ('filename.txt', open('file.txt', 'rb'), 'text/plain'),
    'another_file': ('data.json', open('data.json', 'rb'), 'application/json')
})

response = requests.post(
    'https://httpbin.org/post',
    data=encoder,
    headers={'Content-Type': encoder.content_type}
)

# Streaming large file upload
with open('large_file.zip', 'rb') as f:
    encoder = MultipartEncoder({
        'file': ('large_file.zip', f, 'application/zip'),
        'description': 'Large file upload'
    })
    
    response = requests.post(
        'https://httpbin.org/post',
        data=encoder,
        headers={'Content-Type': encoder.content_type}
    )

Progress Monitoring

Monitor the progress of multipart uploads with callback functions.

class MultipartEncoderMonitor:
    """
    Monitor for tracking multipart upload progress.
    
    Parameters:
    - encoder: MultipartEncoder instance to monitor
    - callback: function called with monitor instance on each read (optional)
    """
    def __init__(self, encoder, callback=None): ...
    
    # Instance attributes
    encoder: MultipartEncoder  # The wrapped encoder
    callback: callable         # Progress callback function
    bytes_read: int            # Number of bytes read so far
    len: int                   # Total length of the multipart content
    content_type: str          # Delegates to encoder.content_type
    
    @classmethod
    def from_fields(cls, fields, boundary=None, encoding='utf-8', callback=None):
        """
        Create a monitor from fields directly.
        
        Parameters:
        - fields: dict or list of tuples containing form fields
        - boundary: str, custom boundary string (optional)
        - encoding: str, character encoding (default: 'utf-8')
        - callback: function called with monitor instance on each read (optional)
        
        Returns:
        MultipartEncoderMonitor: configured monitor instance
        """
    
    def read(self, size: int = -1) -> bytes:
        """
        Read data and trigger progress callback.
        
        Parameters:
        - size: int, number of bytes to read
        
        Returns:
        bytes: data chunk
        """
    
    def to_string(self) -> bytes:
        """Returns the entire multipart content as bytes."""

def IDENTITY(monitor):
    """Default callback function that returns monitor unchanged."""

Usage Examples

import requests
from requests_toolbelt import MultipartEncoder, MultipartEncoderMonitor

def progress_callback(monitor):
    percent = (monitor.bytes_read / monitor.len) * 100
    print(f"Upload progress: {percent:.1f}%")

encoder = MultipartEncoder({
    'file': ('large_file.zip', open('large_file.zip', 'rb'), 'application/zip')
})

monitor = MultipartEncoderMonitor(encoder, progress_callback)

response = requests.post(
    'https://httpbin.org/post',
    data=monitor,
    headers={'Content-Type': monitor.content_type}
)

Multipart Decoding

Parse multipart responses into individual parts with headers and content.

class MultipartDecoder:
    """
    Decoder for multipart HTTP responses.
    
    Parameters:
    - content: bytes, the multipart response body
    - content_type: str, the Content-Type header value
    - encoding: str, character encoding (default: 'utf-8')
    """
    def __init__(self, content, content_type, encoding='utf-8'): ...
    
    # Instance attributes
    content_type: str   # Original Content-Type header
    encoding: str      # Response body encoding
    boundary: bytes    # Extracted boundary
    
    @property
    def parts(self) -> tuple:
        """Returns tuple of BodyPart objects."""
    
    @classmethod
    def from_response(cls, response, encoding='utf-8'):
        """
        Create decoder from HTTP response.
        
        Parameters:
        - response: requests.Response object
        - encoding: str, character encoding (default: 'utf-8')
        
        Returns:
        MultipartDecoder: configured decoder instance
        """

class BodyPart:
    """
    Individual part of a multipart response.
    
    Parameters:
    - content: bytes, raw part content including headers
    - encoding: str, character encoding
    """
    def __init__(self, content, encoding): ...
    
    # Instance attributes
    encoding: str                      # Character encoding
    content: bytes                     # Raw content (attribute, not property)
    headers: CaseInsensitiveDict       # Part headers
    
    @property
    def text(self) -> str:
        """Returns the part content as decoded text."""

Usage Examples

import requests
from requests_toolbelt.multipart import MultipartDecoder

response = requests.get('https://example.com/multipart-response')

if 'multipart' in response.headers.get('Content-Type', ''):
    decoder = MultipartDecoder(response.content, response.headers['Content-Type'])
    
    for part in decoder.parts:
        print(f"Headers: {part.headers}")
        print(f"Content: {part.content}")
        
        # Access as text if it's textual content
        if 'text' in part.headers.get('Content-Type', ''):
            print(f"Text: {part.text}")

File Streaming from URLs

Stream files directly from URLs without downloading them first.

class FileFromURLWrapper:
    """
    File from URL wrapper for streaming files from remote URLs.
    
    Provides stateless solution for streaming files from one server to another
    without downloading the entire file into memory first.
    
    Parameters:
    - file_url: str, URL of the file to stream
    - session: requests.Session, optional session for requests (default: new session)
    
    Raises:
    - FileNotSupportedError: If URL doesn't provide content-length header
    """
    def __init__(self, file_url: str, session=None): ...
    
    # Instance attributes
    session: requests.Session  # Session used for requests
    len: int                  # Remaining bytes to read
    raw_data: any            # Raw response data stream
    
    def read(self, chunk_size: int) -> bytes:
        """
        Read file in chunks.
        
        Parameters:
        - chunk_size: int, number of bytes to read
        
        Returns:
        bytes: file data chunk
        """

Usage Examples

import requests
from requests_toolbelt.multipart.encoder import FileFromURLWrapper
from requests_toolbelt import MultipartEncoder

# Stream file from URL without session
url = 'https://httpbin.org/image/png'
streaming_encoder = MultipartEncoder({
    'file': FileFromURLWrapper(url),
    'description': 'Image from URL'
})

response = requests.post(
    'https://httpbin.org/post',
    data=streaming_encoder,
    headers={'Content-Type': streaming_encoder.content_type}
)

# Stream file with session
session = requests.Session()
streaming_encoder = MultipartEncoder({
    'file': FileFromURLWrapper(url, session=session)
})

response = session.post(
    'https://httpbin.org/post',
    data=streaming_encoder,
    headers={'Content-Type': streaming_encoder.content_type}
)

Exceptions

class ImproperBodyPartContentException(Exception):
    """Raised when body part content is malformed."""

class NonMultipartContentTypeException(Exception):
    """Raised when content type is not multipart."""

class FileNotSupportedError(Exception):
    """Raised when file from URL is not supported (missing content-length)."""

Install with Tessl CLI

npx tessl i tessl/pypi-requests-toolbelt

docs

adapters.md

authentication.md

cookies-exceptions.md

downloads.md

index.md

multipart.md

sessions-streaming.md

threading.md

utilities.md

tile.json