1"""The ContextMiddleware creates a global context based the scope. It should be last in the
2middleware stack, so it exposes the scope passed to the application"""
3from starlette.types import ASGIApp, Receive, Scope, Send
4
5from connexion.context import _context, _operation, _receive, _scope
6from connexion.middleware.abstract import RoutedAPI, RoutedMiddleware
7from connexion.operations import AbstractOperation
8
9
10class ContextOperation:
11 def __init__(
12 self,
13 next_app: ASGIApp,
14 *,
15 operation: AbstractOperation,
16 ) -> None:
17 self.next_app = next_app
18 self.operation = operation
19
20 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
21 _context.set(scope.get("extensions", {}).get("connexion_context", {}))
22 _operation.set(self.operation)
23 _receive.set(receive)
24 _scope.set(scope)
25 await self.next_app(scope, receive, send)
26
27
28class ContextAPI(RoutedAPI[ContextOperation]):
29 def __init__(self, *args, **kwargs) -> None:
30 super().__init__(*args, **kwargs)
31 self.add_paths()
32
33 def make_operation(self, operation: AbstractOperation) -> ContextOperation:
34 return ContextOperation(self.next_app, operation=operation)
35
36
37class ContextMiddleware(RoutedMiddleware[ContextAPI]):
38 """Middleware to expose operation specific context to application."""
39
40 api_cls = ContextAPI