tessl/pypi-python-multipart

A streaming multipart parser for Python that enables efficient handling of file uploads and form data in web applications

—

Pending

Overview

Eval results

Files

Exception Handling

Name: tessl/pypi-python-multipart
Author: tessl

Comprehensive exception hierarchy for robust error handling across all parsing operations. Python-multipart provides specific exception types for different error conditions, enabling precise error handling and debugging.

Exception Hierarchy

ValueError
└── FormParserError (base exception)
    ├── ParseError
    │   ├── MultipartParseError
    │   ├── QuerystringParseError
    │   └── DecodeError
    └── FileError (also inherits from OSError)

Capabilities

Base Exceptions

FormParserError

Base error class for all form parser exceptions.

class FormParserError(ValueError):
    """
    Base error class for form parser.
    Inherits from ValueError for compatibility.
    """

ParseError

Base class for parsing-related errors with offset tracking.

class ParseError(FormParserError):
    """
    Base exception for parsing errors.
    
    Attributes:
    - offset: Position in input data where error occurred (-1 if not specified)
    """
    
    offset = -1  # Offset in input data chunk where parse error occurred

Specific Parse Errors

MultipartParseError

Specific error raised when MultipartParser detects parsing errors in multipart/form-data content.

class MultipartParseError(ParseError):
    """
    Error raised by MultipartParser when detecting parse errors.
    Used for malformed multipart boundaries, headers, or structure.
    """

QuerystringParseError

Specific error raised when QuerystringParser detects parsing errors in URL-encoded content.

class QuerystringParseError(ParseError):
    """
    Error raised by QuerystringParser when detecting parse errors.
    Used for malformed URL-encoded data or invalid characters.
    """

DecodeError

Error raised when content decoders encounter invalid encoded data.

class DecodeError(ParseError):
    """
    Error raised by Base64Decoder or QuotedPrintableDecoder.
    Used for invalid Base64 data or decoding failures.
    """

FileError

Exception for problems with File class operations, inheriting from both FormParserError and OSError.

class FileError(FormParserError, OSError):
    """
    Exception for File class problems.
    Combines form parsing context with OS-level error information.
    """

Usage Examples

Basic Exception Handling

from python_multipart import parse_form
from python_multipart.exceptions import (
    FormParserError,
    MultipartParseError,
    DecodeError,
    FileError
)

def safe_form_parsing(headers, input_stream):
    """Demonstrate comprehensive exception handling."""
    
    fields = []
    files = []
    errors = []
    
    def on_field(field):
        try:
            fields.append({
                'name': field.field_name.decode('utf-8'),
                'value': field.value.decode('utf-8') if field.value else None
            })
        except UnicodeDecodeError as e:
            errors.append(f"Field encoding error: {e}")
        finally:
            field.close()
    
    def on_file(file):
        try:
            files.append({
                'field_name': file.field_name.decode('utf-8'),
                'filename': file.file_name.decode('utf-8') if file.file_name else None,
                'size': file.size
            })
        except UnicodeDecodeError as e:
            errors.append(f"File name encoding error: {e}")
        except Exception as e:
            errors.append(f"File processing error: {e}")
        finally:
            file.close()
    
    try:
        parse_form(headers, input_stream, on_field, on_file)
        
        return {
            'success': True,
            'fields': fields,
            'files': files,
            'errors': errors
        }
        
    except MultipartParseError as e:
        return {
            'success': False,
            'error': 'Malformed multipart data',
            'details': str(e),
            'offset': getattr(e, 'offset', -1)
        }
    
    except DecodeError as e:
        return {
            'success': False,
            'error': 'Content decoding failed',
            'details': str(e),
            'offset': getattr(e, 'offset', -1)
        }
    
    except FileError as e:
        return {
            'success': False,
            'error': 'File handling error',
            'details': str(e),
            'errno': getattr(e, 'errno', None)
        }
    
    except FormParserError as e:
        return {
            'success': False,
            'error': 'Form parsing error',
            'details': str(e)
        }
    
    except Exception as e:
        return {
            'success': False,
            'error': 'Unexpected error',
            'details': str(e)
        }

Parser-Specific Error Handling

from python_multipart import MultipartParser, QuerystringParser
from python_multipart.exceptions import MultipartParseError, QuerystringParseError

def handle_multipart_errors(boundary, data_stream):
    """Handle MultipartParser specific errors."""
    
    def on_part_data(data, start, end):
        # Process part data
        chunk = data[start:end]
        print(f"Processing {len(chunk)} bytes")
    
    callbacks = {'on_part_data': on_part_data}
    
    try:
        parser = MultipartParser(boundary, callbacks)
        
        while True:
            chunk = data_stream.read(1024)
            if not chunk:
                break
            parser.write(chunk)
        
        parser.finalize()
        return {'success': True}
        
    except MultipartParseError as e:
        error_info = {
            'success': False,
            'error_type': 'multipart_parse_error',
            'message': str(e),
            'offset': e.offset
        }
        
        # Provide context based on offset
        if e.offset >= 0:
            error_info['context'] = f"Error at byte position {e.offset}"
        
        return error_info

def handle_querystring_errors(data_stream):
    """Handle QuerystringParser specific errors."""
    
    fields = {}
    current_field = None
    current_value = b''
    
    def on_field_name(data, start, end):
        nonlocal current_field
        current_field = data[start:end].decode('utf-8')
    
    def on_field_data(data, start, end):
        nonlocal current_value
        current_value += data[start:end]
    
    def on_field_end():
        nonlocal current_field, current_value
        if current_field:
            fields[current_field] = current_value.decode('utf-8')
        current_field = None
        current_value = b''
    
    callbacks = {
        'on_field_name': on_field_name,
        'on_field_data': on_field_data,
        'on_field_end': on_field_end
    }
    
    try:
        parser = QuerystringParser(callbacks, strict_parsing=True)
        
        while True:
            chunk = data_stream.read(1024)
            if not chunk:
                break
            parser.write(chunk)
        
        parser.finalize()
        return {'success': True, 'fields': fields}
        
    except QuerystringParseError as e:
        return {
            'success': False,
            'error_type': 'querystring_parse_error',
            'message': str(e),
            'offset': e.offset,
            'partial_fields': fields
        }

Decoder Error Handling

from python_multipart.decoders import Base64Decoder, QuotedPrintableDecoder
from python_multipart.exceptions import DecodeError
import io

def safe_base64_decode(encoded_data):
    """Safely decode Base64 data with error handling."""
    
    output = io.BytesIO()
    decoder = Base64Decoder(output)
    
    try:
        decoder.write(encoded_data)
        decoder.finalize()
        
        output.seek(0)
        return {
            'success': True,
            'data': output.read()
        }
        
    except DecodeError as e:
        return {
            'success': False,
            'error': 'Base64 decode error',
            'message': str(e),
            'cache_size': len(decoder.cache) if hasattr(decoder, 'cache') else 0
        }
    
    finally:
        decoder.close()

def safe_quoted_printable_decode(encoded_data):
    """Safely decode quoted-printable data."""
    
    output = io.BytesIO()
    decoder = QuotedPrintableDecoder(output)
    
    try:
        decoder.write(encoded_data)
        decoder.finalize()
        
        output.seek(0)
        return {
            'success': True,
            'data': output.read()
        }
        
    except Exception as e:  # QuotedPrintableDecoder rarely raises exceptions
        return {
            'success': False,
            'error': 'Quoted-printable decode error',
            'message': str(e)
        }
    
    finally:
        decoder.close()

# Test error handling
malformed_base64 = b"SGVsbG8gV29ybGQ!"  # Invalid padding
result = safe_base64_decode(malformed_base64)
print(f"Base64 decode result: {result}")

File Error Handling

from python_multipart import File
from python_multipart.exceptions import FileError
import os
import tempfile

def safe_file_handling(file_name, field_name, content):
    """Demonstrate file error handling."""
    
    # Configure file to use a restricted directory
    config = {
        'UPLOAD_DIR': '/restricted/path',  # This might not exist
        'MAX_MEMORY_FILE_SIZE': 1024
    }
    
    try:
        file_obj = File(file_name.encode(), field_name.encode(), config)
        
        # Write content
        file_obj.write(content)
        file_obj.finalize()
        
        result = {
            'success': True,
            'file_name': file_name,
            'size': file_obj.size,
            'in_memory': file_obj.in_memory
        }
        
        if not file_obj.in_memory:
            result['temp_path'] = file_obj.actual_file_name.decode()
        
        file_obj.close()
        return result
        
    except FileError as e:
        return {
            'success': False,
            'error_type': 'file_error',
            'message': str(e),
            'errno': getattr(e, 'errno', None),
            'filename': getattr(e, 'filename', None)
        }
    
    except OSError as e:
        return {
            'success': False,
            'error_type': 'os_error',
            'message': str(e),
            'errno': e.errno
        }
    
    except Exception as e:
        return {
            'success': False,
            'error_type': 'unexpected_error',
            'message': str(e)
        }

# Test file error handling
result = safe_file_handling('test.txt', 'upload', b'File content here')
print(f"File handling result: {result}")

Error Recovery Strategies

from python_multipart import FormParser
from python_multipart.exceptions import FormParserError, ParseError
import logging

class RobustFormParser:
    """Form parser with error recovery and logging."""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.error_count = 0
        self.max_errors = 10
    
    def parse_with_recovery(self, headers, input_stream):
        """Parse form data with error recovery."""
        
        fields = []
        files = []
        errors = []
        
        def on_field(field):
            try:
                fields.append({
                    'name': field.field_name.decode('utf-8', errors='replace'),
                    'value': field.value.decode('utf-8', errors='replace') if field.value else None
                })
            except Exception as e:
                self.logger.warning(f"Field processing error: {e}")
                errors.append(f"Field error: {e}")
            finally:
                field.close()
        
        def on_file(file):
            try:
                files.append({
                    'field_name': file.field_name.decode('utf-8', errors='replace'),
                    'filename': file.file_name.decode('utf-8', errors='replace') if file.file_name else None,
                    'size': file.size
                })
            except Exception as e:
                self.logger.warning(f"File processing error: {e}")
                errors.append(f"File error: {e}")
            finally:
                file.close()
        
        try:
            # Try parsing with error recovery
            parser = FormParser(
                headers.get('Content-Type', 'application/octet-stream'),
                on_field,
                on_file
            )
            
            chunk_count = 0
            while True:
                try:
                    chunk = input_stream.read(8192)
                    if not chunk:
                        break
                    
                    parser.write(chunk)
                    chunk_count += 1
                    
                except ParseError as e:
                    self.error_count += 1
                    self.logger.error(f"Parse error in chunk {chunk_count}: {e}")
                    errors.append(f"Parse error at chunk {chunk_count}: {e}")
                    
                    if self.error_count >= self.max_errors:
                        raise Exception("Too many parse errors, aborting")
                    
                    # Continue with next chunk
                    continue
            
            parser.finalize()
            parser.close()
            
            return {
                'success': True,
                'fields': fields,
                'files': files,
                'errors': errors,
                'error_count': self.error_count
            }
            
        except Exception as e:
            self.logger.error(f"Fatal parsing error: {e}")
            return {
                'success': False,
                'error': str(e),
                'partial_fields': fields,
                'partial_files': files,
                'errors': errors
            }

# Usage
parser = RobustFormParser()
# result = parser.parse_with_recovery(headers, stream)

Install with Tessl CLI