Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/connexion/middleware/routing.py: 47%

1import pathlib 

2import typing as t 

3from contextvars import ContextVar 

4 

5import starlette.convertors 

6from starlette.routing import Router 

7from starlette.types import ASGIApp, Receive, Scope, Send 

8 

9from connexion.frameworks import starlette as starlette_utils 

10from connexion.middleware.abstract import ( 

11 ROUTING_CONTEXT, 

12 AbstractRoutingAPI, 

13 SpecMiddleware, 

14) 

15from connexion.operations import AbstractOperation 

16from connexion.resolver import Resolver 

17 

18_scope: ContextVar[dict] = ContextVar("SCOPE") 

19 

20 

21class RoutingOperation: 

22 def __init__(self, operation_id: t.Optional[str], next_app: ASGIApp) -> None: 

23 self.operation_id = operation_id 

24 self.next_app = next_app 

25 

26 @classmethod 

27 def from_operation(cls, operation: AbstractOperation, next_app: ASGIApp): 

28 return cls(operation.operation_id, next_app) 

29 

30 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

31 """Attach operation to scope and pass it to the next app""" 

32 original_scope = _scope.get() 

33 # Pass resolved path params along 

34 original_scope.setdefault("path_params", {}).update( 

35 scope.get("path_params", {}) 

36 ) 

37 

38 api_base_path = scope.get("root_path", "")[ 

39 len(original_scope.get("root_path", "")) : 

40 ] 

41 

42 extensions = original_scope.setdefault("extensions", {}) 

43 connexion_routing = extensions.setdefault(ROUTING_CONTEXT, {}) 

44 connexion_routing.update( 

45 {"api_base_path": api_base_path, "operation_id": self.operation_id} 

46 ) 

47 await self.next_app(original_scope, receive, send) 

48 

49 

50class RoutingAPI(AbstractRoutingAPI): 

51 def __init__( 

52 self, 

53 specification: t.Union[pathlib.Path, str, dict], 

54 *, 

55 next_app: ASGIApp, 

56 base_path: t.Optional[str] = None, 

57 arguments: t.Optional[dict] = None, 

58 resolver: t.Optional[Resolver] = None, 

59 resolver_error_handler: t.Optional[t.Callable] = None, 

60 debug: bool = False, 

61 **kwargs, 

62 ) -> None: 

63 """API implementation on top of Starlette Router for Connexion middleware.""" 

64 self.next_app = next_app 

65 self.router = Router(default=RoutingOperation(None, next_app)) 

66 

67 super().__init__( 

68 specification, 

69 base_path=base_path, 

70 arguments=arguments, 

71 resolver=resolver, 

72 resolver_error_handler=resolver_error_handler, 

73 debug=debug, 

74 **kwargs, 

75 ) 

76 

77 def make_operation(self, operation: AbstractOperation) -> RoutingOperation: 

78 return RoutingOperation.from_operation(operation, next_app=self.next_app) 

79 

80 @staticmethod 

81 def _framework_path_and_name( 

82 operation: AbstractOperation, path: str 

83 ) -> t.Tuple[str, str]: 

84 types = operation.get_path_parameter_types() 

85 starlette_path = starlette_utils.starlettify_path(path, types) 

86 return starlette_path, starlette_path 

87 

88 def _add_operation_internal( 

89 self, method: str, path: str, operation: RoutingOperation, name: str = None 

90 ) -> None: 

91 self.router.add_route(path, operation, methods=[method]) 

92 

93 

94class RoutingMiddleware(SpecMiddleware): 

95 def __init__(self, app: ASGIApp, **kwargs) -> None: 

96 """Middleware that resolves the Operation for an incoming request and attaches it to the 

97 scope. 

98 

99 :param app: app to wrap in middleware. 

100 """ 

101 self.app = app 

102 # Pass unknown routes to next app 

103 self.router = Router(default=RoutingOperation(None, self.app)) 

104 starlette.convertors.register_url_convertor( 

105 "float", starlette_utils.FloatConverter() 

106 ) 

107 starlette.convertors.register_url_convertor( 

108 "int", starlette_utils.IntegerConverter() 

109 ) 

110 

111 def add_api( 

112 self, 

113 specification: t.Union[pathlib.Path, str, dict], 

114 base_path: t.Optional[str] = None, 

115 arguments: t.Optional[dict] = None, 

116 **kwargs, 

117 ) -> None: 

118 """Add an API to the router based on a OpenAPI spec. 

119 

120 :param specification: OpenAPI spec as dict or path to file. 

121 :param base_path: Base path where to add this API. 

122 :param arguments: Jinja arguments to replace in the spec. 

123 """ 

124 api = RoutingAPI( 

125 specification, 

126 base_path=base_path, 

127 arguments=arguments, 

128 next_app=self.app, 

129 **kwargs, 

130 ) 

131 self.router.mount(api.base_path, app=api.router) 

132 

133 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

134 """Route request to matching operation, and attach it to the scope before calling the 

135 next app.""" 

136 if scope["type"] != "http": 

137 await self.app(scope, receive, send) 

138 return 

139 

140 _scope.set(scope.copy()) # type: ignore 

141 

142 await self.router(scope, receive, send)