Coverage for src/henrri_connect/items/asynchro.py: 61%

1""" 

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

3 

4Classes: 

5- ``henrri_connect.items.asynchro.AsyncItemsClient`` 

6 Accès asynchrone aux endpoints articles. 

7 

8Notes: 

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

10""" 

11 

12from __future__ import annotations 

13 

14from typing import TYPE_CHECKING, overload, Optional 

15 

16from ..models import Item, PagedListResponse, ItemsQuery 

17from ..utils import clean 

18 

19if TYPE_CHECKING: 

20 from ..connect import AsyncHenrriClient 

21 

22ITEMS_ENDPOINT = "/v1/items" 

23 

24 

25class AsyncItemsClient: 

26 """ 

27 Accès asynchrone aux endpoints articles. 

28 

29 Arguments 

30 - ``client`` : Objet ``henrri_connect.AsyncHenrriClient`` pour acceder aux endpoints. 

31 

32 Methods 

33 - ``list_items`` : Liste les articles avec pagination et filtres optionnels. 

34 - ``add`` : Crée un nouvel article. 

35 - ``get`` : Récupère un article par son identifiant. 

36 - ``modify`` : Modifie un article par son identifiant. 

37 - ``delete`` : Supprime un article par son identifiant. 

38 - ``get_most_used`` : Récupère les articles les plus utilisés. 

39 - ``get_best_sales`` : Récupère les articles les plus vendus. 

40 - ``list_with_selected_fields`` : Liste les articles avec sélection de champs. 

41 """ 

42 

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

44 self._c = client 

45 

46 @overload 

47 async def list_items( 

48 self, 

49 *, 

50 request: ItemsQuery, 

51 with_selected_fields: bool = True, 

52 with_totals: bool = False, 

53 only_current_page: bool = True 

54 ) -> PagedListResponse[Item]:... 

55 @overload 

56 async def list_items( 

57 self, 

58 *, 

59 request: ItemsQuery, 

60 with_selected_fields: bool = False, 

61 with_totals: None = None, 

62 only_current_page: None = None, 

63 ) -> PagedListResponse[Item]:... 

64 async def list_items( 

65 self, 

66 *, 

67 request: ItemsQuery, 

68 with_selected_fields: bool = True, 

69 with_totals: Optional[bool] = False, 

70 only_current_page: Optional[bool] = True 

71 ) -> PagedListResponse[Item]: 

72 """ 

73 Liste les articles avec pagination et filtres optionnels. 

74 

75 Arguments 

76 - ``request`` (ItemsQuery) : Paramètres de recherche. 

77 - ``with_selected_fields`` (bool) : Si True, lance une requête de recherche avancée. 

78 - ``with_totals`` (bool) : Si True, renvoie les totaux. 

79 - ``only_current_page`` (bool) : Si True, renvoie uniquement les articles de la page 

80 actuelle. 

81 

82 Returns 

83 - ``PagedListResponse[Item]`` : Liste paginée d'articles. 

84 """ 

85 params = clean(request.model_dump(by_alias=True)) 

86 if with_selected_fields: 

87 if with_totals and only_current_page: 

88 params["with_totals"] = with_totals 

89 params["only_current_page"] = only_current_page 

90 resp = await self._c.request( 

91 "GET", 

92 f"{ITEMS_ENDPOINT}/with-selected-fields", 

93 params=params 

94 ) 

95 else: 

96 raise ValueError( 

97 "with_totals and only_current_page must be True if with_selected_fields is True" 

98 ) 

99 else: 

100 resp = await self._c.request("GET", ITEMS_ENDPOINT, params=params) 

101 return PagedListResponse[Item].model_validate(resp.json()) 

102 

103 

104 async def add(self, item: Item) -> Item: 

105 """ 

106 Crée un nouvel article. 

107 

108 Arguments 

109 - ``item`` (Item) : Article à créer. 

110 

111 Returns 

112 - ``Item`` : Article créé. 

113 """ 

114 resp = await self._c.request( 

115 "POST", 

116 ITEMS_ENDPOINT, 

117 json=item.model_dump(by_alias=True, exclude_unset=True, exclude_none=True), 

118 ) 

119 return Item.model_validate(resp.json()) 

120 

121 async def get(self, item_id: int) -> Item: 

122 """ 

123 Récupère un article par son identifiant. 

124 

125 Arguments 

126 - ``item_id`` (int) : Identifiant de l'article. 

127 

128 Returns 

129 - ``Item`` : Article trouvé. 

130 """ 

131 resp = await self._c.request("GET", f"{ITEMS_ENDPOINT}/{item_id}") 

132 return Item.model_validate(resp.json()) 

133 

134 async def modify(self, item_id: int, item: Item) -> Item: 

135 """ 

136 Met à jour un article existant. 

137 

138 Arguments 

139 - ``item_id`` (int) : Identifiant de l'article. 

140 - ``item`` (Item) : Article à mettre à jour. 

141 

142 Returns 

143 - ``Item`` : Article mis à jour. 

144 """ 

145 resp = await self._c.request( 

146 "PUT", 

147 f"{ITEMS_ENDPOINT}/{item_id}", 

148 json=item.model_dump(by_alias=True, exclude_unset=True, exclude_none=True), 

149 ) 

150 return Item.model_validate(resp.json()) 

151 

152 async def delete(self, item_id: int) -> None: 

153 """ 

154 Supprime un article. 

155 

156 Arguments 

157 - ``item_id`` (int) : Identifiant de l'article à supprimer. 

158 

159 Returns 

160 - ``None`` : Article supprimé. 

161 """ 

162 await self._c.request("DELETE", f"{ITEMS_ENDPOINT}/{item_id}") 

163 

164 async def get_most_used(self) -> PagedListResponse[Item]: 

165 """ 

166 Récupère les articles les plus utilisés. 

167 

168 Arguments 

169 - None 

170 

171 Returns 

172 - ``PagedListResponse[Item]`` : Liste paginée d'articles. 

173 """ 

174 resp = await self._c.request("GET", f"{ITEMS_ENDPOINT}/most-used") 

175 return PagedListResponse[Item].model_validate(resp.json()) 

176 

177 async def get_best_sales(self, *, year: int) -> PagedListResponse[Item]: 

178 """ 

179 Récupère les articles les plus vendus. 

180 

181 Arguments 

182 - ``year`` (int) : Année de recherche. 

183 

184 Returns 

185 - ``PagedListResponse[Item]`` : Liste paginée d'articles. 

186 """ 

187 params = { 

188 "year": year 

189 } 

190 resp = await self._c.request("GET", f"{ITEMS_ENDPOINT}/best-sales", params=params) 

191 return PagedListResponse[Item].model_validate(resp.json())