Coverage for functions \ flipdare \ payments \ dto \ charge_dto.py: 83%

1from __future__ import annotations 

2from datetime import datetime 

3from typing import Self 

4from stripe import Charge as StripeCharge 

5from stripe import BalanceTransaction as StripeBalanceTransaction 

6 

7from flipdare.app_log import LOG 

8from flipdare.error.app_stripe_error import AppStripeError 

9from flipdare.message.user_error_code import UserStripeErrorCode 

10from flipdare.payments.dto.safe_stripe_object import SafeStripeObject 

11from flipdare.util.time_util import TimeUtil 

12from flipdare.constants import DEFAULT_REAUTHORIZATION_DAYS_CUSTOMER 

13 

14 

15class ChargeDTO: 

16 __slots__ = ( 

17 "_amount", 

18 "_amount_captured", 

19 "_amount_refunded", 

20 "_app_fee_amount", 

21 "_available_on", 

22 "_balance_id", 

23 "_balance_status", 

24 "_capture_before", 

25 "_captured", 

26 "_charge_id", 

27 "_extended_auth", 

28 "_extended_auth_status", 

29 "_stripe_error_code", 

30 "_stripe_error_message", 

31 "_stripe_fee_amount", 

32 ) 

33 

34 _charge_id: str 

35 _balance_id: str | None 

36 _balance_status: str | None 

37 _available_on: int | None 

38 _captured: bool 

39 

40 _stripe_error_code: str | None 

41 _stripe_error_message: str | None 

42 

43 _amount: int | None 

44 _amount_captured: int | None 

45 _amount_refunded: int | None 

46 

47 _stripe_fee_amount: int | None 

48 _app_fee_amount: int | None 

49 

50 _extended_auth: bool 

51 _extended_auth_status: str | None 

52 _capture_before: datetime 

53 

54 def __init__( 

55 self, 

56 charge_id: str, 

57 captured: bool, 

58 capture_before: datetime, 

59 amount: int, 

60 amount_captured: int, 

61 amount_refunded: int, 

62 stripe_error_code: str | None = None, 

63 stripe_error_message: str | None = None, 

64 stripe_fee_amount: int | None = None, 

65 app_fee_amount: int | None = None, 

66 extended_auth: bool = False, 

67 balance_id: str | None = None, 

68 balance_status: str | None = None, 

69 available_on: int | None = None, 

70 extended_auth_status: str | None = None, 

71 ) -> None: 

72 self._captured = captured 

73 self._charge_id = charge_id 

74 self._amount = amount 

75 self._amount_captured = amount_captured 

76 self._amount_refunded = amount_refunded 

77 self._stripe_error_code = stripe_error_code 

78 self._stripe_error_message = stripe_error_message 

79 self._stripe_fee_amount = stripe_fee_amount 

80 self._app_fee_amount = app_fee_amount 

81 self._extended_auth = extended_auth 

82 self._extended_auth_status = extended_auth_status 

83 self._capture_before = capture_before 

84 self._balance_id = balance_id 

85 self._balance_status = balance_status 

86 self._available_on = available_on 

87 

88 @classmethod 

89 def from_charge(cls, charge: StripeCharge) -> Self: # noqa: PLR0915 

90 # NOTE: StripeBalanceTransaction is the authoritative source for: 

91 # NOTE: Fee calculations 

92 # NOTE: Actual money movement 

93 # NOTE: Net amounts received 

94 # NOTE: Settlement currency 

95 # NOTE: charge.amount_captured is only authoritative for 

96 # NOTE: customer-facing amounts and billing records. 

97 # this can occur if the charge.currency is different from the settlement currency 

98 charge_id = charge.id 

99 

100 extended_auth = False 

101 extended_auth_status = None 

102 capture_before = None 

103 balance_status = None 

104 available_on = None 

105 

106 safe = SafeStripeObject(charge) 

107 

108 stripe_error_code = safe.failure_code.unwrap() 

109 stripe_error_message = safe.failure_message.unwrap() 

110 

111 captured = charge.captured 

112 created = charge.created 

113 amount = charge.amount 

114 amount_captured = charge.amount_captured 

115 amount_refunded = charge.amount_refunded 

116 

117 # None means the charge has not been captured .. 

118 app_fee_amount = safe.application_fee_amount.unwrap() or None 

119 # this is returned in the charge.currency 

120 stripe_fee_amount = None 

121 balance_transaction_id = None 

122 

123 balance = safe.balance_transaction 

124 if balance: 

125 balance_transaction = balance.unwrap() 

126 if isinstance(balance_transaction, str): 

127 # not expanded, balance is a _balance_transaction_id 

128 # this is an error because we cannot accurately calculate fees 

129 # or net amounts without the balance transaction details. 

130 msg = f"Balance transaction for charge {charge_id} is not expanded. Cant get fee" 

131 LOG().error(msg) 

132 raise AppStripeError.invalid_data( 

133 endpoint="ChargeDTO.from_charge", 

134 invalid_error_code=UserStripeErrorCode.BALANCE_NOT_PRESENT, 

135 cause=msg, 

136 ) 

137 

138 assert isinstance(balance_transaction, StripeBalanceTransaction) 

139 # expanded, balance is a StripeBalanceTransaction object 

140 balance_transaction_id = balance_transaction.id 

141 balance_status = balance_transaction.status 

142 available_on = balance_transaction.available_on 

143 

144 stripe_fee_amount = balance_transaction.fee 

145 bal_amount = balance_transaction.amount 

146 if amount_captured and bal_amount != amount_captured: 

147 # this is a sanity check, if the balance amount doesn't match the charge amount, we log it. 

148 # this should never happen, but if it does, it's important to know about it. 

149 # NOTE: we use the charge amount captured, not the balance amount, 

150 msg = f"Balance amount {bal_amount} does not match captured amount {amount_captured} for charge {charge_id}" 

151 LOG().warning(msg) 

152 

153 # Check if card payment method details exist 

154 card_details = safe.payment_method_details.card 

155 cap_before: int | None = None 

156 

157 if card_details: 

158 extend_auth = card_details.extended_authorization.status 

159 cap_before = card_details.capture_before.unwrap() 

160 LOG().warning(f"Cap before for charge {charge_id} is {cap_before}") 

161 if extend_auth: 

162 status = extend_auth.unwrap() 

163 if status == "enabled": 

164 extended_auth = True 

165 extended_auth_status = status 

166 else: 

167 extended_auth_status = status 

168 

169 if cap_before is not None: 

170 capture_before = TimeUtil.epoch_to_utc_dt(cap_before) 

171 else: 

172 # add DEFAULT_REAUTHORIZATION_DAYS_CUSTOMER days to created 

173 # to get an approximate capture before data 

174 # for customer initiated transactions (Which is all we do) 

175 created_dt = TimeUtil.epoch_to_utc_dt(created) 

176 capture_before = TimeUtil.get_utc_time_future_days( 

177 created_dt, DEFAULT_REAUTHORIZATION_DAYS_CUSTOMER 

178 ) 

179 

180 return cls( 

181 charge_id=charge_id, 

182 captured=captured, 

183 stripe_error_code=stripe_error_code, 

184 stripe_error_message=stripe_error_message, 

185 amount=amount, 

186 amount_captured=amount_captured, 

187 amount_refunded=amount_refunded, 

188 stripe_fee_amount=stripe_fee_amount, 

189 app_fee_amount=app_fee_amount, 

190 extended_auth=extended_auth, 

191 extended_auth_status=extended_auth_status, 

192 capture_before=capture_before, 

193 balance_id=balance_transaction_id, 

194 balance_status=balance_status, 

195 available_on=available_on, 

196 ) 

197 

198 @property 

199 def charge_id(self) -> str: 

200 return self._charge_id 

201 

202 @property 

203 def balance_id(self) -> str | None: 

204 return self._balance_id 

205 

206 @property 

207 def amount(self) -> int | None: 

208 return self._amount 

209 

210 @property 

211 def amount_captured(self) -> int | None: 

212 return self._amount_captured 

213 

214 @property 

215 def amount_refunded(self) -> int | None: 

216 return self._amount_refunded 

217 

218 @property 

219 def stripe_fee_amount(self) -> int | None: 

220 return self._stripe_fee_amount 

221 

222 @property 

223 def app_fee_amount(self) -> int | None: 

224 return self._app_fee_amount 

225 

226 @property 

227 def extended_auth(self) -> bool: 

228 return self._extended_auth 

229 

230 @property 

231 def extended_auth_status(self) -> str | None: 

232 return self._extended_auth_status 

233 

234 @property 

235 def capture_before(self) -> datetime | None: 

236 return self._capture_before 

237 

238 @property 

239 def stripe_error_code(self) -> str | None: 

240 return self._stripe_error_code 

241 

242 @property 

243 def stripe_error_message(self) -> str | None: 

244 return self._stripe_error_message 

245 

246 @property 

247 def is_error(self) -> bool: 

248 return self._stripe_error_code is not None or self._stripe_error_message is not None 

249 

250 @property 

251 def is_available(self) -> bool: 

252 balance_status = self._balance_status 

253 if balance_status is None: 

254 return False 

255 

256 return balance_status == "available" 

257 

258 @property 

259 def is_pending(self) -> bool: 

260 balance_status = self._balance_status 

261 if balance_status is None: 

262 return False 

263 

264 return balance_status == "pending" 

265 

266 @property 

267 def available_on(self) -> datetime | None: 

268 if self._available_on is None: 

269 return None 

270 

271 return TimeUtil.epoch_to_utc_dt(self._available_on) 

272 

273 def debug_str(self) -> str: 

274 return ( 

275 f"ChargeDTO:\n" 

276 f"\tcharge_id={self._charge_id}\n" 

277 f"\tamount={self._amount}\n" 

278 f"\tamount_captured={self._amount_captured}\n" 

279 f"\tamount_refunded={self._amount_refunded}\n" 

280 f"\tstripe_fee_amount={self._stripe_fee_amount}\n" 

281 f"\tapp_fee_amount={self._app_fee_amount}\n" 

282 f"\textended_auth={self._extended_auth}\n" 

283 f"\textended_auth_status={self._extended_auth_status}\n" 

284 f"\tcapture_before={self._capture_before}\n" 

285 f"\tstripe_error_code={self._stripe_error_code}\n" 

286 f"\tstripe_error_message={self._stripe_error_message}\n" 

287 )