tessl/pypi-starlette-context

Middleware for Starlette that allows you to store and access the context data of a request.

Overview

Eval results

Files

Error Handling

Name: tessl/pypi-starlette-context
Author: tessl

Comprehensive error handling for context lifecycle, configuration, and plugin validation. The starlette-context library provides custom exceptions with support for custom error responses and detailed error messages.

Capabilities

Base Exception Classes

Foundation exception classes that establish the error hierarchy for the library.

class StarletteContextError(Exception):
    """Base exception class for all starlette-context errors."""
    pass

class ContextDoesNotExistError(RuntimeError, StarletteContextError):
    """
    Raised when context is accessed outside of a request-response cycle.
    
    This occurs when:
    - Context is accessed before middleware has created it
    - Context is accessed after request processing is complete
    - Context is accessed in code not running within a request context
    """
    
    def __init__(self) -> None:
        self.message = (
            "You didn't use the required middleware or "
            "you're trying to access `context` object "
            "outside of the request-response cycle."
        )
        super().__init__(self.message)

class ConfigurationError(StarletteContextError):
    """
    Raised for configuration errors in middleware or plugins.
    
    Common causes:
    - Invalid plugin instances passed to middleware
    - Unsupported UUID versions in UUID plugins
    - Invalid middleware configuration
    """
    pass

Middleware Validation Errors

Errors that occur during request processing and can return custom HTTP responses.

class MiddleWareValidationError(StarletteContextError):
    """
    Base class for middleware validation errors.
    
    These errors can include custom HTTP responses that will be returned
    to the client instead of the default error response.
    """
    
    def __init__(
        self, *args: Any, error_response: Optional[Response] = None
    ) -> None:
        """
        Initialize validation error.
        
        Parameters:
        - *args: Exception arguments
        - error_response: Optional custom HTTP response for this error
        """
        super().__init__(*args)
        self.error_response = error_response

class WrongUUIDError(MiddleWareValidationError):
    """
    Raised when UUID validation fails in UUID-based plugins.
    
    Occurs when:
    - Invalid UUID format in request headers  
    - UUID version mismatch
    - Malformed UUID strings
    """
    pass

class DateFormatError(MiddleWareValidationError):
    """
    Raised when date header parsing fails in DateHeaderPlugin.
    
    Occurs when:
    - Invalid RFC1123 date format
    - Non-GMT timezone specified
    - Malformed date strings
    """
    pass

Usage Examples

Context Existence Checking

from starlette_context import context
from starlette_context.errors import ContextDoesNotExistError

def safe_context_access():
    """Safely access context with error handling."""
    try:
        user_id = context["user_id"]
        return f"User: {user_id}"
    except ContextDoesNotExistError:
        return "Context not available"

# Better approach: check existence first
def preferred_safe_access():
    """Preferred way to safely access context."""
    if context.exists():
        return context.get("user_id", "No user ID")
    return "Context not available"

Plugin Configuration Errors

from starlette_context.middleware import ContextMiddleware
from starlette_context.errors import ConfigurationError
from starlette_context.plugins import RequestIdPlugin

try:
    # This will raise ConfigurationError - invalid plugin
    app.add_middleware(
        ContextMiddleware,
        plugins=["not_a_plugin", RequestIdPlugin()]  # String is not a Plugin
    )
except ConfigurationError as e:
    print(f"Configuration error: {e}")
    # Fix: Use only valid Plugin instances
    app.add_middleware(
        ContextMiddleware,
        plugins=[RequestIdPlugin()]
    )

UUID Validation Errors

from starlette_context.plugins import RequestIdPlugin, CorrelationIdPlugin
from starlette_context.errors import WrongUUIDError
from starlette.responses import JSONResponse

# Custom error response for UUID validation
uuid_error_response = JSONResponse(
    {
        "error": "Invalid UUID format",
        "message": "Request ID must be a valid UUID v4",
        "example": "550e8400-e29b-41d4-a716-446655440000"
    },
    status_code=422
)

# Plugin with custom error response
request_id_plugin = RequestIdPlugin(
    validate=True,
    error_response=uuid_error_response
)

app.add_middleware(
    ContextMiddleware,
    plugins=[request_id_plugin]
)

# Test invalid UUID
# Request with header: X-Request-ID: invalid-uuid
# Will return the custom JSON error response

Date Format Errors

from starlette_context.plugins import DateHeaderPlugin
from starlette_context.errors import DateFormatError
from starlette.responses import JSONResponse

# Custom error response for date parsing
date_error_response = JSONResponse(
    {
        "error": "Invalid date format",
        "message": "Date header must be in RFC1123 format",
        "expected_format": "Wed, 01 Jan 2020 04:27:12 GMT",
        "received": None  # Will be filled by middleware
    },
    status_code=422
)

date_plugin = DateHeaderPlugin(error_response=date_error_response)

app.add_middleware(ContextMiddleware, plugins=[date_plugin])

# Test invalid date
# Request with header: Date: invalid-date-format
# Will return the custom JSON error response

Middleware Error Handling

from starlette_context.middleware import ContextMiddleware
from starlette_context.errors import MiddleWareValidationError
from starlette.responses import JSONResponse

# Default error response for all validation errors
default_error = JSONResponse(
    {"error": "Request validation failed"},
    status_code=400
)

app.add_middleware(
    ContextMiddleware,
    plugins=[
        RequestIdPlugin(validate=True),  # No custom error response
        CorrelationIdPlugin(validate=True)  # No custom error response
    ],
    default_error_response=default_error
)

# Validation errors without custom responses will use default_error_response

Custom Plugin Error Handling

from starlette_context.plugins import Plugin
from starlette_context.errors import MiddleWareValidationError
from starlette.responses import JSONResponse

class ApiKeyValidationPlugin(Plugin):
    key = "X-API-Key"
    
    def __init__(self, valid_keys, error_response=None):
        self.valid_keys = set(valid_keys)
        self.error_response = error_response or JSONResponse(
            {"error": "Invalid API key"}, 
            status_code=401
        )
    
    async def process_request(self, request):
        api_key = await self.extract_value_from_header_by_key(request)
        
        if not api_key:
            raise MiddleWareValidationError(
                "Missing API key",
                error_response=JSONResponse(
                    {"error": "API key required"}, 
                    status_code=401
                )
            )
        
        if api_key not in self.valid_keys:
            raise MiddleWareValidationError(
                "Invalid API key",
                error_response=self.error_response
            )
        
        return api_key

# Usage
api_plugin = ApiKeyValidationPlugin(
    valid_keys=["key1", "key2", "key3"]
)

app.add_middleware(ContextMiddleware, plugins=[api_plugin])

Error Response Precedence

The error response precedence order is:

Plugin-specific error response (from exception)
Plugin's configured error response
Middleware's default error response

from starlette_context.plugins import PluginUUIDBase
from starlette_context.errors import WrongUUIDError
from starlette.responses import JSONResponse

class CustomUUIDPlugin(PluginUUIDBase):
    key = "X-Custom-ID"
    
    def __init__(self):
        # Plugin-level error response
        super().__init__(
            validate=True,
            error_response=JSONResponse(
                {"error": "Plugin-level error"}, 
                status_code=422
            )
        )
    
    async def process_request(self, request):
        try:
            return await super().process_request(request)
        except WrongUUIDError:
            # Exception-specific error response (highest precedence)
            raise WrongUUIDError(
                "Custom UUID validation failed",
                error_response=JSONResponse(
                    {"error": "Exception-level error"}, 
                    status_code=400
                )
            )

# Middleware-level default (lowest precedence)
app.add_middleware(
    ContextMiddleware,
    plugins=[CustomUUIDPlugin()],
    default_error_response=JSONResponse(
        {"error": "Middleware-level error"}, 
        status_code=500
    )
)

Logging Errors

import logging
from starlette_context.middleware import ContextMiddleware
from starlette_context.errors import MiddleWareValidationError

logger = logging.getLogger(__name__)

class LoggingContextMiddleware(ContextMiddleware):
    async def dispatch(self, request, call_next):
        try:
            return await super().dispatch(request, call_next)
        except MiddleWareValidationError as e:
            # Log validation errors
            logger.warning(
                f"Context validation error: {e}",
                extra={
                    "path": request.url.path,
                    "method": request.method,
                    "headers": dict(request.headers)
                }
            )
            # Re-raise to return error response
            raise

app.add_middleware(LoggingContextMiddleware, plugins=[...])

Testing Error Scenarios

import pytest
from starlette.testclient import TestClient
from starlette_context.errors import ContextDoesNotExistError
from starlette_context import context

def test_context_without_middleware():
    """Test accessing context without middleware raises error."""
    with pytest.raises(ContextDoesNotExistError):
        value = context["key"]

def test_invalid_uuid_plugin():
    """Test UUID plugin with invalid UUID."""
    client = TestClient(app)  # App with RequestIdPlugin(validate=True)
    
    response = client.get("/", headers={"X-Request-ID": "invalid-uuid"})
    assert response.status_code == 422
    assert "Invalid UUID" in response.json()["error"]

def test_custom_error_response():
    """Test plugin with custom error response."""
    # Setup app with custom error response
    client = TestClient(app)
    
    response = client.get("/", headers={"Date": "invalid-date"})
    assert response.status_code == 422
    assert response.json()["error"] == "Invalid date format"

Best Practices

Error Response Design

# Good: Structured error responses
error_response = JSONResponse(
    {
        "error": "validation_failed",
        "message": "Request validation failed",
        "details": {
            "field": "X-Request-ID",
            "expected": "Valid UUID v4",
            "received": "invalid-format"
        }
    },
    status_code=422
)

# Avoid: Generic or unclear errors
error_response = JSONResponse({"error": "Bad request"}, status_code=400)

Error Handling Strategy

# Context access pattern
def get_user_context():
    """Get user data from context with proper error handling."""
    if not context.exists():
        return None
    
    return {
        "user_id": context.get("user_id"),
        "request_id": context.get("X-Request-ID"),
        "session": context.get("session_id")
    }

# Plugin validation pattern  
class ValidationPlugin(Plugin):
    def __init__(self, error_response=None):
        self.error_response = error_response or self.default_error_response
    
    @property
    def default_error_response(self):
        return JSONResponse(
            {"error": f"Invalid {self.key} header"},
            status_code=422
        )
    
    async def process_request(self, request):
        value = await self.extract_value_from_header_by_key(request)
        if not self.validate_value(value):
            raise MiddleWareValidationError(
                f"Invalid {self.key}: {value}",
                error_response=self.error_response
            )
        return value

Install with Tessl CLI

npx tessl i tessl/pypi-starlette-context

docs

context-management.md