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

Operation Resolution

Name: tessl/pypi-connexion
Author: tessl

System for mapping OpenAPI operations to Python functions using various resolution strategies. Resolvers provide flexible ways to connect OpenAPI operation definitions to your Python implementation code.

Capabilities

Base Resolver

Basic operation resolver that maps operations to functions based on operationId.

class Resolver:
    def __init__(self, function_resolver=None):
        """
        Initialize basic resolver.
        
        Parameters:
        - function_resolver: Custom function resolver callable
        """
    
    def resolve(self, operation):
        """
        Resolve an operation to a Python function.
        
        Parameters:
        - operation: Operation object to resolve
        
        Returns:
        Resolution object containing function and metadata
        """
    
    def resolve_function_from_operation_id(self, operation_id: str):
        """
        Resolve function directly from operation ID.
        
        Parameters:
        - operation_id: OpenAPI operation ID
        
        Returns:
        Python function or None if not found
        """
    
    def resolve_operation_id(self, operation):
        """
        Get operation ID from operation definition.
        
        Parameters:
        - operation: Operation object
        
        Returns:
        str: Operation ID
        """

RestyResolver

RESTful operation resolver that maps operations to functions based on HTTP method and path patterns.

class RestyResolver(Resolver):
    def __init__(
        self,
        default_module_name: str,
        collection_endpoint_name: str = "search"
    ):
        """
        Initialize RESTful resolver.
        
        Parameters:
        - default_module_name: Default module to search for functions
        - collection_endpoint_name: Name for collection endpoints
        """
    
    def resolve_operation_id(self, operation):
        """
        Generate operation ID based on HTTP method and path.
        
        Uses RESTful conventions:
        - GET /users -> get_users
        - GET /users/{id} -> get_user
        - POST /users -> post_users
        - PUT /users/{id} -> put_user
        - DELETE /users/{id} -> delete_user
        
        Returns:
        str: Generated operation ID
        """

MethodResolver

Resolver for method-based operation handling, typically used with class-based views.

class MethodResolver(Resolver):
    def __init__(self, api_name: str, **kwargs):
        """
        Initialize method resolver.
        
        Parameters:
        - api_name: Name of the API for method resolution
        - **kwargs: Additional resolver options
        """
    
    def resolve_operation_id_using_rest_semantics(self, operation):
        """
        Resolve operation ID using REST semantic conventions.
        
        Parameters:
        - operation: Operation object
        
        Returns:
        str: Method-based operation ID
        """

MethodViewResolver

Resolver for Flask MethodView-style class-based views.

class MethodViewResolver(MethodResolver):
    def __init__(self, api_name: str, **kwargs):
        """
        Initialize MethodView resolver for Flask compatibility.
        
        Parameters:
        - api_name: API name for view resolution
        - **kwargs: Additional resolver options
        """

Resolution Result

Container for resolved operation information.

class Resolution:
    def __init__(self, function, operation_id: str):
        """
        Initialize resolution result.
        
        Parameters:
        - function: Resolved Python function
        - operation_id: Operation identifier
        """
    
    @property
    def function(self):
        """Resolved Python function"""
    
    @property
    def operation_id(self) -> str:
        """Operation identifier"""

Custom Function Resolvers

Support for custom function resolution logic.

def default_function_resolver(function_name: str):
    """
    Default function resolver implementation.
    
    Parameters:
    - function_name: Name of function to resolve
    
    Returns:
    Function object or None if not found
    """

# Custom resolver example
def my_custom_resolver(function_name: str):
    """Custom function resolution logic"""
    # Your custom logic here
    pass

Usage Examples

Basic Resolver Usage

from connexion import AsyncApp
from connexion.resolver import Resolver

# Use default resolver (looks for operationId in current module)
app = AsyncApp(__name__)
app.add_api('openapi.yaml')  # Uses default Resolver

# Custom resolver with specific function resolver
def custom_function_resolver(function_name):
    # Custom logic to find functions
    return my_functions.get(function_name)

custom_resolver = Resolver(function_resolver=custom_function_resolver)
app.add_api('api.yaml', resolver=custom_resolver)

RESTful Resolution

from connexion.resolver import RestyResolver

# Create RESTful resolver
resolver = RestyResolver('api.handlers')

app = AsyncApp(__name__)
app.add_api('restful-api.yaml', resolver=resolver)

# With custom collection endpoint name
resolver = RestyResolver(
    'api.handlers',
    collection_endpoint_name='list'
)
app.add_api('api.yaml', resolver=resolver)

Method-Based Resolution

from connexion.resolver import MethodResolver

# For class-based views
resolver = MethodResolver('UserAPI')

app = AsyncApp(__name__)
app.add_api('users-api.yaml', resolver=resolver)

# Implementation class
class UserAPI:
    def get(self, user_id=None):
        if user_id:
            return get_user(user_id)
        return list_users()
    
    def post(self):
        return create_user(request.json)
    
    def put(self, user_id):
        return update_user(user_id, request.json)
    
    def delete(self, user_id):
        delete_user(user_id)
        return NoContent, 204

Custom Resolution Strategy

from connexion.resolver import Resolver

class DatabaseResolver(Resolver):
    """Resolver that maps operations to database-stored procedures"""
    
    def __init__(self, db_connection):
        self.db = db_connection
        super().__init__()
    
    def resolve_function_from_operation_id(self, operation_id):
        # Look up stored procedure by operation ID
        procedure = self.db.get_procedure(operation_id)
        if procedure:
            return lambda **kwargs: procedure.execute(**kwargs)
        return None

# Usage
db_resolver = DatabaseResolver(my_db_connection)
app.add_api('database-api.yaml', resolver=db_resolver)

Module-Based Organization

from connexion.resolver import RestyResolver

# Organize handlers in modules
# api/
#   users.py      - User operations
#   products.py   - Product operations
#   orders.py     - Order operations

resolver = RestyResolver('api')

# OpenAPI spec maps to functions like:
# GET /users -> api.users.get_users()
# POST /users -> api.users.post_users()
# GET /users/{id} -> api.users.get_user(id)
# GET /products -> api.products.get_products()

app.add_api('ecommerce-api.yaml', resolver=resolver)

Resolver Error Handling

from connexion.exceptions import ResolverError

def custom_resolver_error_handler(exception):
    """Handle resolver errors gracefully"""
    if isinstance(exception, ResolverError):
        return {
            "error": "Operation not implemented",
            "operation": exception.operation_id
        }, 501
    
    return {"error": "Internal server error"}, 500

app.add_api(
    'api.yaml',
    resolver=RestyResolver('handlers'),
    resolver_error_handler=custom_resolver_error_handler
)

Dynamic Function Loading

import importlib
from connexion.resolver import Resolver

class DynamicResolver(Resolver):
    """Resolver that dynamically loads functions from modules"""
    
    def __init__(self, module_prefix='api.v1'):
        self.module_prefix = module_prefix
        super().__init__()
    
    def resolve_function_from_operation_id(self, operation_id):
        # Split operation_id to get module and function
        # e.g., "users_get_user" -> module="users", func="get_user"
        parts = operation_id.split('_', 1)
        if len(parts) != 2:
            return None
        
        module_name, func_name = parts
        full_module = f"{self.module_prefix}.{module_name}"
        
        try:
            module = importlib.import_module(full_module)
            return getattr(module, func_name, None)
        except (ImportError, AttributeError):
            return None

# Usage
dynamic_resolver = DynamicResolver('api.v2')
app.add_api('dynamic-api.yaml', resolver=dynamic_resolver)

Install with Tessl CLI