Coverage for polar/transaction/service/processor_fee.py: 13%

1from datetime import UTC, datetime 

2from typing import Literal 

3 

4from polar.enums import PaymentProcessor 

5from polar.integrations.stripe.service import stripe as stripe_service 

6from polar.integrations.stripe.utils import get_expandable_id 

7from polar.models import Refund, Transaction 

8from polar.models.transaction import Processor, ProcessorFeeType, TransactionType 

9from polar.postgres import AsyncSession 

10 

11from .base import BaseTransactionService, BaseTransactionServiceError 

12 

13 

14class ProcessorFeeTransactionError(BaseTransactionServiceError): ... 14 ↛ 17line 14 didn't jump to line 17 because

15 

16 

17class BalanceTransactionNotFound(ProcessorFeeTransactionError): 

18 def __init__(self, payment_transaction: Transaction) -> None: 

19 message = ( 

20 f"Balance transaction not found for payment transaction " 

21 f"{payment_transaction.id} with charge ID {payment_transaction.charge_id}" 

22 ) 

23 super().__init__(message) 

24 

25 

26class UnsupportedStripeFeeType(ProcessorFeeTransactionError): 

27 def __init__(self, description: str) -> None: 

28 self.description = description 

29 message = f"Unsupported Stripe fee type: {description}" 

30 super().__init__(message) 

31 

32 

33def _get_stripe_processor_fee_type(description: str) -> ProcessorFeeType: 

34 description = description.lower() 

35 if "payout fee" in description or "account volume" in description: 

36 return ProcessorFeeType.payout 

37 if "cross-border transfers" in description: 

38 return ProcessorFeeType.cross_border_transfer 

39 if "active account" in description: 

40 return ProcessorFeeType.account 

41 if "billing" in description: 

42 return ProcessorFeeType.subscription 

43 if ( 

44 "automatic tax" in description 

45 or "tax api calculation" in description 

46 or "tax api transaction" in description 

47 ): 

48 return ProcessorFeeType.tax 

49 if "invoicing" in description or "post payment invoices" in description: 

50 return ProcessorFeeType.invoice 

51 # Instant Bank Account Validation for ACH payments 

52 if "connections verification" in description: 

53 return ProcessorFeeType.payment 

54 if "radar" in description: 

55 return ProcessorFeeType.security 

56 if "3d secure" in description: 

57 return ProcessorFeeType.payment 

58 if "authorization optimization" in description: 

59 return ProcessorFeeType.payment 

60 if "card account updater" in description: 

61 return ProcessorFeeType.payment 

62 if "tax reporting for connect" in description: 

63 return ProcessorFeeType.tax 

64 if "identity document check" in description: 

65 return ProcessorFeeType.security 

66 if "payments" in description: 

67 return ProcessorFeeType.payment 

68 if "card dispute countered fee" in description: 

69 return ProcessorFeeType.dispute 

70 raise UnsupportedStripeFeeType(description) 

71 

72 

73class ProcessorFeeTransactionService(BaseTransactionService): 

74 async def create_payment_fees( 

75 self, session: AsyncSession, *, payment_transaction: Transaction 

76 ) -> list[Transaction]: 

77 fee_transactions: list[Transaction] = [] 

78 

79 if payment_transaction.processor != Processor.stripe: 

80 return fee_transactions 

81 

82 if payment_transaction.charge_id is None: 

83 return fee_transactions 

84 

85 charge = await stripe_service.get_charge(payment_transaction.charge_id) 

86 

87 if charge.balance_transaction is None: 

88 raise BalanceTransactionNotFound(payment_transaction) 

89 

90 stripe_balance_transaction = await stripe_service.get_balance_transaction( 

91 get_expandable_id(charge.balance_transaction) 

92 ) 

93 payment_fee_transaction = Transaction( 

94 type=TransactionType.processor_fee, 

95 processor=Processor.stripe, 

96 processor_fee_type=ProcessorFeeType.payment, 

97 currency=payment_transaction.currency, 

98 amount=-stripe_balance_transaction.fee, 

99 account_currency=payment_transaction.currency, 

100 account_amount=-stripe_balance_transaction.fee, 

101 tax_amount=0, 

102 incurred_by_transaction=payment_transaction, 

103 ) 

104 session.add(payment_fee_transaction) 

105 fee_transactions.append(payment_fee_transaction) 

106 

107 await session.flush() 

108 

109 return fee_transactions 

110 

111 async def create_refund_fees( 

112 self, 

113 session: AsyncSession, 

114 *, 

115 refund: Refund, 

116 refund_transaction: Transaction, 

117 ) -> list[Transaction]: 

118 fee_transactions: list[Transaction] = [] 

119 

120 is_stripe_refund = refund.processor == PaymentProcessor.stripe 

121 is_stripe_refund_trx = refund_transaction.processor == Processor.stripe 

122 if not (is_stripe_refund and is_stripe_refund_trx): 

123 return fee_transactions 

124 

125 if refund_transaction.refund_id is None: 

126 return fee_transactions 

127 

128 if refund.processor_balance_transaction_id is None: 

129 return fee_transactions 

130 

131 balance_transaction = await stripe_service.get_balance_transaction( 

132 refund.processor_balance_transaction_id, 

133 ) 

134 

135 refund_fee_transaction = Transaction( 

136 type=TransactionType.processor_fee, 

137 processor=refund_transaction.processor, 

138 processor_fee_type=ProcessorFeeType.refund, 

139 currency=refund_transaction.currency, 

140 amount=-balance_transaction.fee, 

141 account_currency=refund_transaction.currency, 

142 account_amount=-balance_transaction.fee, 

143 tax_amount=0, 

144 incurred_by_transaction_id=refund_transaction.id, 

145 ) 

146 

147 session.add(refund_fee_transaction) 

148 fee_transactions.append(refund_fee_transaction) 

149 

150 await session.flush() 

151 

152 return fee_transactions 

153 

154 async def create_dispute_fees( 

155 self, 

156 session: AsyncSession, 

157 *, 

158 dispute_transaction: Transaction, 

159 category: Literal["dispute", "dispute_reversal"], 

160 ) -> list[Transaction]: 

161 fee_transactions: list[Transaction] = [] 

162 

163 if dispute_transaction.processor != Processor.stripe: 

164 return fee_transactions 

165 

166 if dispute_transaction.dispute_id is None: 

167 return fee_transactions 

168 

169 dispute = await stripe_service.get_dispute(dispute_transaction.dispute_id) 

170 for balance_transaction in dispute.balance_transactions: 

171 if ( 

172 balance_transaction.reporting_category == category 

173 and balance_transaction.fee > 0 

174 ): 

175 dispute_fee_transaction = Transaction( 

176 type=TransactionType.processor_fee, 

177 processor=Processor.stripe, 

178 processor_fee_type=ProcessorFeeType.dispute, 

179 currency=dispute_transaction.currency, 

180 amount=-balance_transaction.fee, 

181 account_currency=dispute_transaction.currency, 

182 account_amount=-balance_transaction.fee, 

183 tax_amount=0, 

184 incurred_by_transaction_id=dispute_transaction.id, 

185 ) 

186 session.add(dispute_fee_transaction) 

187 fee_transactions.append(dispute_fee_transaction) 

188 

189 await session.flush() 

190 

191 return fee_transactions 

192 

193 async def sync_stripe_fees(self, session: AsyncSession) -> list[Transaction]: 

194 transactions: list[Transaction] = [] 

195 

196 balance_transactions = await stripe_service.list_balance_transactions( 

197 type="stripe_fee", expand=["data.source"] 

198 ) 

199 async for balance_transaction in balance_transactions: 

200 transaction = await self.get_by( 

201 session, fee_balance_transaction_id=balance_transaction.id 

202 ) 

203 

204 # We reached the point where we have already synced all the fees 

205 if transaction is not None: 

206 break 

207 

208 if balance_transaction.description is None: 

209 continue 

210 

211 processor_fee_type = _get_stripe_processor_fee_type( 

212 balance_transaction.description 

213 ) 

214 transaction = Transaction( 

215 created_at=datetime.fromtimestamp(balance_transaction.created, tz=UTC), 

216 type=TransactionType.processor_fee, 

217 processor=Processor.stripe, 

218 processor_fee_type=processor_fee_type, 

219 currency=balance_transaction.currency, 

220 amount=balance_transaction.net, 

221 account_currency=balance_transaction.currency, 

222 account_amount=balance_transaction.net, 

223 tax_amount=0, 

224 fee_balance_transaction_id=balance_transaction.id, 

225 ) 

226 session.add(transaction) 

227 transactions.append(transaction) 

228 

229 await session.flush() 

230 

231 return transactions 

232 

233 

234processor_fee_transaction = ProcessorFeeTransactionService(Transaction)