Coverage for src/henrri_connect/users/asynchro.py: 88%

1""" 

2Sous-client pour les endpoints /v1/users. 

3 

4Classes: 

5- ``henrri_connect.users.asynchro.AsyncUsersClient`` : Accès asynchrone aux endpoints utilisateurs. 

6 

7Notes: 

8- Utiliser de préférence l'objet ``henrri_connect.AsyncHenrriClient`` pour acceder aux endpoints. 

9""" 

10 

11from __future__ import annotations 

12 

13from typing import TYPE_CHECKING 

14 

15from ..models import Address, TokenResponse, User, UserAndCompany 

16 

17if TYPE_CHECKING: 

18 from ..connect import AsyncHenrriClient 

19 

20USERS_ENDPOINT = "/v1/users" 

21 

22class AsyncUsersClient: 

23 """ 

24 Accès asynchrone aux endpoints utilisateurs. 

25 

26 Arguments 

27 - client (AsyncHenrriClient) : Client asynchrone pour acceder aux endpoints. 

28 

29 Methods 

30 - get(user_id: int) : Récupère un utilisateur par son identifiant. 

31 - get_address(user_id: int) : Récupère l'adresse d'un utilisateur. 

32 - get_companies() : Récupère les entreprises associées à l'utilisateur courant. 

33 - authenticate() : Authentifie via les identifiants du client. 

34 - refresh_token(refresh_token: str) : Rafraîchit le token d'accès. 

35 """ 

36 

37 def __init__(self, client: AsyncHenrriClient) -> None: 

38 self._c = client 

39 

40 async def get(self, user_id: int) -> User: 

41 """ 

42 Récupère un utilisateur par son identifiant. 

43 

44 Arguments 

45 - ``user_id`` (int) : Identifiant de l'utilisateur. 

46 

47 Returns 

48 - ``User`` : Utilisateur. 

49 """ 

50 resp = await self._c.request("GET", f"{USERS_ENDPOINT}/{user_id}") 

51 return User.model_validate(resp.json()) 

52 

53 async def get_address(self, user_id: int) -> Address: 

54 """ 

55 Récupère l'adresse d'un utilisateur. 

56 

57 Arguments: 

58 - ``user_id`` (int) : Identifiant de l'utilisateur. 

59 

60 Returns: 

61 - ``henrri_connect.models.Address`` : Adresse de l'utilisateur. 

62 """ 

63 resp = await self._c.request("GET", f"{USERS_ENDPOINT}/{user_id}/address") 

64 return Address.model_validate(resp.json()) 

65 

66 async def get_companies(self) -> list[UserAndCompany]: 

67 """ 

68 Récupère les entreprises associées à l'utilisateur courant. 

69 

70 Returns: 

71 - ``list[UserAndCompany]`` : Entreprises. 

72 """ 

73 resp = await self._c.request("GET", f"{USERS_ENDPOINT}/companies") 

74 return [UserAndCompany.model_validate(item) for item in resp.json()] 

75 

76 async def authenticate(self) -> TokenResponse: 

77 """ 

78 Authentifie via les identifiants du client. 

79 

80 Returns: 

81 - ``henrri_connect.models.TokenResponse`` : Objet contenant access_token et refresh_token. 

82 """ 

83 return await self._c.authenticate() 

84 

85 async def refresh_token(self, refresh_token: str) -> TokenResponse: 

86 """ 

87 Rafraîchit le token d'accès. 

88 

89 Arguments: 

90 - ``refresh_token`` (str) : Token de rafraîchissement. 

91 

92 Returns: 

93 - ``henrri_connect.models.TokenResponse`` : Objet contenant access_token et refresh_token. 

94 """ 

95 resp = await self._c.request( 

96 "POST", 

97 f"{USERS_ENDPOINT}/refresh-token", 

98 authenticated=False, 

99 json={"refreshToken": refresh_token}, 

100 ) 

101 token = TokenResponse.model_validate(resp.json()) 

102 self._c._access_token = token.access_token # pylint: disable=W0212 

103 if token.refresh_token: 

104 self._c._refresh_token_str = token.refresh_token # pylint: disable=W0212 

105 return token