Coverage for packages/clopos/src/integrify/clopos/handlers.py: 92%

1# pylint: disable=wrong-import-order,ungrouped-imports 

2import json 

3from functools import cached_property 

4from typing import Annotated 

5 

6from integrify.api import APIPayloadHandler 

7from integrify.clopos import env 

8from integrify.clopos.schemas.auth.request import AuthRequest 

9from integrify.clopos.schemas.auth.response import AuthResponse 

10from integrify.clopos.schemas.categories.object import Category 

11from integrify.clopos.schemas.categories.request import GetCategoriesRequest, GetCategoryByIDRequest 

12from integrify.clopos.schemas.common.request import ByIDRequest, PaginatedDataRequest 

13from integrify.clopos.schemas.common.response import ( 

14 ErrorResponse, 

15 ObjectListResponse, 

16 ObjectResponse, 

17) 

18from integrify.clopos.schemas.customers.object import Customer 

19from integrify.clopos.schemas.customers.request import CreateCustomerRequest, GetCustomersRequest 

20from integrify.clopos.schemas.orders.object import Order 

21from integrify.clopos.schemas.orders.request import ( 

22 CreateOrderRequest, 

23 GetOrderByIDRequest, 

24 GetOrdersRequest, 

25 UpdateOrderRequest, 

26) 

27from integrify.clopos.schemas.products.object import Product, StopList 

28from integrify.clopos.schemas.products.request import ( 

29 GetProductByIDRequest, 

30 GetProductsRequest, 

31 GetStopListRequest, 

32) 

33from integrify.clopos.schemas.receipts.object import Receipt, ReceiptStockOperation 

34from integrify.clopos.schemas.receipts.request import ( 

35 CloseReceiptRequest, 

36 GetReceiptsRequest, 

37 UpdateClosedReceiptRequest, 

38) 

39from integrify.clopos.schemas.stations.object import Station 

40from integrify.clopos.schemas.stations.request import GetStationsRequest 

41from pydantic import Field 

42 

43 

44class AuthHandler(APIPayloadHandler): 

45 def __init__( 

46 self, 

47 req_model=AuthRequest, 

48 resp_model=Annotated[AuthResponse | ErrorResponse, Field(discriminator='success')], 

49 dry=False, 

50 ): 

51 super().__init__(req_model, resp_model, dry) 

52 

53 

54class AuthedAPIPayloadHandler(APIPayloadHandler): 

55 def __init__(self, req_model=None, resp_model=None, dry=False): 

56 super().__init__( 

57 req_model, 

58 Annotated[resp_model | ErrorResponse, Field(discriminator='success')], 

59 dry, 

60 ) 

61 

62 @cached_property 

63 def headers(self): 

64 default = super().headers 

65 

66 # v2: yalnız `x-token` tələb olunur (JWT brand/venue/integrator-i encode edir). 

67 # `x-venue` opsional olaraq JWT-dəki venue-nu konkret sorğu üçün override edir. 

68 if env.CLOPOS_VENUE_ID: 68 ↛ 69line 68 didn't jump to line 69 because the condition on line 68 was never true

69 default['x-venue'] = env.CLOPOS_VENUE_ID 

70 

71 return default 

72 

73 

74def GetPaginatedDataHandler(object_type, req_model=PaginatedDataRequest): # pylint: disable=invalid-name 

75 """Function to dynamically create ObjectListResponse[object_type]""" 

76 

77 class _GetPaginatedDataHandler(AuthedAPIPayloadHandler): 

78 def __init__(self, dry=False): 

79 super().__init__( 

80 req_model=req_model, 

81 resp_model=ObjectListResponse[object_type], 

82 dry=dry, 

83 ) 

84 

85 return _GetPaginatedDataHandler 

86 

87 

88def GetByIDHandler(object_type, req_model=ByIDRequest): # pylint: disable=invalid-name 

89 """Function to dynamically create ObjectResponse[object_type]""" 

90 

91 class _GetByIDHandler(AuthedAPIPayloadHandler): 

92 def __init__(self, dry=False): 

93 super().__init__( 

94 req_model=req_model, 

95 resp_model=ObjectResponse[object_type], 

96 dry=dry, 

97 ) 

98 

99 return _GetByIDHandler 

100 

101 

102################################################################################################### 

103 

104 

105class GetCustomersHandler(AuthedAPIPayloadHandler): 

106 def __init__( 

107 self, 

108 req_model=GetCustomersRequest, 

109 resp_model=ObjectListResponse[Customer], 

110 dry=False, 

111 ): 

112 super().__init__(req_model, resp_model, dry) 

113 

114 

115class CreateCustomerHandler(AuthedAPIPayloadHandler): 

116 def __init__( 

117 self, 

118 req_model=CreateCustomerRequest, 

119 resp_model=ObjectResponse[Customer], 

120 dry=False, 

121 ): 

122 super().__init__(req_model, resp_model, dry) 

123 

124 

125class GetCategoriesHandler(AuthedAPIPayloadHandler): 

126 def __init__( 

127 self, 

128 req_model=GetCategoriesRequest, 

129 resp_model=ObjectListResponse[Category], 

130 dry=False, 

131 ): 

132 super().__init__(req_model, resp_model, dry) 

133 

134 

135class GetCategoryByIDHandler(AuthedAPIPayloadHandler): 

136 def __init__( 

137 self, 

138 req_model=GetCategoryByIDRequest, 

139 resp_model=ObjectResponse[Category], 

140 dry=False, 

141 ): 

142 super().__init__(req_model, resp_model, dry) 

143 

144 

145class GetStationsHandler(AuthedAPIPayloadHandler): 

146 def __init__( 

147 self, 

148 req_model=GetStationsRequest, 

149 resp_model=ObjectListResponse[Station], 

150 dry=False, 

151 ): 

152 super().__init__(req_model, resp_model, dry) 

153 

154 

155class GetProductsHandler(AuthedAPIPayloadHandler): 

156 def __init__( 

157 self, 

158 req_model=GetProductsRequest, 

159 resp_model=ObjectListResponse[Product], 

160 dry=False, 

161 ): 

162 super().__init__(req_model, resp_model, dry) 

163 

164 def post_handle_payload(self, data): 

165 return json.dumps(data) # for urlencoding 

166 

167 

168class GetProductByIDHandler(AuthedAPIPayloadHandler): 

169 def __init__( 

170 self, 

171 req_model=GetProductByIDRequest, 

172 resp_model=ObjectResponse[Product], 

173 dry=False, 

174 ): 

175 super().__init__(req_model, resp_model, dry) 

176 

177 

178class GetStopListHandler(AuthedAPIPayloadHandler): 

179 def __init__( 

180 self, 

181 req_model=GetStopListRequest, 

182 resp_model=ObjectListResponse[StopList], 

183 dry=False, 

184 ): 

185 super().__init__(req_model, resp_model, dry) 

186 

187 

188class GetOrdersHandler(AuthedAPIPayloadHandler): 

189 def __init__( 

190 self, 

191 req_model=GetOrdersRequest, 

192 resp_model=ObjectListResponse[Order], 

193 dry=False, 

194 ): 

195 super().__init__(req_model, resp_model, dry) 

196 

197 

198class GetOrderByIDHandler(AuthedAPIPayloadHandler): 

199 def __init__(self, req_model=GetOrderByIDRequest, resp_model=ObjectResponse[Order], dry=False): 

200 super().__init__(req_model, resp_model, dry) 

201 

202 

203class CreateOrderHandler(AuthedAPIPayloadHandler): 

204 def __init__(self, req_model=CreateOrderRequest, resp_model=ObjectResponse[Order], dry=False): 

205 super().__init__(req_model, resp_model, dry) 

206 

207 

208class UpdateOrderHandler(AuthedAPIPayloadHandler): 

209 def __init__( 

210 self, 

211 req_model=UpdateOrderRequest, 

212 resp_model=ObjectResponse[Order], 

213 dry=False, 

214 ): 

215 super().__init__(req_model, resp_model, dry) 

216 

217 

218class GetReceiptsHandler(AuthedAPIPayloadHandler): 

219 def __init__( 

220 self, 

221 req_model=GetReceiptsRequest, 

222 resp_model=ObjectListResponse[Receipt], 

223 dry=False, 

224 ): 

225 super().__init__(req_model, resp_model, dry) 

226 

227 

228class _UpdateReceiptHandler(AuthedAPIPayloadHandler): 

229 def handle_payload(self, req_model, *args, **kwds): 

230 # Fərqli olaraq base handler-dən, burada `URL_PARAM_FIELDS` body-dən çıxarılmır 

231 # (yəni id həm URL-də, həm body-də göndərilir). 

232 if self.req_model and req_model is not None: 

233 return req_model.model_dump( 

234 by_alias=True, 

235 mode='json', 

236 ) 

237 

238 # `req_model` yoxdursa, o zaman `*args` boş olmalıdır, çünki onların key-ləri bilinmir 

239 assert not args 

240 

241 return kwds 

242 

243 

244class UpdateClosedReceiptHandler(_UpdateReceiptHandler): 

245 def __init__( 

246 self, 

247 req_model=UpdateClosedReceiptRequest, 

248 resp_model=ObjectResponse[Receipt], 

249 dry=False, 

250 ): 

251 super().__init__(req_model, resp_model, dry) 

252 

253 

254class CloseReceiptHandler(_UpdateReceiptHandler): 

255 def __init__( 

256 self, 

257 req_model=CloseReceiptRequest, 

258 resp_model=ObjectResponse[Receipt], 

259 dry=False, 

260 ): 

261 super().__init__(req_model, resp_model, dry) 

262 

263 

264class GetReceiptStockOperationsHandler(AuthedAPIPayloadHandler): 

265 def __init__( 

266 self, 

267 req_model=ByIDRequest, 

268 resp_model=ObjectListResponse[ReceiptStockOperation], 

269 dry=False, 

270 ): 

271 super().__init__(req_model, resp_model, dry)