tessl/pypi-connexion

OpenAPI/Swagger spec-first web framework for Python with automatic request validation and response serialization

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Validation System

Name: tessl/pypi-connexion
Author: tessl

Automatic request and response validation based on OpenAPI schema definitions. Connexion provides comprehensive validation for various content types with customizable validation strategies.

Capabilities

Validator Mapping

Central registry of validators for different content types and validation scenarios.

VALIDATOR_MAP: dict
"""
Global mapping of validation types to validator classes.
Keys include:
- 'body': Request body validators by content type
- 'parameter': Parameter validators
- 'response': Response validators by content type
"""

Abstract Base Validators

Base classes for implementing custom validators.

class AbstractRequestBodyValidator:
    """Base class for request body validators"""
    
    def __init__(self, schema: dict, validator=None, **kwargs):
        """
        Initialize request body validator.
        
        Parameters:
        - schema: JSON schema for validation
        - validator: Custom validator instance
        - **kwargs: Additional validator options
        """
    
    def validate_schema(self, body, url: str) -> dict:
        """
        Validate request body against schema.
        
        Parameters:
        - body: Request body to validate
        - url: Request URL for error context
        
        Returns:
        dict: Validated and possibly transformed body
        
        Raises:
        BadRequestProblem: If validation fails
        """
    
    @classmethod
    def validate_formdata_parameter_list(cls, request_body: dict) -> dict:
        """
        Validate form data parameters.
        
        Parameters:
        - request_body: Form data dictionary
        
        Returns:
        dict: Validated form data
        """

class AbstractParameterValidator:
    """Base class for parameter validators"""
    
    def __init__(self, parameters: list, uri_parser=None, **kwargs):
        """
        Initialize parameter validator.
        
        Parameters:
        - parameters: List of parameter definitions
        - uri_parser: URI parsing utility
        - **kwargs: Additional options
        """
    
    def validate_query_parameter_list(self, request) -> dict:
        """
        Validate query parameters.
        
        Parameters:
        - request: Request object
        
        Returns:
        dict: Validated query parameters
        """
    
    def validate_path_parameter_list(self, request) -> dict:
        """
        Validate path parameters.
        
        Parameters:
        - request: Request object
        
        Returns:
        dict: Validated path parameters
        """
    
    def validate_header_parameter_list(self, request) -> dict:
        """
        Validate header parameters.
        
        Parameters:
        - request: Request object
        
        Returns:
        dict: Validated header parameters
        """

class AbstractResponseBodyValidator:
    """Base class for response body validators"""
    
    def __init__(self, schema: dict, validator=None, **kwargs):
        """
        Initialize response body validator.
        
        Parameters:
        - schema: JSON schema for validation
        - validator: Custom validator instance
        - **kwargs: Additional options
        """
    
    def validate_schema(self, data, url: str) -> dict:
        """
        Validate response body against schema.
        
        Parameters:
        - data: Response data to validate
        - url: Request URL for error context
        
        Returns:
        dict: Validated response data
        
        Raises:
        InternalServerError: If validation fails
        """

Request Body Validators

Concrete validators for different request content types.

class JSONRequestBodyValidator(AbstractRequestBodyValidator):
    """Validator for application/json request bodies"""
    
    def validate_schema(self, body, url: str) -> dict:
        """Validate JSON request body against OpenAPI schema"""

class FormDataValidator(AbstractRequestBodyValidator):
    """Validator for application/x-www-form-urlencoded request bodies"""
    
    def validate_schema(self, body, url: str) -> dict:
        """Validate form data against OpenAPI schema"""

class MultiPartFormDataValidator(AbstractRequestBodyValidator):
    """Validator for multipart/form-data request bodies"""
    
    def validate_schema(self, body, url: str) -> dict:
        """Validate multipart form data including file uploads"""

Parameter Validators

Validators for URL parameters, query strings, and headers.

class ParameterValidator(AbstractParameterValidator):
    """Validator for URL path, query, and header parameters"""
    
    def validate_query_parameter_list(self, request) -> dict:
        """
        Validate query parameters against OpenAPI parameter definitions.
        Handles type coercion, required parameter checking, and format validation.
        """
    
    def validate_path_parameter_list(self, request) -> dict:
        """
        Validate path parameters against OpenAPI parameter definitions.
        Performs type conversion and validation.
        """
    
    def validate_header_parameter_list(self, request) -> dict:
        """
        Validate header parameters against OpenAPI parameter definitions.
        Handles case-insensitive header names and type validation.
        """

Response Validators

Validators for response data validation.

class JSONResponseBodyValidator(AbstractResponseBodyValidator):
    """Validator for JSON response bodies"""
    
    def validate_schema(self, data, url: str) -> dict:
        """
        Validate JSON response data against OpenAPI response schema.
        Ensures API responses match the specification.
        """

Validation Configuration

Configuration options for validation behavior.

class ValidationOptions:
    """Configuration for validation behavior"""
    
    def __init__(
        self,
        strict_validation: bool = False,
        validate_responses: bool = False,
        pythonic_params: bool = False
    ):
        """
        Initialize validation options.
        
        Parameters:
        - strict_validation: Enable strict validation mode
        - validate_responses: Enable response validation
        - pythonic_params: Convert parameter names to pythonic style
        """

Usage Examples

Custom Request Body Validator

from connexion.validators import AbstractRequestBodyValidator
from connexion.exceptions import BadRequestProblem

class CustomJSONValidator(AbstractRequestBodyValidator):
    """Custom JSON validator with additional business rules"""
    
    def validate_schema(self, body, url):
        # First run standard JSON schema validation
        validated_body = super().validate_schema(body, url)
        
        # Add custom validation rules
        if 'email' in validated_body:
            email = validated_body['email']
            if not self.is_valid_email(email):
                raise BadRequestProblem(
                    detail=f"Invalid email format: {email}"
                )
        
        # Check for business rule violations
        if validated_body.get('age', 0) < 0:
            raise BadRequestProblem(
                detail="Age cannot be negative"
            )
        
        return validated_body
    
    def is_valid_email(self, email):
        # Custom email validation logic
        return '@' in email and '.' in email

# Register custom validator
from connexion.validators import VALIDATOR_MAP
VALIDATOR_MAP['body']['application/json'] = CustomJSONValidator

Custom Parameter Validator

from connexion.validators import AbstractParameterValidator
from connexion.exceptions import BadRequestProblem

class EnhancedParameterValidator(AbstractParameterValidator):
    """Parameter validator with enhanced type coercion"""
    
    def validate_query_parameter_list(self, request):
        validated_params = super().validate_query_parameter_list(request)
        
        # Custom parameter processing
        if 'date_range' in validated_params:
            date_range = validated_params['date_range']
            try:
                start, end = date_range.split(',')
                validated_params['start_date'] = datetime.fromisoformat(start)
                validated_params['end_date'] = datetime.fromisoformat(end)
                del validated_params['date_range']
            except ValueError:
                raise BadRequestProblem(
                    detail="Invalid date range format. Use: YYYY-MM-DD,YYYY-MM-DD"
                )
        
        return validated_params

# Use custom validator
app.add_api(
    'api.yaml',
    validator_map={
        'parameter': EnhancedParameterValidator
    }
)

Response Validation

from connexion import AsyncApp

# Enable response validation
app = AsyncApp(__name__, validate_responses=True)
app.add_api('api.yaml')

# Responses that don't match the OpenAPI schema will raise errors
def get_user(user_id):
    # This response must match the OpenAPI response schema
    return {
        "id": user_id,
        "name": "John Doe",
        "email": "john@example.com"
    }

Validation Error Handling

from connexion.exceptions import BadRequestProblem, ValidationError

def custom_validation_error_handler(exception):
    """Handle validation errors with custom responses"""
    
    if isinstance(exception, ValidationError):
        return {
            "error": "Validation failed",
            "details": str(exception),
            "field": getattr(exception, 'field', None)
        }, 400
    
    return {"error": "Bad request"}, 400

app.add_error_handler(ValidationError, custom_validation_error_handler)

Strict Validation Mode

# Enable strict validation for all APIs
app = AsyncApp(__name__, strict_validation=True)

# Or per API
app.add_api('api.yaml', strict_validation=True)

# Strict mode enforces:
# - All required parameters must be present
# - No additional properties in request bodies (if additionalProperties: false)
# - Exact type matching
# - Format validation (email, date, etc.)

Pythonic Parameter Names

# Enable pythonic parameter conversion
app = AsyncApp(__name__, pythonic_params=True)

# OpenAPI parameter: user-id becomes user_id in Python function
def get_user(user_id: int):  # Maps from user-id parameter
    return {"user_id": user_id}

# Query parameter: max-results becomes max_results
def search_users(max_results: int = 10):
    return {"results": [], "limit": max_results}

File Upload Validation

from connexion.validators import MultiPartFormDataValidator

def upload_avatar():
    # Validation automatically handles:
    # - File size limits (if specified in OpenAPI)
    # - Content type validation
    # - Required file fields
    
    file = request.files['avatar']
    
    # Additional custom validation
    if file.content_type not in ['image/jpeg', 'image/png']:
        raise BadRequestProblem(
            detail="Only JPEG and PNG images allowed"
        )
    
    if len(file.read()) > 5 * 1024 * 1024:  # 5MB limit
        raise BadRequestProblem(
            detail="File size must be less than 5MB"
        )
    
    # Process file...
    return {"status": "uploaded"}, 201

Custom Validator Integration

from connexion.validators import VALIDATOR_MAP

# Register custom validators globally
VALIDATOR_MAP['body']['application/xml'] = XMLValidator
VALIDATOR_MAP['body']['text/csv'] = CSVValidator
VALIDATOR_MAP['parameter'] = CustomParameterValidator

# Or use per-API validator mapping
custom_validators = {
    'body': {
        'application/json': CustomJSONValidator,
        'multipart/form-data': CustomMultipartValidator
    },
    'parameter': EnhancedParameterValidator,
    'response': {
        'application/json': StrictJSONResponseValidator
    }
}

app.add_api('api.yaml', validator_map=custom_validators)

Install with Tessl CLI