Skip to content

Middleware API Reference

Middleware protocol, stack, request scope, and built-in middlewares.

Middleware pipeline for cross-cutting concerns.

CorrelationIdMiddleware

Ensures every request has a correlation ID.

Source code in src/istos/middleware/base.py 
123
124
125
126
127
128
129
130
131
132
133
class CorrelationIdMiddleware:
    """Ensures every request has a correlation ID."""

    async def __call__(
        self,
        scope: RequestScope,
        call_next: HandlerCallable,
    ) -> Any:
        if not scope.context.correlation_id:
            scope.context = RequestContext(prefix=scope.prefix, operation=scope.operation)
        return await call_next(scope)

LoggingMiddleware

Logs request start/end with duration and correlation ID.

Source code in src/istos/middleware/base.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class LoggingMiddleware:
    """Logs request start/end with duration and correlation ID."""

    def __init__(self, logger: Any) -> None:
        self._logger = logger

    async def __call__(
        self,
        scope: RequestScope,
        call_next: HandlerCallable,
    ) -> Any:
        ctx = get_request_context()
        start = time.perf_counter()
        self._logger.info(
            "%s %s started",
            scope.operation, scope.prefix,
            extra={
                "correlation_id": ctx.correlation_id,
                "prefix": scope.prefix,
                "operation": scope.operation,
            },
        )
        try:
            result = await call_next(scope)
            duration_ms = (time.perf_counter() - start) * 1000
            self._logger.info(
                "%s %s completed in %.2fms",
                scope.operation, scope.prefix, duration_ms,
                extra={
                    "correlation_id": ctx.correlation_id,
                    "prefix": scope.prefix,
                    "operation": scope.operation,
                    "duration_ms": round(duration_ms, 2),
                },
            )
            return result
        except Exception as exc:
            duration_ms = (time.perf_counter() - start) * 1000
            self._logger.error(
                "%s %s failed after %.2fms: %s",
                scope.operation, scope.prefix, duration_ms, exc,
                extra={
                    "correlation_id": ctx.correlation_id,
                    "prefix": scope.prefix,
                    "operation": scope.operation,
                    "duration_ms": round(duration_ms, 2),
                    "error": str(exc),
                },
            )
            raise

Middleware

Bases: Protocol

Middleware processes requests before/after handler execution.

Source code in src/istos/middleware/base.py 
25
26
27
28
29
30
31
32
33
34
@runtime_checkable
class Middleware(Protocol):
    """Middleware processes requests before/after handler execution."""

    async def __call__(
        self,
        scope: RequestScope,
        call_next: HandlerCallable,
    ) -> Any:
        ...

MiddlewareStack

Composes middleware into a single callable.

Source code in src/istos/middleware/base.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class MiddlewareStack:
    """Composes middleware into a single callable."""

    def __init__(self, middlewares: Optional[List[Middleware]] = None) -> None:
        self._middlewares = list(middlewares or [])

    def add(self, middleware: Middleware) -> None:
        self._middlewares.append(middleware)

    async def invoke(
        self,
        scope: RequestScope,
        handler: HandlerCallable,
    ) -> Any:
        set_request_context(scope.context)
        scope.context.prefix = scope.prefix
        scope.context.operation = scope.operation

        async def dispatch(index: int) -> Any:
            if index >= len(self._middlewares):
                return await handler(scope)
            middleware = self._middlewares[index]

            async def call_next(next_scope: RequestScope) -> Any:
                return await dispatch(index + 1)

            return await middleware(scope, call_next)

        try:
            return await dispatch(0)
        finally:
            pass

RequestScope dataclass

Metadata passed through the middleware chain.

Source code in src/istos/middleware/base.py 
12
13
14
15
16
17
18
19
@dataclass
class RequestScope:
    """Metadata passed through the middleware chain."""

    prefix: str
    operation: str  # "handle", "subscribe", "publish", "query"
    params: dict[str, Any] = field(default_factory=dict)
    context: RequestContext = field(default_factory=RequestContext)