or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

context-management.mderror-handling.mdindex.mdmiddleware.mdplugins.md
tile.json

tessl/pypi-starlette-context

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/starlette-context@0.4.x

To install, run

npx @tessl/cli install tessl/pypi-starlette-context@0.4.0

index.mddocs/

Starlette Context

A middleware library for Starlette and FastAPI applications that provides request-scoped context storage and access. Enables automatic inclusion of request headers like x-request-id or x-correlation-id in logs and throughout the request lifecycle with plugin-based architecture.

Package Information

  • Package Name: starlette-context
  • Language: Python
  • Installation: pip install starlette-context
  • Requires: Python 3.9+, starlette >= 0.27.0

Core Imports

from starlette_context import context, request_cycle_context

For middleware:

from starlette_context.middleware import ContextMiddleware, RawContextMiddleware

For plugins:

from starlette_context.plugins import (
    Plugin,
    RequestIdPlugin, CorrelationIdPlugin, ApiKeyPlugin,
    UserAgentPlugin, ForwardedForPlugin, DateHeaderPlugin
)

For plugin base classes (not directly exported):

from starlette_context.plugins.base import PluginUUIDBase

For header constants:

from starlette_context.header_keys import HeaderKeys

Basic Usage

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette_context import context
from starlette_context.middleware import ContextMiddleware
from starlette_context.plugins import RequestIdPlugin, CorrelationIdPlugin

async def homepage(request):
    # Access context data like a dictionary
    request_id = context.get("X-Request-ID")
    correlation_id = context["X-Correlation-ID"]
    
    # Store additional data
    context["custom_data"] = "some value"
    
    return JSONResponse({
        "request_id": request_id,
        "correlation_id": correlation_id,
        "context_data": context.data
    })

app = Starlette(routes=[Route("/", homepage)])

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

Architecture

The starlette-context library provides a plugin-based middleware system for request-scoped data management:

  • Context: Global dict-like object accessible throughout request lifecycle
  • Middleware: Two implementations (ContextMiddleware, RawContextMiddleware) that manage context lifecycle
  • Plugins: Extensible components that extract/process headers and enrich responses
  • Request Cycle: Context storage using Python's contextvars for async-safe isolation

The design enables automatic header extraction, context enrichment, response modification, and logging integration while maintaining clean separation between data extraction (plugins) and access (context object).

Capabilities

Context Management

Core functionality for accessing and managing request-scoped context data through a global dictionary-like interface.

context: _Context  # Global context instance

class _Context(UserDict):
    """A mapping with dict-like interface using request context as data store."""
    @property
    def data(self) -> dict:
        """Get context data as dict. Object itself is not serializable."""
    def exists(self) -> bool:
        """Check if context exists in current request cycle."""
    def copy(self) -> dict:
        """Read-only copy of context data."""

@contextmanager
def request_cycle_context(initial_data: Optional[dict] = None) -> Iterator[None]:
    """Context manager for creating request-scoped context."""

Context Management

Middleware Integration

Two middleware implementations for integrating context management into Starlette/FastAPI applications with different performance characteristics.

class ContextMiddleware(BaseHTTPMiddleware):
    def __init__(
        self,
        app: ASGIApp,
        plugins: Optional[Sequence[Plugin]] = None,
        default_error_response: Response = Response(status_code=400)
    ): ...

class RawContextMiddleware:
    def __init__(
        self,
        app: ASGIApp,
        plugins: Optional[Sequence[Plugin]] = None,
        default_error_response: Response = Response(status_code=400)
    ): ...
    async def set_context(self, request: Union[Request, HTTPConnection]) -> dict: ...
    @staticmethod
    def get_request_object(scope: Scope, receive: Receive, send: Send) -> Union[Request, HTTPConnection]: ...
    async def send_response(self, error_response: Response, send: Send) -> None: ...
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: ...

Middleware Integration

Plugin System

Extensible plugin architecture for extracting data from requests and enriching responses, with built-in plugins for common use cases.

class Plugin(metaclass=abc.ABCMeta):
    key: str
    async def process_request(self, request: Union[Request, HTTPConnection]) -> Optional[Any]: ...
    async def enrich_response(self, arg: Union[Response, Message]) -> None: ...

class PluginUUIDBase(Plugin):
    def __init__(
        self,
        force_new_uuid: bool = False,
        version: int = 4,
        validate: bool = True,
        error_response: Optional[Response] = None
    ): ...

Plugin System

Error Handling

Comprehensive error handling for context lifecycle, configuration, and plugin validation with custom response support.

class ContextDoesNotExistError(RuntimeError, StarletteContextError): ...
class ConfigurationError(StarletteContextError): ...
class MiddleWareValidationError(StarletteContextError):
    def __init__(self, *args: Any, error_response: Optional[Response] = None): ...
class WrongUUIDError(MiddleWareValidationError): ...
class DateFormatError(MiddleWareValidationError): ...

Error Handling

Types

from typing import Any, Optional, Union, Sequence, Iterator
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar, Token
from enum import Enum
import abc
import datetime
from starlette.requests import Request, HTTPConnection
from starlette.responses import Response
from starlette.types import ASGIApp, Message, Receive, Scope, Send
from starlette.middleware.base import BaseHTTPMiddleware

class HeaderKeys(str, Enum):
    api_key = "X-API-Key"
    correlation_id = "X-Correlation-ID"
    request_id = "X-Request-ID"
    date = "Date"
    forwarded_for = "X-Forwarded-For"
    user_agent = "User-Agent"