tessl/pypi-starlette-context

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

Overview

Eval results

Files

Middleware Integration

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

Two middleware implementations for integrating context management into Starlette/FastAPI applications. Both create and manage request-scoped context, but differ in their integration approach and performance characteristics.

Capabilities

ContextMiddleware

HTTP middleware that extends Starlette's BaseHTTPMiddleware, providing easier integration and debugging at the cost of some performance overhead.

class ContextMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: ASGIApp,
        plugins: Optional[Sequence[Plugin]] = None,
        default_error_response: Response = Response(status_code=400),
        *args: Any,
        **kwargs: Any
    ) -> None:
        """
        Middleware that creates empty context for request it's used on.
        
        Parameters:
        - app: ASGI application
        - plugins: Optional sequence of plugins to process requests
        - default_error_response: Response to return on plugin validation errors
        - *args, **kwargs: Additional arguments passed to BaseHTTPMiddleware
        """
    
    async def set_context(self, request: Request) -> dict:
        """
        You might want to override this method.
        
        The dict it returns will be saved in the scope of a context. You can
        always do that later.
        
        Parameters:
        - request: Starlette Request object
        
        Returns:
        dict: Initial context data from plugins
        """
    
    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        """
        Process request with context lifecycle management.
        
        Parameters:
        - request: Starlette Request object
        - call_next: Next middleware/handler in chain
        
        Returns:
        Response: HTTP response
        
        Raises:
        MiddleWareValidationError: If plugin validation fails
        """

RawContextMiddleware

Low-level ASGI middleware that operates directly on ASGI scope, providing better performance by avoiding BaseHTTPMiddleware's overhead.

class RawContextMiddleware:
    def __init__(
        self,
        app: ASGIApp,
        plugins: Optional[Sequence[Plugin]] = None,
        default_error_response: Response = Response(status_code=400)
    ) -> None:
        """
        Raw ASGI middleware for context management.
        
        Parameters:
        - app: ASGI application
        - plugins: Optional sequence of plugins to process requests
        - default_error_response: Response to return on plugin validation errors
        """
    
    async def set_context(
        self, request: Union[Request, HTTPConnection]
    ) -> dict:
        """
        You might want to override this method.
        
        The dict it returns will be saved in the scope of a context. You can
        always do that later.
        
        Parameters:
        - request: Request or HTTPConnection object
        
        Returns:
        dict: Initial context data from plugins
        """
    
    @staticmethod
    def get_request_object(
        scope: Scope, receive: Receive, send: Send
    ) -> Union[Request, HTTPConnection]:
        """
        Create request object from ASGI components.
        
        Parameters:
        - scope: ASGI scope dict
        - receive: ASGI receive callable
        - send: ASGI send callable
        
        Returns:
        Union[Request, HTTPConnection]: Request object for header access
        """
    
    async def send_response(
        self, error_response: Response, send: Send
    ) -> None:
        """
        Send error response via ASGI send interface.
        
        Parameters:
        - error_response: Response object to send
        - send: ASGI send callable
        """
    
    async def __call__(
        self, scope: Scope, receive: Receive, send: Send
    ) -> None:
        """
        ASGI application interface.
        
        Parameters:
        - scope: ASGI scope dict
        - receive: ASGI receive callable  
        - send: ASGI send callable
        """

Usage Examples

Basic ContextMiddleware Setup

from starlette.applications import Starlette
from starlette_context.middleware import ContextMiddleware
from starlette_context.plugins import RequestIdPlugin, CorrelationIdPlugin

app = Starlette()

# Add context middleware with plugins
app.add_middleware(
    ContextMiddleware,
    plugins=[
        RequestIdPlugin(),
        CorrelationIdPlugin(force_new_uuid=True)
    ]
)

RawContextMiddleware Setup

from starlette.applications import Starlette
from starlette_context.middleware import RawContextMiddleware
from starlette_context.plugins import RequestIdPlugin, UserAgentPlugin

app = Starlette()

# Add raw context middleware for better performance
app.add_middleware(
    RawContextMiddleware,
    plugins=[
        RequestIdPlugin(),
        UserAgentPlugin()
    ]
)

Custom Error Response

from starlette.responses import JSONResponse
from starlette_context.middleware import ContextMiddleware
from starlette_context.plugins import DateHeaderPlugin

# Custom error response for validation failures
error_response = JSONResponse(
    {"error": "Invalid request format"}, 
    status_code=422
)

app.add_middleware(
    ContextMiddleware,
    plugins=[DateHeaderPlugin()],
    default_error_response=error_response
)

Custom Context Setup

Override set_context to customize initial context data:

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

class CustomContextMiddleware(ContextMiddleware):
    async def set_context(self, request):
        # Get plugin data
        context_data = await super().set_context(request)
        
        # Add custom data
        context_data.update({
            "request_path": request.url.path,
            "request_method": request.method,
            "timestamp": time.time()
        })
        
        return context_data

app.add_middleware(
    CustomContextMiddleware,
    plugins=[RequestIdPlugin()]
)

FastAPI Integration

from fastapi import FastAPI
from starlette_context.middleware import ContextMiddleware
from starlette_context.plugins import RequestIdPlugin, CorrelationIdPlugin
from starlette_context import context

app = FastAPI()

# Add middleware
app.add_middleware(
    ContextMiddleware,
    plugins=[
        RequestIdPlugin(),
        CorrelationIdPlugin()
    ]
)

@app.get("/")
async def root():
    # Context is automatically available
    return {
        "request_id": context["X-Request-ID"],
        "correlation_id": context["X-Correlation-ID"]
    }

Multiple Middleware Layers

from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette_context.middleware import ContextMiddleware
from starlette_context.plugins import RequestIdPlugin

# Define middleware stack
middleware = [
    Middleware(
        ContextMiddleware,
        plugins=[RequestIdPlugin()]
    ),
    # Other middleware...
]

app = Starlette(middleware=middleware)

Performance Considerations

ContextMiddleware vs RawContextMiddleware

ContextMiddleware:

Easier to use and debug
Integrates with Starlette's middleware system
Slight performance overhead from BaseHTTPMiddleware
Better for development and moderate traffic

RawContextMiddleware:

Better performance (no BaseHTTPMiddleware overhead)
Direct ASGI integration
More complex to debug
Better for high-performance production environments

Plugin Performance

# Efficient: Only essential plugins
app.add_middleware(
    RawContextMiddleware,
    plugins=[RequestIdPlugin()]  # Minimal overhead
)

# Less efficient: Many complex plugins
app.add_middleware(
    ContextMiddleware,
    plugins=[
        RequestIdPlugin(),
        CorrelationIdPlugin(validate=True),  # UUID validation
        DateHeaderPlugin(),                  # Date parsing
        ApiKeyPlugin(),
        UserAgentPlugin(),
        ForwardedForPlugin()
    ]
)

Error Handling

Both middleware types handle plugin validation errors:

class MiddleWareValidationError(StarletteContextError):
    def __init__(self, *args: Any, error_response: Optional[Response] = None):
        """
        Base exception for middleware validation errors.
        
        Parameters:
        - *args: Exception arguments
        - error_response: Optional custom response for this error
        """
    
    error_response: Optional[Response]  # Custom error response

Error handling flow:

try:
    # Plugin processes request
    context_data = await plugin.process_request(request)
except MiddleWareValidationError as e:
    # Return plugin's custom error response or middleware default
    return e.error_response or self.default_error_response

Middleware Execution Flow

Request arrives: Middleware intercepts incoming request
Plugin processing: Each plugin extracts/validates data from request
Context creation: Context is created with plugin data using request_cycle_context
Request processing: Application handlers execute with context available
Response enrichment: Plugins can modify outgoing response
Context cleanup: Context is automatically reset after request completes

# Middleware execution order
async def middleware_flow(request):
    try:
        # 1. Process plugins
        context_data = {}
        for plugin in plugins:
            context_data[plugin.key] = await plugin.process_request(request)
        
        # 2. Create context scope
        with request_cycle_context(context_data):
            # 3. Process request
            response = await call_next(request)
            
            # 4. Enrich response
            for plugin in plugins:
                await plugin.enrich_response(response)
                
            return response
        # 5. Context automatically cleaned up
    except MiddleWareValidationError as e:
        return e.error_response or default_error_response

Install with Tessl CLI

npx tessl i tessl/pypi-starlette-context

docs

context-management.md