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