Coverage for src/henrri_connect/connect/asynchro.py: 80%

1""" 

2Client HTTP Henrri : implémentations asynchrone. 

3 

4Constants: 

5- ``henrri_connect.connect.asynchro._BASE_URL`` 

6 URL de base de l'API Henrri. 

7- ``henrri_connect.connect.asynchro.APP_VERSION`` 

8 Version de l'application. 

9- ``henrri_connect.connect.asynchro.APP_X_VERSION`` 

10 Version de l'application. 

11 

12Classes: 

13- ``henrri_connect.connect.asynchro.AsyncHenrriClient`` 

14 Client HTTP Henrri asynchrone. 

15 

16Exceptions: 

17- ``henrri_connect.exc.HenrriAuthError`` 

18 Erreur d'authentification (HTTP 401). 

19 

20Exemples: 

21.. code-block:: python 

22 

23 # Client asynchrone 

24 from henrri_connect import AsyncHenrriClient 

25 

26 client = AsyncHenrriClient("client_id", "client_secret", base_url="<production_url>") 

27""" 

28 

29from __future__ import annotations 

30 

31import logging 

32from typing import Any, TYPE_CHECKING 

33import httpx 

34from ..exc import HenrriAuthError 

35from ..models import TokenResponse 

36from ..utils import raise_for_status 

37if TYPE_CHECKING: 

38 from ..companies import AsyncCompaniesClient 

39 from ..customers import AsyncCustomersClient 

40 from ..document_line_types import AsyncDocumentLineTypesClient 

41 from ..document_lines import AsyncDocumentLinesClient 

42 from ..document_types import AsyncDocumentTypesClient 

43 from ..documents import AsyncDocumentsClient 

44 from ..item_categories import AsyncItemCategoriesClient 

45 from ..items import AsyncItemsClient 

46 from ..revenues import AsyncRevenuesClient 

47 from ..secures import AsyncSecuresClient 

48 from ..units import AsyncUnitsClient 

49 from ..users import AsyncUsersClient 

50 

51logger = logging.getLogger(__name__) 

52 

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

54APP_VERSION = "application/json" 

55APP_X_VERSION = "1.0" 

56 

57class AsyncHenrriClient: 

58 """ 

59 Client HTTP asynchrone pour l'API Henrri. 

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

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

62 Authentification automatique avec gestion du refresh token 

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

64 

65 Arguments 

66 - client_id: Identifiant client pour l'authentification. 

67 - client_secret: Secret client pour l'authentification. 

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

69 

70 Methodes 

71 - authenticate: Authentifie le client. 

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

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

74 """ 

75 

76 users: "AsyncUsersClient" 

77 companies: "AsyncCompaniesClient" 

78 customers: "AsyncCustomersClient" 

79 documents: "AsyncDocumentsClient" 

80 document_lines: "AsyncDocumentLinesClient" 

81 document_line_types: "AsyncDocumentLineTypesClient" 

82 document_types: "AsyncDocumentTypesClient" 

83 items: "AsyncItemsClient" 

84 item_categories: "AsyncItemCategoriesClient" 

85 units: "AsyncUnitsClient" 

86 revenues: "AsyncRevenuesClient" 

87 secures: "AsyncSecuresClient" 

88 

89 def __init__( 

90 self, 

91 client_id: str, 

92 client_secret: str, 

93 *, 

94 base_url: str = _BASE_URL, 

95 ) -> None: 

96 self._client_id = client_id 

97 self._client_secret = client_secret 

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

99 self._access_token: str | None = None 

100 self._refresh_token_str: str | None = None 

101 self._http = httpx.AsyncClient(timeout=30.0) 

102 self._init_subclients() 

103 

104 def _init_subclients(self) -> None: 

105 from ..companies import AsyncCompaniesClient # pylint: disable=C0415 

106 from ..customers import AsyncCustomersClient # pylint: disable=C0415 

107 from ..document_line_types import AsyncDocumentLineTypesClient # pylint: disable=C0415 

108 from ..document_lines import AsyncDocumentLinesClient # pylint: disable=C0415 

109 from ..document_types import AsyncDocumentTypesClient # pylint: disable=C0415 

110 from ..documents import AsyncDocumentsClient # pylint: disable=C0415 

111 from ..item_categories import AsyncItemCategoriesClient # pylint: disable=C0415 

112 from ..items import AsyncItemsClient # pylint: disable=C0415 

113 from ..revenues import AsyncRevenuesClient # pylint: disable=C0415 

114 from ..secures import AsyncSecuresClient # pylint: disable=C0415 

115 from ..units import AsyncUnitsClient # pylint: disable=C0415 

116 from ..users import AsyncUsersClient # pylint: disable=C0415 

117 

118 self.users = AsyncUsersClient(self) 

119 self.companies = AsyncCompaniesClient(self) 

120 self.customers = AsyncCustomersClient(self) 

121 self.documents = AsyncDocumentsClient(self) 

122 self.document_lines = AsyncDocumentLinesClient(self) 

123 self.document_line_types = AsyncDocumentLineTypesClient(self) 

124 self.document_types = AsyncDocumentTypesClient(self) 

125 self.items = AsyncItemsClient(self) 

126 self.item_categories = AsyncItemCategoriesClient(self) 

127 self.units = AsyncUnitsClient(self) 

128 self.revenues = AsyncRevenuesClient(self) 

129 self.secures = AsyncSecuresClient(self) 

130 

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

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

133 

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

135 headers: dict[str, str] = { 

136 "Content-Type": APP_VERSION, 

137 "Accept": APP_VERSION, 

138 "X-Version": APP_X_VERSION, 

139 } 

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

141 if authenticated: 

142 if not self._access_token: 

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

144 raise HenrriAuthError( 

145 401, 

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

147 ) 

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

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

150 return headers 

151 

152 async def authenticate(self) -> TokenResponse: 

153 """ 

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

155 

156 Arguments 

157 - None 

158 

159 Returns 

160 - TokenResponse : objet contenant access_token et refresh_token. 

161 """ 

162 resp: httpx.Response = await self._http.post( 

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

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

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

166 ) 

167 raise_for_status(resp) 

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

169 self._access_token = token.access_token 

170 self._refresh_token_str = token.refresh_token 

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

172 logger.info( 

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

174 token.expires_in 

175 ) 

176 return token 

177 

178 async def _do_refresh(self) -> None: 

179 """ 

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

181 

182 Arguments 

183 - None 

184 

185 Returns 

186 - None 

187 """ 

188 resp: httpx.Response = await self._http.post( 

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

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

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

192 ) 

193 if resp.is_success: 

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

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

196 self._access_token = token.access_token 

197 if token.refresh_token: 

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

199 self._refresh_token_str = token.refresh_token 

200 else: 

201 await self.authenticate() 

202 

203 async def _request( 

204 self, 

205 method: str, 

206 path: str, 

207 *, 

208 authenticated: bool = True, 

209 **kwargs: Any, 

210 ) -> httpx.Response: 

211 """ 

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

213 

214 Arguments 

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

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

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

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

219 

220 Returns 

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

222 

223 Raises 

224 - HenrriAuthError : Erreur d'authentification. 

225 """ 

226 if authenticated and not self._access_token: 

227 await self.authenticate() 

228 

229 logger.debug( 

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

231 method, 

232 path, 

233 authenticated, 

234 ) 

235 resp: httpx.Response = await self._http.request( 

236 method, 

237 self._url(path), 

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

239 **kwargs, 

240 ) 

241 

242 if resp.status_code == 401 and authenticated: 

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

244 if self._refresh_token_str: 

245 await self._do_refresh() 

246 else: 

247 await self.authenticate() 

248 resp: httpx.Response = await self._http.request( 

249 method, 

250 self._url(path), 

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

252 **kwargs, 

253 ) 

254 

255 raise_for_status(resp) 

256 return resp 

257 

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

259 """ 

260 Effectue une requête HTTP authentifiée (asynchrone). 

261 

262 Arguments 

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

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

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

266 

267 Returns 

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

269 """ 

270 return await self._request(method, endpoint, **kwargs) 

271 

272 async def close(self) -> None: 

273 """ 

274 Ferme le client HTTP sous-jacent. 

275 

276 Arguments 

277 - None 

278 

279 Returns 

280 - None 

281 """ 

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

283 await self._http.aclose() 

284 

285 async def __aenter__(self) -> AsyncHenrriClient: 

286 return self 

287 

288 async def __aexit__(self, *args: Any) -> None: 

289 await self.close()