Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/connexion/middleware/exceptions.py: 58%

1import asyncio 

2import functools 

3import inspect 

4import logging 

5import typing as t 

6 

7from starlette.concurrency import run_in_threadpool 

8from starlette.exceptions import HTTPException 

9from starlette.middleware.exceptions import ( 

10 ExceptionMiddleware as StarletteExceptionMiddleware, 

11) 

12from starlette.requests import Request as StarletteRequest 

13from starlette.responses import Response as StarletteResponse 

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

15 

16from connexion import http_facts 

17from connexion.exceptions import InternalServerError, ProblemException, problem 

18from connexion.lifecycle import ConnexionRequest, ConnexionResponse 

19from connexion.types import MaybeAwaitable 

20 

21logger = logging.getLogger(__name__) 

22 

23 

24def connexion_wrapper( 

25 handler: t.Callable[ 

26 [ConnexionRequest, Exception], MaybeAwaitable[ConnexionResponse] 

27 ], 

28) -> t.Callable[[StarletteRequest, Exception], t.Awaitable[StarletteResponse]]: 

29 """Wrapper that translates Starlette requests to Connexion requests before passing 

30 them to the error handler, and translates the returned Connexion responses to 

31 Starlette responses.""" 

32 

33 @functools.wraps(handler) 

34 async def wrapper(request: StarletteRequest, exc: Exception) -> StarletteResponse: 

35 request = ConnexionRequest.from_starlette_request(request) 

36 

37 if inspect.iscoroutinefunction(handler): 

38 response = await handler(request, exc) # type: ignore 

39 else: 

40 response = await run_in_threadpool(handler, request, exc) 

41 

42 while asyncio.iscoroutine(response): 

43 response = await response 

44 

45 return StarletteResponse( 

46 content=response.body, 

47 status_code=response.status_code, 

48 media_type=response.mimetype, 

49 headers=response.headers, 

50 ) 

51 

52 return wrapper 

53 

54 

55class ExceptionMiddleware(StarletteExceptionMiddleware): 

56 """Subclass of starlette ExceptionMiddleware to change handling of HTTP exceptions to 

57 existing connexion behavior.""" 

58 

59 def __init__(self, next_app: ASGIApp): 

60 super().__init__(next_app) 

61 self.add_exception_handler(ProblemException, self.problem_handler) # type: ignore 

62 self.add_exception_handler(Exception, self.common_error_handler) 

63 

64 def add_exception_handler( 

65 self, 

66 exc_class_or_status_code: t.Union[int, t.Type[Exception]], 

67 handler: t.Callable[[ConnexionRequest, Exception], StarletteResponse], 

68 ) -> None: 

69 super().add_exception_handler( 

70 exc_class_or_status_code, handler=connexion_wrapper(handler) 

71 ) 

72 

73 @staticmethod 

74 def problem_handler(_request: ConnexionRequest, exc: ProblemException): 

75 """Default handler for Connexion ProblemExceptions""" 

76 

77 if 400 <= exc.status <= 499: 

78 logger.warning("%r", exc) 

79 else: 

80 logger.error("%r", exc) 

81 

82 return exc.to_problem() 

83 

84 @staticmethod 

85 @connexion_wrapper 

86 def http_exception( 

87 _request: StarletteRequest, exc: HTTPException, **kwargs 

88 ) -> StarletteResponse: 

89 """Default handler for Starlette HTTPException""" 

90 

91 if 400 <= exc.status_code <= 499: 

92 logger.warning("%r", exc) 

93 else: 

94 logger.error("%r", exc) 

95 

96 return problem( 

97 title=http_facts.HTTP_STATUS_CODES.get(exc.status_code), 

98 detail=exc.detail, 

99 status=exc.status_code, 

100 headers=exc.headers, 

101 ) 

102 

103 @staticmethod 

104 def common_error_handler( 

105 _request: StarletteRequest, exc: Exception 

106 ) -> ConnexionResponse: 

107 """Default handler for any unhandled Exception""" 

108 logger.error("%r", exc, exc_info=exc) 

109 return InternalServerError().to_problem() 

110 

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

112 await super().__call__(scope, receive, send)