Coverage for src/integrify/kapital/handlers.py: 94%

1import base64 

2import json 

3from functools import cached_property 

4from typing import Optional, Type 

5 

6import httpx 

7from pydantic import BaseModel 

8 

9from integrify.api import APIPayloadHandler, APIResponse, _ResponseT 

10from integrify.kapital.env import ( 

11 KAPITAL_PASSWORD, 

12 KAPITAL_USERNAME, 

13) 

14from integrify.kapital.schemas.request import ( 

15 ClearingOrderRequestSchema, 

16 CreateOrderRequestSchema, 

17 FullReverseOrderRequestSchema, 

18 LinkCardTokenRequestSchema, 

19 OrderInformationRequestSchema, 

20 OrderWithSavedCardRequestSchema, 

21 PartialReverseOrderRequestSchema, 

22 PayAndSaveCardRequestSchema, 

23 ProcessPaymentWithSavedCardRequestSchema, 

24 RefundOrderRequestSchema, 

25 SaveCardRequestSchema, 

26) 

27from integrify.kapital.schemas.response import ( 

28 BaseResponseSchema, 

29 ClearingOrderResponseSchema, 

30 CreateOrderResponseSchema, 

31 DetailedOrderInformationResponseSchema, 

32 ErrorResponseBodySchema, 

33 FullReverseOrderResponseSchema, 

34 LinkCardTokenResponseSchema, 

35 OrderInformationResponseSchema, 

36 PartialReverseOrderResponseSchema, 

37 ProcessPaymentWithSavedCardResponseSchema, 

38 RefundOrderResponseSchema, 

39) 

40from integrify.schemas import PayloadBaseModel 

41 

42 

43class BasePayloadHandler(APIPayloadHandler): 

44 def __init__( 

45 self, 

46 req_model: Type[PayloadBaseModel], 

47 resp_model: Type[_ResponseT], 

48 data_key: Optional[str] = None, 

49 ): 

50 super().__init__(req_model, resp_model) 

51 self.data_key = data_key 

52 

53 @cached_property 

54 def headers(self): 

55 credentials = f'{KAPITAL_USERNAME}:{KAPITAL_PASSWORD}' 

56 encoded_credentials = base64.b64encode(credentials.encode()).decode() 

57 return { 

58 'Authorization': f'Basic {encoded_credentials}', 

59 'Content-Type': 'application/json', 

60 } 

61 

62 def post_handle_payload(self, data): 

63 if self.data_key: 63 ↛ 66line 63 didn't jump to line 66 because the condition on line 63 was always true

64 return json.dumps({self.data_key: data}) 

65 

66 return json.dumps(data) 

67 

68 def handle_response(self, resp: httpx.Response) -> APIResponse[_ResponseT]: 

69 """ 

70 Bu funksiya API-dən gələn cavabı status koduna görə tənzimləyir. 

71 Əgər status kodu 200-dürsə, gələn cavabı modelə uyğunlaşdırır və APIResponse obyektini qaytarır. 

72 200-dən fərqli status kodu gələrsə, gələn cavabı modelə uyğunlaşdırır və error obyektini APIResponse obyektinə əlavə edir. 

73 """ # noqa: E501 

74 

75 api_resp = APIResponse[BaseResponseSchema].model_validate(resp, from_attributes=True) 

76 

77 if resp.status_code == 200: 

78 if not self.resp_model: 78 ↛ 79line 78 didn't jump to line 79 because the condition on line 78 was never true

79 raise ValueError('Response model is not set for this handler.') 

80 

81 data = self.get_response_data(resp.json()) 

82 

83 assert issubclass(self.resp_model, BaseModel) 

84 api_resp.body.data = self.resp_model.model_validate(data, from_attributes=True) 

85 else: 

86 api_resp.body.error = ErrorResponseBodySchema.model_validate( 

87 resp.json(), 

88 from_attributes=True, 

89 ) 

90 

91 return api_resp # type: ignore[return-value] 

92 

93 def get_response_data(self, response_json: dict) -> dict: 

94 """`self.data_key` varsa, o key-dəki datanı götürmək""" 

95 if not self.data_key: 

96 raise NotImplementedError("Subclasses must define 'response_data_key'") 

97 

98 return response_json.get(self.data_key, {}) 

99 

100 

101class CreateOrderPayloadHandler(BasePayloadHandler): 

102 def __init__(self): 

103 super().__init__( 

104 CreateOrderRequestSchema, 

105 CreateOrderResponseSchema, 

106 data_key='order', 

107 ) 

108 

109 

110class OrderInformationPayloadHandler(BasePayloadHandler): 

111 def __init__(self): 

112 super().__init__( 

113 OrderInformationRequestSchema, 

114 OrderInformationResponseSchema, 

115 data_key='order', 

116 ) 

117 

118 def post_handle_payload(self, data): 

119 return data 

120 

121 

122class DetailedOrderInformationPayloadHandler(BasePayloadHandler): 

123 def __init__(self): 

124 super().__init__( 

125 OrderInformationRequestSchema, 

126 DetailedOrderInformationResponseSchema, 

127 data_key='order', 

128 ) 

129 

130 def post_handle_payload(self, data): 

131 return data 

132 

133 

134class RefundOrderPayloadHandler(BasePayloadHandler): 

135 def __init__(self): 

136 super().__init__( 

137 RefundOrderRequestSchema, 

138 RefundOrderResponseSchema, 

139 data_key='tran', 

140 ) 

141 

142 

143class SaveCardPayloadHandler(BasePayloadHandler): 

144 def __init__(self): 

145 super().__init__( 

146 SaveCardRequestSchema, 

147 CreateOrderResponseSchema, 

148 data_key='order', 

149 ) 

150 

151 

152class PayAndSaveCardPayloadHandler(BasePayloadHandler): 

153 def __init__(self): 

154 super().__init__( 

155 PayAndSaveCardRequestSchema, 

156 CreateOrderResponseSchema, 

157 data_key='order', 

158 ) 

159 

160 

161class FullReverseOrderPayloadHandler(BasePayloadHandler): 

162 def __init__(self): 

163 super().__init__( 

164 FullReverseOrderRequestSchema, 

165 FullReverseOrderResponseSchema, 

166 data_key='tran', 

167 ) 

168 

169 

170class ClearingOrderPayloadHandler(BasePayloadHandler): 

171 def __init__(self): 

172 super().__init__( 

173 ClearingOrderRequestSchema, 

174 ClearingOrderResponseSchema, 

175 data_key='tran', 

176 ) 

177 

178 

179class PartialReverseOrderPayloadHandler(BasePayloadHandler): 

180 def __init__(self): 

181 super().__init__( 

182 PartialReverseOrderRequestSchema, 

183 PartialReverseOrderResponseSchema, 

184 data_key='tran', 

185 ) 

186 

187 

188class OrderWithSavedCardPayloadHandler(BasePayloadHandler): 

189 def __init__(self): 

190 super().__init__( 

191 OrderWithSavedCardRequestSchema, 

192 CreateOrderResponseSchema, 

193 data_key='order', 

194 ) 

195 

196 

197class LinkCardTokenPayloadHandler(BasePayloadHandler): 

198 def __init__(self): 

199 super().__init__( 

200 LinkCardTokenRequestSchema, 

201 LinkCardTokenResponseSchema, 

202 data_key='order', 

203 ) 

204 

205 def post_handle_payload(self, data): 

206 return json.dumps( 

207 { 

208 'order': {'initiationEnvKind': 'Server'}, 

209 'token': {'storedId': data['token']}, 

210 } 

211 ) 

212 

213 

214class ProcessPaymentWithSavedCardPayloadHandler(BasePayloadHandler): 

215 def __init__(self): 

216 super().__init__( 

217 ProcessPaymentWithSavedCardRequestSchema, 

218 ProcessPaymentWithSavedCardResponseSchema, 

219 data_key='tran', 

220 )