ASGI Middleware

@dataclass
class CorrelationIdMiddleware:
    """
    ASGI middleware for correlation ID management.
    
    Parameters:
    - app: The ASGI application to wrap
    - header_name: Header name to look for correlation ID (default: 'X-Request-ID')
    - update_request_header: Whether to update request headers with correlation ID
    - generator: Function to generate new correlation IDs (default: uuid4().hex)
    - validator: Function to validate existing correlation IDs (default: is_valid_uuid4)
    - transformer: Function to transform correlation IDs (default: identity function)
    """
    app: ASGIApp
    header_name: str = 'X-Request-ID'
    update_request_header: bool = True
    generator: Callable[[], str] = field(default=lambda: uuid4().hex)
    validator: Optional[Callable[[str], bool]] = field(default=is_valid_uuid4)
    transformer: Optional[Callable[[str], str]] = field(default=lambda a: a)
    
    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        Process ASGI request with correlation ID handling.
        
        For HTTP and WebSocket connections:
        1. Extracts correlation ID from request headers
        2. Validates ID using configured validator
        3. Generates new ID if none found or validation fails
        4. Transforms ID using configured transformer
        5. Sets correlation ID in context variable
        6. Updates response headers with correlation ID
        7. Integrates with Sentry and Celery extensions if available
        """

def is_valid_uuid4(uuid_: str) -> bool:
    """
    Check whether a string is a valid v4 uuid.
    
    Parameters:
    - uuid_: String to validate as UUID v4
    
    Returns:
    - bool: True if string is valid UUID v4, False otherwise
    
    Raises:
    - No exceptions (catches ValueError internally)
    """

from fastapi import FastAPI
from asgi_correlation_id import CorrelationIdMiddleware

app = FastAPI()
app.add_middleware(CorrelationIdMiddleware)

from asgi_correlation_id import CorrelationIdMiddleware
import uuid

def custom_generator():
    return f"req-{uuid.uuid4().hex[:8]}"

def custom_validator(id_value):
    return id_value.startswith("req-") and len(id_value) == 12

app.add_middleware(
    CorrelationIdMiddleware,
    header_name='X-Correlation-ID',
    update_request_header=True,
    generator=custom_generator,
    validator=custom_validator,
    transformer=lambda x: x.upper()
)

from starlette.applications import Starlette
from starlette.middleware import Middleware
from asgi_correlation_id import CorrelationIdMiddleware

middleware = [
    Middleware(CorrelationIdMiddleware, header_name='X-Request-ID')
]

app = Starlette(middleware=middleware)

from typing import TYPE_CHECKING, Callable, Optional
from dataclasses import dataclass, field

if TYPE_CHECKING:
    from starlette.types import ASGIApp, Message, Receive, Scope, Send