Coverage for src\baobab_web_api_caller\core\error_response_mapper.py: 95%

1"""Mapping des erreurs de réponse vers des exceptions projet.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import ClassVar, Mapping 

7 

8from baobab_web_api_caller.core.baobab_response import BaobabResponse 

9from baobab_web_api_caller.exceptions.authentication_exception import AuthenticationException 

10from baobab_web_api_caller.exceptions.client_http_exception import ClientHttpException 

11from baobab_web_api_caller.exceptions.rate_limit_exception import RateLimitException 

12from baobab_web_api_caller.exceptions.resource_not_found_exception import ( 

13 ResourceNotFoundException, 

14) 

15from baobab_web_api_caller.exceptions.server_http_exception import ServerHttpException 

16 

17 

18@dataclass(frozen=True, slots=True) 

19class ErrorResponseMapper: 

20 """Transforme une réponse en exception projet lorsque nécessaire.""" 

21 

22 _STATUS_REASONS: ClassVar[dict[int, str]] = { 

23 400: "Bad Request", 

24 401: "Unauthorized", 

25 403: "Forbidden", 

26 404: "Not Found", 

27 429: "Too Many Requests", 

28 500: "Internal Server Error", 

29 502: "Bad Gateway", 

30 503: "Service Unavailable", 

31 504: "Gateway Timeout", 

32 } 

33 

34 def raise_for_error(self, response: BaobabResponse) -> None: 

35 """Lève une exception projet si le status code indique une erreur. 

36 

37 Le message et les attributs de l'exception exposent un sous-ensemble des informations 

38 de la réponse (status, extrait de body texte, quelques en-têtes utiles) pour faciliter 

39 le diagnostic tout en évitant de logguer des payloads trop volumineux. 

40 

41 Le champ `message` est de la forme `HTTP {status_code} {raison}` lorsqu'une raison 

42 standard est connue, et sinon `HTTP {status_code} Client Error` / `Server Error`. 

43 """ 

44 

45 status = response.status_code 

46 if status < 400: 

47 return 

48 

49 body_excerpt = self._extract_body_excerpt(response.text) 

50 headers_subset = self._extract_diagnostic_headers(response.headers) 

51 message = self._build_error_message(status) 

52 

53 if status == 401: 

54 raise AuthenticationException( 

55 status_code=status, 

56 message=message, 

57 body_excerpt=body_excerpt, 

58 headers=headers_subset, 

59 ) 

60 if status == 404: 

61 raise ResourceNotFoundException( 

62 status_code=status, 

63 message=message, 

64 body_excerpt=body_excerpt, 

65 headers=headers_subset, 

66 ) 

67 if status == 429: 

68 raise RateLimitException( 

69 status_code=status, 

70 message=message, 

71 body_excerpt=body_excerpt, 

72 headers=headers_subset, 

73 ) 

74 if 400 <= status <= 499: 

75 raise ClientHttpException( 

76 status_code=status, 

77 message=message, 

78 body_excerpt=body_excerpt, 

79 headers=headers_subset, 

80 ) 

81 raise ServerHttpException( 

82 status_code=status, 

83 message=message, 

84 body_excerpt=body_excerpt, 

85 headers=headers_subset, 

86 ) 

87 

88 @classmethod 

89 def _build_error_message(cls, status_code: int) -> str: 

90 reason = cls._STATUS_REASONS.get(status_code) 

91 if reason is not None: 

92 return f"HTTP {status_code} {reason}" 

93 if 400 <= status_code <= 499: 93 ↛ 95line 93 didn't jump to line 95 because the condition on line 93 was always true

94 return f"HTTP {status_code} Client Error" 

95 return f"HTTP {status_code} Server Error" 

96 

97 @staticmethod 

98 def _extract_body_excerpt(text: str | None, *, max_length: int = 256) -> str | None: 

99 if text is None: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 return None 

101 

102 stripped = text.strip() 

103 if not stripped: 

104 return None 

105 if len(stripped) <= max_length: 

106 return stripped 

107 return f"{stripped[:max_length]}…" 

108 

109 @staticmethod 

110 def _extract_diagnostic_headers(headers: Mapping[str, str]) -> dict[str, str] | None: 

111 interesting_keys = { 

112 "content-type", 

113 "x-request-id", 

114 "x-correlation-id", 

115 "retry-after", 

116 "www-authenticate", 

117 } 

118 

119 subset: dict[str, str] = {} 

120 for key, value in headers.items(): 

121 lowered = key.lower() 

122 if lowered in interesting_keys: 

123 subset[key] = value 

124 

125 return subset or None