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

Request Response Handling

Name: tessl/pypi-connexion
Author: tessl

Comprehensive system for processing HTTP requests and responses with automatic validation, serialization, and error handling based on OpenAPI specifications.

Capabilities

Request Context

Access to current request information through the context system.

from connexion.context import request

# Request context properties
class RequestContext:
    @property
    def method(self) -> str:
        """HTTP method (GET, POST, etc.)"""
    
    @property
    def path(self) -> str:
        """Request path"""
    
    @property
    def query(self) -> dict:
        """Query parameters"""
    
    @property
    def headers(self) -> dict:
        """Request headers"""
    
    @property
    def body(self) -> bytes:
        """Raw request body"""
    
    @property
    def json(self) -> dict:
        """Parsed JSON body"""
    
    @property
    def form(self) -> dict:
        """Form data"""
    
    @property
    def files(self) -> dict:
        """Uploaded files"""

ConnexionRequest

Enhanced request object with Connexion-specific functionality.

class ConnexionRequest:
    def __init__(self, *args, uri_parser=None, **kwargs):
        """
        Initialize ConnexionRequest wrapping a Starlette request.
        
        Parameters:
        - *args: Arguments passed to Starlette Request
        - uri_parser: URI parser instance
        - **kwargs: Keyword arguments passed to Starlette Request
        """
    
    @classmethod
    def from_starlette_request(cls, request, uri_parser=None):
        """Create ConnexionRequest from existing Starlette request."""
    
    async def get_body(self):
        """
        Get processed request body based on content type.
        
        Returns:
        JSON dict for JSON content, form dict for form content, or None
        """
    
    async def json(self):
        """
        Parse request body as JSON.
        
        Returns:
        dict: Parsed JSON data or None if invalid
        """
    
    @property
    def content_type(self) -> str:
        """Content-Type header value"""
    
    @property
    def mimetype(self) -> str:
        """Media type without parameters"""
    
    async def form(self) -> dict:
        """Form data from request body"""
    
    async def files(self) -> dict:
        """Uploaded files from multipart request"""
    
    @property
    def query_params(self) -> dict:
        """URL query parameters"""
    
    @property
    def path_params(self) -> dict:
        """Path parameters from URL routing"""
    
    @property
    def context(self) -> dict:
        """Connexion request context"""

ConnexionResponse

Response handling with automatic serialization and content negotiation.

class ConnexionResponse:
    def __init__(
        self,
        body=None,
        status_code: int = 200,
        headers: dict = None,
        content_type: str = None
    ):
        """
        Create a response object.
        
        Parameters:
        - body: Response body (dict, list, str, bytes, or None)
        - status_code: HTTP status code
        - headers: Response headers
        - content_type: Content-Type header
        """
    
    @property
    def status_code(self) -> int:
        """HTTP status code"""
    
    @property
    def headers(self) -> dict:
        """Response headers"""
    
    @property
    def body(self):
        """Response body"""
    
    @property
    def content_type(self) -> str:
        """Content-Type header value"""

Problem Response Handling

RFC 7807 Problem Details for HTTP APIs support.

def problem(
    status: int,
    title: str,
    detail: str = None,
    type: str = None,
    instance: str = None,
    **kwargs
):
    """
    Create an RFC 7807 problem response.
    
    Parameters:
    - status: HTTP status code
    - title: Short, human-readable problem summary
    - detail: Human-readable explanation specific to this occurrence
    - type: URI that identifies the problem type
    - instance: URI that identifies the specific occurrence
    - **kwargs: Additional problem-specific properties
    
    Returns:
    Problem response dict with appropriate status code
    """

NoContent Response

Special response type for empty responses.

class NoContent:
    """
    Represents an empty HTTP response body.
    Used for 204 No Content and similar responses.
    """
    pass

# Usage in endpoint functions
def delete_resource(resource_id: int):
    # Delete the resource
    return NoContent, 204

Media Type Utilities

Utilities for handling media types and content negotiation.

class MediaTypeDict(dict):
    """
    Dictionary subclass for media type mapping with wildcard support.
    """
    
    def best_match(self, supported_types: list) -> str:
        """
        Find best matching media type from supported types.
        
        Parameters:
        - supported_types: List of supported media types
        
        Returns:
        str: Best matching media type or None
        """

Usage Examples

Accessing Request Data

from connexion.context import request

def create_user():
    # Access JSON body
    user_data = request.json
    
    # Access form data
    form_data = request.form
    
    # Access uploaded files
    uploaded_file = request.files.get('avatar')
    
    # Access query parameters
    page = request.args.get('page', 1)
    
    # Access headers
    auth_header = request.headers.get('Authorization')
    
    return {"status": "created"}, 201

Custom Response Handling

from connexion import NoContent
from connexion.lifecycle import ConnexionResponse

def update_user(user_id: int):
    # Return different response types
    if not user_exists(user_id):
        return {"error": "User not found"}, 404
    
    # Update user...
    
    # Return empty response
    return NoContent, 204

def get_user_avatar(user_id: int):
    # Return binary content with custom headers
    avatar_data = get_avatar_bytes(user_id)
    
    return ConnexionResponse(
        body=avatar_data,
        status_code=200,
        headers={'Content-Type': 'image/jpeg'},
        content_type='image/jpeg'
    )

Problem Responses

from connexion import problem

def validate_user_input(user_data):
    if not user_data.get('email'):
        return problem(
            status=400,
            title="Missing Required Field",
            detail="Email address is required",
            type="https://api.example.com/problems/missing-field",
            field="email"
        )
    
    if not is_valid_email(user_data['email']):
        return problem(
            status=400,
            title="Invalid Email Format",
            detail=f"'{user_data['email']}' is not a valid email address",
            type="https://api.example.com/problems/invalid-format",
            field="email",
            value=user_data['email']
        )
    
    return None  # No problems

Content Negotiation

from connexion.lifecycle import ConnexionRequest, ConnexionResponse

def get_data():
    # Check Accept header for content negotiation
    if request.headers.get('Accept') == 'application/xml':
        xml_data = convert_to_xml(data)
        return ConnexionResponse(
            body=xml_data,
            content_type='application/xml'
        )
    
    # Default to JSON
    return {"data": "json format"}

File Upload Handling

def upload_file():
    uploaded_file = request.files.get('file')
    
    if not uploaded_file:
        return problem(400, "No file uploaded")
    
    # Validate file
    if uploaded_file.content_type not in ['image/jpeg', 'image/png']:
        return problem(
            400, 
            "Invalid file type",
            detail="Only JPEG and PNG images are allowed"
        )
    
    # Process file
    file_path = save_uploaded_file(uploaded_file)
    
    return {"file_path": file_path, "size": len(uploaded_file.read())}, 201

Install with Tessl CLI