Coverage for src\baobab_web_api_caller\core\baobab_request.py: 88%

1"""Modèle de requête HTTP.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from types import MappingProxyType 

7from typing import Mapping, Sequence 

8from urllib.parse import quote 

9 

10from baobab_web_api_caller.core.http_method import HttpMethod 

11from baobab_web_api_caller.exceptions.configuration_exception import ConfigurationException 

12from baobab_web_api_caller.utils.mapping_utils import freeze_str_mapping 

13 

14 

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

16class BaobabRequest: 

17 """Représentation typée d'une requête HTTP. 

18 

19 La requête est volontairement indépendante du transport concret (requests, httpx, ...). 

20 

21 :param method: Méthode HTTP. 

22 :type method: HttpMethod 

23 :param path: Chemin relatif (ex: ``/v1/items``). 

24 :type path: str 

25 :param query_params: Paramètres de query string. 

26 Les valeurs peuvent être soit une chaîne (clé simple), soit une séquence de chaînes pour 

27 représenter des clés répétées. 

28 :type query_params: Mapping[str, str | Sequence[str]] 

29 :param headers: En-têtes HTTP. 

30 :type headers: Mapping[str, str] 

31 :param json_body: Corps JSON (déjà sérialisé en types Python). 

32 :type json_body: object | None 

33 :param form_body: Corps de formulaire (application/x-www-form-urlencoded). 

34 :type form_body: Mapping[str, str] | None 

35 :param timeout_seconds: Timeout en secondes. 

36 :type timeout_seconds: float | None 

37 :raises ConfigurationException: Si les paramètres de la requête sont invalides. 

38 """ 

39 

40 method: HttpMethod 

41 path: str 

42 query_params: Mapping[str, str | Sequence[str]] 

43 headers: Mapping[str, str] 

44 json_body: object | None = None 

45 form_body: Mapping[str, str] | None = None 

46 timeout_seconds: float | None = None 

47 

48 def __post_init__(self) -> None: 

49 if not isinstance(self.method, HttpMethod): 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 raise ConfigurationException("method must be an HttpMethod") 

51 

52 normalized_path = self._normalize_and_validate_path(self.path) 

53 object.__setattr__(self, "path", normalized_path) 

54 

55 if self.timeout_seconds is not None and self.timeout_seconds <= 0: 

56 raise ConfigurationException("timeout_seconds must be positive when provided") 

57 

58 if self.json_body is not None and self.form_body is not None: 

59 raise ConfigurationException("json_body and form_body are mutually exclusive") 

60 

61 object.__setattr__(self, "query_params", self._freeze_query_params(self.query_params)) 

62 object.__setattr__(self, "headers", freeze_str_mapping(self.headers, "headers")) 

63 if self.form_body is not None: 

64 object.__setattr__(self, "form_body", freeze_str_mapping(self.form_body, "form_body")) 

65 

66 @staticmethod 

67 def _freeze_query_params( 

68 value: Mapping[str, str | Sequence[str]], 

69 ) -> Mapping[str, str | Sequence[str]]: 

70 if not isinstance(value, Mapping): 70 ↛ 71line 70 didn't jump to line 71 because the condition on line 70 was never true

71 raise ConfigurationException("query_params must be a mapping") 

72 

73 frozen: dict[str, str | tuple[str, ...]] = {} 

74 for k, v in value.items(): 

75 if not isinstance(k, str) or k.strip() == "": 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 raise ConfigurationException("query_params keys must be non-empty strings") 

77 

78 if isinstance(v, str): 

79 frozen[k] = v 

80 else: 

81 if not isinstance(v, Sequence): 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true

82 raise ConfigurationException( 

83 "query_params values must be strings or sequences of strings" 

84 ) 

85 collected: list[str] = [] 

86 for item in v: 

87 if not isinstance(item, str): 87 ↛ 88line 87 didn't jump to line 88 because the condition on line 87 was never true

88 raise ConfigurationException("query_params sequence values must be strings") 

89 collected.append(item) 

90 frozen[k] = tuple(collected) 

91 

92 return MappingProxyType(frozen) 

93 

94 @staticmethod 

95 def _normalize_and_validate_path(path: str) -> str: 

96 if not isinstance(path, str) or path.strip() == "": 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true

97 raise ConfigurationException("path must be a non-empty string") 

98 

99 stripped = path.strip() 

100 lowered = stripped.lower() 

101 if lowered.startswith("http://") or lowered.startswith("https://"): 

102 raise ConfigurationException("path must be a relative path, not an absolute URL") 

103 

104 if not stripped.startswith("/"): 

105 stripped = f"/{stripped}" 

106 

107 if " " in stripped: 

108 raise ConfigurationException("path must not contain spaces") 

109 

110 # Normalisation minimale: encode les caractères non sûrs sans toucher au '/'. 

111 return quote(stripped, safe="/-._~") 

112 

113 def with_header(self, name: str, value: str) -> "BaobabRequest": 

114 """Retourne une nouvelle requête avec un header ajouté/écrasé. 

115 

116 :param name: Nom du header. 

117 :type name: str 

118 :param value: Valeur du header. 

119 :type value: str 

120 :return: Nouvelle instance. 

121 :rtype: BaobabRequest 

122 """ 

123 

124 headers = dict(self.headers) 

125 headers[name] = value 

126 return BaobabRequest( 

127 method=self.method, 

128 path=self.path, 

129 query_params=self.query_params, 

130 headers=headers, 

131 json_body=self.json_body, 

132 form_body=self.form_body, 

133 timeout_seconds=self.timeout_seconds, 

134 )