tessl/pypi-falcon

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

—

Pending

Overview

Eval results

Files

Middleware and Hooks

Name: tessl/pypi-falcon
Author: tessl

Middleware system for implementing cross-cutting concerns like authentication, CORS, and logging, combined with hook decorators for before/after request processing in Falcon applications.

Capabilities

Middleware System

Pluggable middleware components that process requests and responses across all routes.

# Middleware interface (informal protocol)
class MiddlewareComponent:
    def process_request(self, req: Request, resp: Response):
        """
        Process request before routing to resource.
        
        Args:
            req: Request object
            resp: Response object (can be modified)
        """
    
    def process_resource(self, req: Request, resp: Response, resource: object, params: dict):
        """
        Process request after routing but before calling resource method.
        
        Args:
            req: Request object
            resp: Response object
            resource: Target resource object
            params: Route parameters
        """
    
    def process_response(self, req: Request, resp: Response, resource: object, req_succeeded: bool):
        """
        Process response after resource method execution.
        
        Args:
            req: Request object
            resp: Response object  
            resource: Resource object that handled the request
            req_succeeded: Whether request processing succeeded
        """

Middleware Registration

import falcon

class AuthMiddleware:
    def process_request(self, req, resp):
        # Check authentication
        token = req.get_header('Authorization')
        if not token:
            raise falcon.HTTPUnauthorized(
                title='Authentication required',
                description='Please provide valid authentication token'
            )
        
        # Validate token and set user context
        user = validate_token(token)
        req.context.user = user

class LoggingMiddleware:
    def process_request(self, req, resp):
        req.context.start_time = time.time()
        print(f"Request: {req.method} {req.path}")
    
    def process_response(self, req, resp, resource, req_succeeded):
        duration = time.time() - req.context.start_time
        print(f"Response: {resp.status} ({duration:.3f}s)")

# Register middleware
app = falcon.App(middleware=[
    AuthMiddleware(),
    LoggingMiddleware()
])

# Or add middleware after app creation
app.add_middleware(AuthMiddleware())

Built-in CORS Middleware

Comprehensive CORS (Cross-Origin Resource Sharing) middleware with configurable policies.

class CORSMiddleware:
    def __init__(
        self,
        allow_origins: str | list = '*',
        expose_headers: list = None,
        allow_credentials: bool = None,
        allow_private_network: bool = False,
        allow_methods: list = None,
        allow_headers: list = None,
        max_age: int = None
    ):
        """
        CORS middleware for handling cross-origin requests.
        
        Args:
            allow_origins: Allowed origins ('*' or list of origins)
            expose_headers: Headers to expose to client
            allow_credentials: Allow credentials in CORS requests
            allow_private_network: Allow private network access
            allow_methods: Allowed HTTP methods (defaults to all)
            allow_headers: Allowed request headers
            max_age: Preflight cache time in seconds
        """
    
    def process_request(self, req: Request, resp: Response):
        """Process CORS preflight requests"""
    
    def process_response(self, req: Request, resp: Response, resource: object, req_succeeded: bool):
        """Add CORS headers to responses"""

CORS Middleware Usage

import falcon

# Basic CORS setup
cors = falcon.CORSMiddleware(
    allow_origins=['https://example.com', 'https://app.example.com'],
    allow_credentials=True
)

app = falcon.App(middleware=[cors])

# Or enable CORS with app creation
app = falcon.App(cors_enable=True)  # Uses default CORS settings

# Advanced CORS configuration
cors = falcon.CORSMiddleware(
    allow_origins='*',
    expose_headers=['X-Custom-Header'],
    allow_credentials=False,
    allow_methods=['GET', 'POST', 'PUT', 'DELETE'],
    allow_headers=['Content-Type', 'Authorization'],
    max_age=86400  # 24 hours
)

Hook Decorators

Decorators for executing functions before and after resource method execution.

def before(action: callable, *args, **kwargs) -> callable:
    """
    Execute function before responder method.
    
    Args:
        action: Function to execute before responder
        *args: Arguments to pass to action function
        **kwargs: Keyword arguments to pass to action function
        
    Returns:
        Decorator function
    """

def after(action: callable, *args, **kwargs) -> callable:
    """
    Execute function after responder method.
    
    Args:
        action: Function to execute after responder
        *args: Arguments to pass to action function  
        **kwargs: Keyword arguments to pass to action function
        
    Returns:
        Decorator function
    """

# Hook action function signature
def hook_action(req: Request, resp: Response, resource: object, params: dict, *args, **kwargs):
    """
    Hook action function signature.
    
    Args:
        req: Request object
        resp: Response object
        resource: Resource object being processed
        params: Route parameters
        *args: Additional positional arguments
        **kwargs: Additional keyword arguments
    """

Hook Usage Examples

import falcon
from falcon import before, after

def validate_user_input(req, resp, resource, params):
    """Validate request data before processing"""
    if req.content_length and req.content_length > 1024 * 1024:  # 1MB limit
        raise falcon.HTTPContentTooLarge(
            title='Request too large',
            description='Request body cannot exceed 1MB'
        )

def log_user_action(req, resp, resource, params, action_name):
    """Log user actions after processing"""
    user_id = req.context.get('user_id')
    if user_id:
        log_action(user_id, action_name, params)

def cache_response(req, resp, resource, params, cache_time=300):
    """Set cache headers after processing"""
    if resp.status.startswith('2'):  # Success responses only
        resp.set_header('Cache-Control', f'max-age={cache_time}')

class UserResource:
    @before(validate_user_input)
    @after(log_user_action, 'user_create')
    @after(cache_response, cache_time=600)
    def on_post(self, req, resp):
        """Create new user with validation and logging"""
        user_data = req.media
        new_user = create_user(user_data)
        resp.status = falcon.HTTP_201
        resp.media = new_user
    
    @after(log_user_action, 'user_view')
    def on_get(self, req, resp, user_id):
        """Get user with action logging"""
        user = get_user(user_id)
        if not user:
            raise falcon.HTTPNotFound()
        resp.media = user

Custom Middleware Examples

Common middleware implementation patterns for various use cases.

Authentication Middleware

import jwt
import falcon

class JWTMiddleware:
    def __init__(self, secret_key, algorithm='HS256', exempt_routes=None):
        self.secret_key = secret_key
        self.algorithm = algorithm
        self.exempt_routes = exempt_routes or []
    
    def process_request(self, req, resp):
        # Skip authentication for exempt routes
        if req.path in self.exempt_routes:
            return
        
        # Extract JWT token
        auth_header = req.get_header('Authorization')
        if not auth_header or not auth_header.startswith('Bearer '):
            raise falcon.HTTPUnauthorized(
                title='Missing authentication',
                description='Bearer token required in Authorization header'
            )
        
        token = auth_header[7:]  # Remove 'Bearer ' prefix
        
        try:
            # Decode and validate JWT
            payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
            req.context.user = payload
        except jwt.ExpiredSignatureError:
            raise falcon.HTTPUnauthorized(
                title='Token expired',
                description='Authentication token has expired'
            )
        except jwt.InvalidTokenError:
            raise falcon.HTTPUnauthorized(
                title='Invalid token',
                description='Authentication token is invalid'
            )

# Usage
jwt_middleware = JWTMiddleware(
    secret_key='your-secret-key',
    exempt_routes=['/health', '/login', '/register']
)
app = falcon.App(middleware=[jwt_middleware])

Request/Response Logging Middleware

import json
import logging
import time
import falcon

class DetailedLoggingMiddleware:
    def __init__(self, logger=None, log_body=False):
        self.logger = logger or logging.getLogger(__name__)
        self.log_body = log_body
    
    def process_request(self, req, resp):
        req.context.start_time = time.time()
        
        # Log request details
        self.logger.info(f"Request: {req.method} {req.path}")
        self.logger.debug(f"Headers: {dict(req.headers)}")
        self.logger.debug(f"Query params: {req.params}")
        
        if self.log_body and req.content_length:
            try:
                # Log request body (be careful with sensitive data)
                body = req.bounded_stream.read()
                req.bounded_stream = io.BytesIO(body)  # Reset stream
                self.logger.debug(f"Request body: {body.decode('utf-8')[:1000]}")
            except Exception as e:
                self.logger.warning(f"Could not log request body: {e}")
    
    def process_response(self, req, resp, resource, req_succeeded):
        duration = time.time() - req.context.start_time
        
        # Log response details
        self.logger.info(f"Response: {resp.status} ({duration:.3f}s)")
        
        if not req_succeeded:
            self.logger.error(f"Request failed for {req.method} {req.path}")

# Usage
logging_middleware = DetailedLoggingMiddleware(
    logger=logging.getLogger('falcon.requests'),
    log_body=True
)

Rate Limiting Middleware

import time
from collections import defaultdict, deque
import falcon

class RateLimitMiddleware:
    def __init__(self, calls_per_minute=100, per_ip=True):
        self.calls_per_minute = calls_per_minute
        self.per_ip = per_ip
        self.call_history = defaultdict(deque)
    
    def _get_client_id(self, req):
        """Get client identifier for rate limiting"""
        if self.per_ip:
            # Use real IP if behind proxy
            return req.get_header('X-Forwarded-For') or req.env.get('REMOTE_ADDR')
        else:
            # Use user ID from context if authenticated
            return getattr(req.context, 'user_id', 'anonymous')
    
    def process_request(self, req, resp):
        client_id = self._get_client_id(req)
        now = time.time()
        
        # Clean old entries (older than 1 minute)
        history = self.call_history[client_id]
        while history and history[0] < now - 60:
            history.popleft()
        
        # Check rate limit
        if len(history) >= self.calls_per_minute:
            raise falcon.HTTPTooManyRequests(
                title='Rate limit exceeded',
                description=f'Maximum {self.calls_per_minute} calls per minute allowed'
            )
        
        # Record this call
        history.append(now)

# Usage
rate_limiter = RateLimitMiddleware(calls_per_minute=60, per_ip=True)

Async Middleware

Middleware support for ASGI applications with async processing.

# Async middleware interface
class AsyncMiddlewareComponent:
    async def process_request(self, req: Request, resp: Response):
        """Async request processing"""
    
    async def process_resource(self, req: Request, resp: Response, resource: object, params: dict):
        """Async resource processing"""
    
    async def process_response(self, req: Request, resp: Response, resource: object, req_succeeded: bool):
        """Async response processing"""

Async Middleware Example

import falcon.asgi
import asyncio
import aioredis

class AsyncCacheMiddleware:
    def __init__(self, redis_url='redis://localhost'):
        self.redis = None
    
    async def process_request(self, req, resp):
        # Initialize Redis connection if needed
        if not self.redis:
            self.redis = await aioredis.from_url('redis://localhost')
        
        # Check cache for GET requests
        if req.method == 'GET':
            cache_key = f"cache:{req.path}:{req.query_string}"
            cached_response = await self.redis.get(cache_key)
            
            if cached_response:
                resp.media = json.loads(cached_response)
                resp.complete = True  # Skip further processing
    
    async def process_response(self, req, resp, resource, req_succeeded):
        # Cache successful GET responses
        if (req.method == 'GET' and req_succeeded and 
            resp.status.startswith('2') and hasattr(resp, 'media')):
            
            cache_key = f"cache:{req.path}:{req.query_string}"
            await self.redis.setex(
                cache_key, 
                300,  # 5 minutes TTL
                json.dumps(resp.media)
            )

# Usage with ASGI app
app = falcon.asgi.App(middleware=[AsyncCacheMiddleware()])

Error Handling in Middleware

Best practices for error handling within middleware components.

class SafeMiddleware:
    def process_request(self, req, resp):
        try:
            # Middleware logic here
            self._process_request_logic(req, resp)
        except Exception as e:
            # Log error and optionally convert to HTTP error
            logger.error(f"Middleware error: {e}")
            
            # Re-raise as HTTP error for proper handling
            raise falcon.HTTPInternalServerError(
                title='Middleware processing failed',
                description='An internal error occurred while processing the request'
            )
    
    def _process_request_logic(self, req, resp):
        """Actual middleware logic"""
        pass

Types

# Middleware system (informal protocols - no concrete types)
MiddlewareComponent: protocol  # Middleware interface pattern

# Built-in middleware
CORSMiddleware: type

# Hook decorators
before: callable  # Before hook decorator
after: callable   # After hook decorator

# Hook action signature (function type)
HookAction: callable  # (req, resp, resource, params, *args, **kwargs) -> None

Install with Tessl CLI