Coverage for odmpy/libby_errors.py: 93.5%

31 statements  

« prev     ^ index     » next       coverage.py v7.3.1, created at 2023-09-14 08:51 +0000

1# Copyright (C) 2023 github.com/ping 

2# 

3# This file is part of odmpy. 

4# 

5# odmpy is free software: you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation, either version 3 of the License, or 

8# (at your option) any later version. 

9# 

10# odmpy is distributed in the hope that it will be useful, 

11# but WITHOUT ANY WARRANTY; without even the implied warranty of 

12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

13# GNU General Public License for more details. 

14# 

15# You should have received a copy of the GNU General Public License 

16# along with odmpy. If not, see <http://www.gnu.org/licenses/>. 

17# 

18 

19import json 

20from http import HTTPStatus 

21 

22import requests as requests 

23 

24# 

25# For use with LibbyClient 

26# 

27 

28 

29class ClientError(Exception): 

30 """Generic error class, catch-all for most client issues.""" 

31 

32 def __init__( 

33 self, 

34 msg: str, 

35 http_status: int = 0, 

36 error_response: str = "", 

37 ): 

38 self.http_status = http_status or 0 

39 self.error_response = error_response 

40 try: 

41 self.error_response_obj = json.loads(self.error_response) 

42 except ValueError: 

43 self.error_response_obj = {} 

44 super(ClientError, self).__init__(msg) 

45 

46 @property 

47 def msg(self): 

48 return self.args[0] 

49 

50 def __str__(self): 

51 return ( 

52 f"<{type(self).__module__}.{type(self).__name__}; http_status={self.http_status}, " 

53 f"msg='{self.msg}', error_response='{self.error_response}''>" 

54 ) 

55 

56 

57class ClientConnectionError(ClientError): 

58 """Connection error""" 

59 

60 

61class ClientTimeoutError(ClientError): 

62 """Timeout error""" 

63 

64 

65class ClientBadRequestError(ClientError): 

66 """Raised when an HTTP 400 response is received.""" 

67 

68 

69class ErrorHandler(object): 

70 @staticmethod 

71 def process(http_err: requests.HTTPError) -> None: 

72 """ 

73 Try to process an HTTP error from the api appropriately. 

74 

75 :param http_err: requests.HTTPError instance 

76 :raises ClientError: 

77 :return: 

78 """ 

79 # json response 

80 if ( 

81 http_err.response.status_code == HTTPStatus.BAD_REQUEST 

82 and http_err.response.headers.get("content-type", "").startswith( 

83 "application/json" 

84 ) 

85 ): 

86 error = http_err.response.json() 

87 if error.get("result", "") == "upstream_failure": 

88 upstream = error.get("upstream", {}) 

89 if upstream: 

90 raise ClientBadRequestError( 

91 msg=f'{upstream.get("userExplanation", "")} [errorcode: {upstream.get("errorCode", "")}]', 

92 http_status=http_err.response.status_code, 

93 error_response=http_err.response.text, 

94 ) from http_err 

95 

96 raise ClientBadRequestError( 

97 msg=str(error), 

98 http_status=http_err.response.status_code, 

99 error_response=http_err.response.text, 

100 ) from http_err 

101 

102 # final fallback 

103 raise ClientError( 

104 msg=str(http_err), 

105 http_status=http_err.response.status_code, 

106 error_response=http_err.response.text, 

107 ) from http_err