tessl/pypi-falcon

The ultra-reliable, fast ASGI+WSGI framework for building data plane APIs at scale.

—

Pending

Overview

Eval results

Files

Media Processing

Name: tessl/pypi-falcon
Author: tessl

Pluggable media handling system for serializing and deserializing request/response bodies. Supports JSON, MessagePack, multipart forms, URL-encoded data, and custom media types with extensible handler architecture.

Capabilities

Base Media Handler Classes

Foundation classes for implementing custom media processors.

class BaseHandler:
    def serialize(self, media: object, content_type: str) -> bytes:
        """
        Serialize Python object to bytes.
        
        Args:
            media: Python object to serialize
            content_type: MIME content type
            
        Returns:
            Serialized bytes
        """
    
    def deserialize(self, stream: object, content_type: str, content_length: int) -> object:
        """
        Deserialize bytes stream to Python object.
        
        Args:
            stream: Input byte stream
            content_type: MIME content type
            content_length: Content length in bytes
            
        Returns:
            Deserialized Python object
        """

class TextBaseHandlerWS:
    def serialize(self, media: object) -> str:
        """
        Serialize Python object to text for WebSocket.
        
        Args:
            media: Python object to serialize
            
        Returns:
            Serialized text string
        """
    
    def deserialize(self, payload: str) -> object:
        """
        Deserialize text payload to Python object.
        
        Args:
            payload: Text payload from WebSocket
            
        Returns:
            Deserialized Python object
        """

class BinaryBaseHandlerWS:
    def serialize(self, media: object) -> bytes:
        """
        Serialize Python object to bytes for WebSocket.
        
        Args:
            media: Python object to serialize
            
        Returns:
            Serialized bytes
        """
    
    def deserialize(self, payload: bytes) -> object:
        """
        Deserialize binary payload to Python object.
        
        Args:
            payload: Binary payload from WebSocket
            
        Returns:
            Deserialized Python object
        """

Media Handler Registry

Central registry for managing media type handlers and content negotiation.

class Handlers:
    def __init__(self, initial: dict = None):
        """
        Create media handler registry.
        
        Args:
            initial: Initial handler mappings {media_type: handler}
        """
    
    def find_by_media_type(self, media_type: str, default: object = None) -> BaseHandler:
        """
        Find handler for specific media type.
        
        Args:
            media_type: MIME media type (e.g., 'application/json')
            default: Default handler if none found
            
        Returns:
            Media handler instance or default
        """
    
    def __setitem__(self, media_type: str, handler: BaseHandler):
        """
        Register handler for media type.
        
        Args:
            media_type: MIME media type
            handler: Handler instance
        """
    
    def __getitem__(self, media_type: str) -> BaseHandler:
        """
        Get handler for media type.
        
        Args:
            media_type: MIME media type
            
        Returns:
            Handler instance
        """

class MissingDependencyHandler:
    def __init__(self, library: str, media_type: str):
        """
        Placeholder handler when dependencies are missing.
        
        Args:
            library: Name of missing library
            media_type: Media type this handler would support
        """
    
    def serialize(self, media: object, content_type: str = None) -> bytes:
        """Raises error about missing dependency"""
    
    def deserialize(self, stream: object, content_type: str, content_length: int) -> object:
        """Raises error about missing dependency"""

Handler Registry Usage

import falcon

# Create custom handlers registry
handlers = falcon.media.Handlers({
    'application/json': falcon.media.JSONHandler(),
    'application/msgpack': falcon.media.MessagePackHandler(),
})

# Configure app with custom handlers
req_options = falcon.RequestOptions(media_handlers=handlers)
resp_options = falcon.ResponseOptions(media_handlers=handlers)

app = falcon.App(
    req_options=req_options,
    resp_options=resp_options
)

Built-in JSON Handler

High-performance JSON serialization and deserialization using standard library or orjson.

class JSONHandler(BaseHandler):
    def __init__(self, loads: callable = None, dumps: callable = None):
        """
        JSON media handler.
        
        Args:
            loads: Custom JSON decoder function
            dumps: Custom JSON encoder function
        """
    
    def serialize(self, media: object, content_type: str = None) -> bytes:
        """
        Serialize Python object to JSON bytes.
        
        Args:
            media: Python object (dict, list, etc.)
            content_type: Content type (ignored)
            
        Returns:
            JSON bytes
        """
    
    def deserialize(self, stream: object, content_type: str, content_length: int) -> object:
        """
        Deserialize JSON bytes to Python object.
        
        Args:
            stream: JSON byte stream
            content_type: Content type (ignored) 
            content_length: Content length
            
        Returns:
            Python object
        """

class JSONHandlerWS(TextBaseHandlerWS):
    def __init__(self, loads: callable = None, dumps: callable = None):
        """
        JSON WebSocket handler.
        
        Args:
            loads: Custom JSON decoder function
            dumps: Custom JSON encoder function
        """

MessagePack Handler

Efficient MessagePack binary serialization support.

class MessagePackHandler(BaseHandler):
    def __init__(self, **kwargs):
        """
        MessagePack media handler.
        
        Args:
            **kwargs: Additional options for msgpack
        """
    
    def serialize(self, media: object, content_type: str = None) -> bytes:
        """
        Serialize Python object to MessagePack bytes.
        
        Args:
            media: Python object
            content_type: Content type (ignored)
            
        Returns:
            MessagePack bytes
        """
    
    def deserialize(self, stream: object, content_type: str, content_length: int) -> object:
        """
        Deserialize MessagePack bytes to Python object.
        
        Args:
            stream: MessagePack byte stream
            content_type: Content type (ignored)
            content_length: Content length
            
        Returns:
            Python object
        """

class MessagePackHandlerWS(BinaryBaseHandlerWS):
    def __init__(self, **kwargs):
        """
        MessagePack WebSocket handler.
        
        Args:
            **kwargs: Additional options for msgpack
        """

Multipart Form Handler

Support for multipart/form-data file uploads and form processing.

class MultipartFormHandler(BaseHandler):
    def __init__(self):
        """Multipart form data handler."""
    
    def serialize(self, media: object, content_type: str = None) -> bytes:
        """
        Serialize form data to multipart bytes.
        
        Args:
            media: Form data dictionary
            content_type: Content type with boundary
            
        Returns:
            Multipart bytes
        """
    
    def deserialize(self, stream: object, content_type: str, content_length: int) -> object:
        """
        Parse multipart form data from stream.
        
        Args:
            stream: Multipart byte stream
            content_type: Content type with boundary parameter
            content_length: Content length
            
        Returns:
            Form data dictionary
        """

Multipart Usage Example

class FileUploadResource:
    def on_post(self, req, resp):
        """Handle file upload"""
        # req.media contains parsed multipart data
        form_data = req.media
        
        # Access uploaded file
        uploaded_file = form_data.get('file')
        if uploaded_file:
            # Process file upload
            filename = uploaded_file.filename
            content_type = uploaded_file.content_type
            file_data = uploaded_file.data  # or use .stream
            
            # Save file
            save_uploaded_file(filename, file_data)
            
            resp.status = falcon.HTTP_201
            resp.media = {'uploaded': filename}
        else:
            raise falcon.HTTPBadRequest(
                title='Missing file',
                description='No file provided in upload'
            )

URL-Encoded Form Handler

Support for application/x-www-form-urlencoded form data.

class URLEncodedFormHandler(BaseHandler):
    def __init__(self):
        """URL-encoded form handler."""
    
    def serialize(self, media: object, content_type: str = None) -> bytes:
        """
        Serialize form data to URL-encoded bytes.
        
        Args:
            media: Form data dictionary
            content_type: Content type (ignored)
            
        Returns:
            URL-encoded bytes
        """
    
    def deserialize(self, stream: object, content_type: str, content_length: int) -> object:
        """
        Parse URL-encoded form data from stream.
        
        Args:
            stream: URL-encoded byte stream
            content_type: Content type (ignored)
            content_length: Content length
            
        Returns:
            Form data dictionary
        """

Media Validation

JSON Schema validation support for request/response media.

# Validation support (falcon.media.validators.jsonschema module)
def validate(schema: dict) -> callable:
    """
    Create JSON Schema validator decorator.
    
    Args:
        schema: JSON Schema dictionary
        
    Returns:
        Decorator function for validation
    """

Validation Example

from falcon.media.validators import jsonschema

user_schema = {
    'type': 'object',
    'properties': {
        'name': {'type': 'string', 'minLength': 1},
        'email': {'type': 'string', 'format': 'email'},
        'age': {'type': 'integer', 'minimum': 0}
    },
    'required': ['name', 'email']
}

class UserResource:
    @jsonschema.validate(req_schema=user_schema)
    def on_post(self, req, resp):
        # req.media is automatically validated against schema
        user_data = req.media
        new_user = create_user(user_data)
        resp.status = falcon.HTTP_201
        resp.media = new_user

Custom Media Handler

Creating custom media handlers for specialized content types.

class CustomHandler(BaseHandler):
    def __init__(self, **options):
        """
        Custom media handler implementation.
        
        Args:
            **options: Handler-specific options
        """
    
    def serialize(self, media: object, content_type: str = None) -> bytes:
        """Implement custom serialization logic"""
    
    def deserialize(self, stream: object, content_type: str, content_length: int) -> object:
        """Implement custom deserialization logic"""

Custom Handler Example

import yaml
import falcon

class YAMLHandler(falcon.media.BaseHandler):
    """Custom YAML media handler"""
    
    def serialize(self, media, content_type=None):
        """Serialize Python object to YAML"""
        return yaml.dump(media, default_flow_style=False).encode('utf-8')
    
    def deserialize(self, stream, content_type, content_length):
        """Deserialize YAML to Python object"""
        data = stream.read(content_length).decode('utf-8')
        return yaml.safe_load(data)

# Register custom handler
app = falcon.App()
app.req_options.media_handlers['application/yaml'] = YAMLHandler()
app.resp_options.media_handlers['application/yaml'] = YAMLHandler()

class YAMLResource:
    def on_post(self, req, resp):
        # Automatically uses YAML handler based on Content-Type
        data = req.media  # YAML parsed to Python object
        resp.content_type = 'application/yaml'
        resp.media = {'processed': data}  # Python object serialized to YAML

Error Handling

Media processing specific error classes for robust error handling.

class MediaNotFoundError(Exception):
    """Raised when no media handler found for content type"""

class MediaMalformedError(Exception):  
    """Raised when media content is malformed or invalid"""

class MediaValidationError(Exception):
    """Raised when media validation fails"""

class MultipartParseError(Exception):
    """Raised when multipart form parsing fails"""

Types

# Media handler base classes
BaseHandler: type
TextBaseHandlerWS: type
BinaryBaseHandlerWS: type

# Handler registry
Handlers: type
MissingDependencyHandler: type

# Built-in handlers
JSONHandler: type
JSONHandlerWS: type
MessagePackHandler: type
MessagePackHandlerWS: type
MultipartFormHandler: type
URLEncodedFormHandler: type

# Validation
falcon.media.validators.jsonschema.validate: callable

# Media errors
MediaNotFoundError: type
MediaMalformedError: type
MediaValidationError: type
MultipartParseError: type

Install with Tessl CLI