Coverage for src/henrri_connect/connect/synchro.py: 89%

1""" 

2Client HTTP Henrri synchrone. 

3 

4Constants: 

5- ``henrri_connect.connect.synchro._BASE_URL`` 

6 URL de base de l'API Henrri. 

7- ``henrri_connect.connect.synchro.APP_VERSION`` 

8 Version de l'application. 

9- ``henrri_connect.connect.synchro.APP_X_VERSION`` 

10 Version de l'application. 

11 

12Classes: 

13- ``henrri_connect.connect.synchro.SyncHenrriClient`` 

14 Client HTTP Henrri synchrone. 

15 

16Exemples: 

17.. code-block:: python 

18 

19 # Client synchrone 

20 from henrri_connect import SyncHenrriClient 

21 

22 client = SyncHenrriClient("client_id", "client_secret", base_url=<URL de base de l'API>) 

23 

24Exceptions: 

25- ``henrri_connect.exc.HenrriAuthError`` 

26 Erreur d'authentification (HTTP 401). 

27""" 

28 

29from __future__ import annotations 

30 

31import logging 

32from typing import Any, TYPE_CHECKING # pylint: disable=W0611 # type: ignore[import] 

33import httpx 

34from ..exc import HenrriAuthError 

35from ..models import TokenResponse 

36from ..utils import raise_for_status 

37if TYPE_CHECKING: 

38 from ..companies import SyncCompaniesClient 

39 from ..customers import SyncCustomersClient 

40 from ..document_line_types import SyncDocumentLineTypesClient 

41 from ..document_lines import SyncDocumentLinesClient 

42 from ..document_types import SyncDocumentTypesClient 

43 from ..documents import SyncDocumentsClient 

44 from ..item_categories import SyncItemCategoriesClient 

45 from ..items import SyncItemsClient 

46 from ..revenues import SyncRevenuesClient 

47 from ..secures import SyncSecuresClient 

48 from ..units import SyncUnitsClient 

49 from ..users import SyncUsersClient 

50 

51logger = logging.getLogger(__name__) 

52 

53_BASE_URL = "https://api-sandbox.henrri.io" 

54APP_VERSION = "application/json" 

55APP_X_VERSION = "1.0" 

56 

57 

58class SyncHenrriClient: 

59 """ 

60 Client HTTP synchrone pour l'API Henrri. 

61 Gère l'authentification, le refresh-token et les erreurs HTTP. 

62 Fournit des sous-clients pour chaque groupe d'endpoints (users, companies, etc.) 

63 Authentification automatique avec gestion du refresh token 

64 Gestion centralisée des erreurs HTTP avec exceptions personnalisées 

65 

66 Arguments 

67 - client_id: Identifiant client pour l'authentification. 

68 - client_secret: Secret client pour l'authentification. 

69 - base_url: URL de base de l'API (défaut : sandbox Henrri). 

70 

71 Methodes 

72 - authenticate: Authentifie le client. 

73 - request: Effectue une requête HTTP authentifiée. 

74 - close: Ferme le client HTTP sous-jacent. 

75 """ 

76 

77 users: "SyncUsersClient" 

78 companies: "SyncCompaniesClient" 

79 customers: "SyncCustomersClient" 

80 documents: "SyncDocumentsClient" 

81 document_lines: "SyncDocumentLinesClient" 

82 document_line_types: "SyncDocumentLineTypesClient" 

83 document_types: "SyncDocumentTypesClient" 

84 items: "SyncItemsClient" 

85 item_categories: "SyncItemCategoriesClient" 

86 units: "SyncUnitsClient" 

87 revenues: "SyncRevenuesClient" 

88 secures: "SyncSecuresClient" 

89 

90 def __init__( 

91 self, 

92 client_id: str, 

93 client_secret: str, 

94 *, 

95 base_url: str = _BASE_URL, 

96 ) -> None: 

97 self._client_id = client_id 

98 self._client_secret = client_secret 

99 self._base_url = base_url.rstrip("/") 

100 self._access_token: str | None = None 

101 self._refresh_token_str: str | None = None 

102 self._http = httpx.Client() 

103 self._init_subclients() 

104 

105 def _init_subclients(self) -> None: 

106 from ..companies import SyncCompaniesClient # pylint: disable=C0415 

107 from ..customers import SyncCustomersClient # pylint: disable=C0415 

108 from ..document_line_types import SyncDocumentLineTypesClient # pylint: disable=C0415 

109 from ..document_lines import SyncDocumentLinesClient # pylint: disable=C0415 

110 from ..document_types import SyncDocumentTypesClient # pylint: disable=C0415 

111 from ..documents import SyncDocumentsClient # pylint: disable=C0415 

112 from ..item_categories import SyncItemCategoriesClient # pylint: disable=C0415 

113 from ..items import SyncItemsClient # pylint: disable=C0415 

114 from ..revenues import SyncRevenuesClient # pylint: disable=C0415 

115 from ..secures import SyncSecuresClient # pylint: disable=C0415 

116 from ..units import SyncUnitsClient # pylint: disable=C0415 

117 from ..users import SyncUsersClient # pylint: disable=C0415 

118 

119 self.users = SyncUsersClient(self) 

120 self.companies = SyncCompaniesClient(self) 

121 self.customers = SyncCustomersClient(self) 

122 self.documents = SyncDocumentsClient(self) 

123 self.document_lines = SyncDocumentLinesClient(self) 

124 self.document_line_types = SyncDocumentLineTypesClient(self) 

125 self.document_types = SyncDocumentTypesClient(self) 

126 self.items = SyncItemsClient(self) 

127 self.item_categories = SyncItemCategoriesClient(self) 

128 self.units = SyncUnitsClient(self) 

129 self.revenues = SyncRevenuesClient(self) 

130 self.secures = SyncSecuresClient(self) 

131 

132 def _url(self, path: str) -> str: 

133 return f"{self._base_url}{path}" 

134 

135 def _headers(self, *, authenticated: bool = True) -> dict[str, str]: 

136 headers: dict[str, str] = { 

137 "Content-Type": APP_VERSION, 

138 "Accept": APP_VERSION, 

139 "X-Version": APP_X_VERSION, 

140 } 

141 logger.debug("Construction des headers (authentifié=%s)", authenticated) 

142 if authenticated: 

143 if not self._access_token: 

144 logger.error("Tentative de requête authentifiée sans token d'accès.") 

145 raise HenrriAuthError( 

146 401, 

147 "Non authentifié. Appelez authenticate() avant toute requête." 

148 ) 

149 headers["Authorization"] = f"Bearer {self._access_token}" 

150 logger.debug("Headers avec token d'accès ajouté.") 

151 return headers 

152 

153 def authenticate(self) -> TokenResponse: 

154 """ 

155 Authentifie le client et stocke le token d'accès. 

156 

157 Arguments 

158 - None 

159 

160 Returns 

161 - TokenResponse : objet contenant access_token et refresh_token. 

162 """ 

163 resp: httpx.Response = self._http.post( 

164 self._url("/v1/users/authenticate"), 

165 headers=self._headers(authenticated=False), 

166 json={"clientId": self._client_id, "clientSecret": self._client_secret}, 

167 ) 

168 raise_for_status(resp) 

169 token: TokenResponse = TokenResponse.model_validate(resp.json()) 

170 self._access_token = token.access_token 

171 self._refresh_token_str = token.refresh_token 

172 logger.debug("Authentification réussie, token d'accès obtenu.") 

173 logger.info( 

174 "Authentification réussie. Access token valide pour %d secondes.", 

175 token.expires_in 

176 ) 

177 return token 

178 

179 def _do_refresh(self) -> None: 

180 """ 

181 Rafraîchit le token via le refresh token, ou ré-authentifie en cas d'échec. 

182 

183 Arguments 

184 - None 

185 

186 Returns 

187 - None 

188 """ 

189 resp: httpx.Response = self._http.post( 

190 self._url("/v1/users/refresh-token"), 

191 headers=self._headers(authenticated=False), 

192 json={"refreshToken": self._refresh_token_str}, 

193 ) 

194 if resp.is_success: 

195 logger.debug("Rafraîchissement du token via le refresh token.") 

196 token: TokenResponse = TokenResponse.model_validate(resp.json()) 

197 self._access_token = token.access_token 

198 if token.refresh_token: 

199 logger.debug("Nouveau refresh token reçu, mise à jour du refresh token stocké.") 

200 self._refresh_token_str = token.refresh_token 

201 else: 

202 self.authenticate() 

203 

204 def _request( 

205 self, 

206 method: str, 

207 path: str, 

208 *, 

209 authenticated: bool = True, 

210 **kwargs: Any, 

211 ) -> httpx.Response: 

212 """ 

213 Effectue une requête HTTP avec gestion automatique de l'authentification. 

214 

215 Arguments 

216 - method (str) : Le type de la requête (GET, POST, etc.). 

217 - path (str) : L'URL de la requête. 

218 - authenticated (bool, optional) : Indique si la requête doit быть authentifiée. 

219 - kwargs (dict, optional) : Dictionnaire de paramètres supplémentaires pour la requête. 

220 

221 Returns 

222 - httpx.Response : La réponse de la requête. 

223 

224 Raises 

225 - HenrriAuthError : Erreur d'authentification. 

226 """ 

227 if authenticated and not self._access_token: 

228 self.authenticate() 

229 

230 logger.debug( 

231 "Envoi de la requête %s %s (authentifié=%s)", 

232 method, 

233 path, 

234 authenticated, 

235 ) 

236 resp: httpx.Response = self._http.request( 

237 method, 

238 self._url(path), 

239 headers=self._headers(authenticated=authenticated), 

240 **kwargs, 

241 ) 

242 

243 if resp.status_code == 401 and authenticated: 

244 logger.debug("Réponse 401 reçue, tentative de rafraîchissement du token.") 

245 if self._refresh_token_str: 

246 self._do_refresh() 

247 else: 

248 self.authenticate() 

249 resp: httpx.Response = self._http.request( 

250 method, 

251 self._url(path), 

252 headers=self._headers(authenticated=True), 

253 **kwargs, 

254 ) 

255 

256 raise_for_status(resp) 

257 return resp 

258 

259 def request(self, method: str, endpoint: str, **kwargs: Any) -> httpx.Response: 

260 """ 

261 Effectue une requête HTTP authentifiée (synchrone). 

262 

263 Arguments 

264 - method: Méthode HTTP (GET, POST, etc.). 

265 - endpoint: Chemin relatif (ex: '/v1/companies/42'). 

266 - kwargs: Arguments supplémentaires pour httpx.request(). 

267 

268 Returns 

269 - httpx.Response: La réponse HTTP. 

270 """ 

271 return self._request(method, endpoint, **kwargs) 

272 

273 def close(self) -> None: 

274 """ 

275 Ferme le client HTTP sous-jacent. 

276 

277 Arguments 

278 - None 

279 

280 Returns 

281 - None 

282 """ 

283 logger.info("Fermeture du client HTTP.") 

284 self._http.close() 

285 

286 def __enter__(self) -> SyncHenrriClient: 

287 return self 

288 

289 def __exit__(self, *args: Any) -> None: 

290 self.close()