Coverage for polar/refund/schemas.py: 59%

1import inspect 

2from typing import Annotated, Any 

3from uuid import UUID 

4 

5import stripe as stripe_lib 

6from babel.numbers import format_currency 

7from fastapi import Path 

8from pydantic import UUID4, Field 

9 

10from polar.enums import PaymentProcessor 

11from polar.kit.metadata import ( 

12 MetadataInputMixin, 

13 MetadataOutputMixin, 

14) 

15from polar.kit.schemas import IDSchema, Schema, TimestampedSchema 

16from polar.models.refund import ( 

17 RefundFailureReason, 

18 RefundReason, 

19 RefundStatus, 

20) 

21 

22RefundID = Annotated[UUID4, Path(description="The refund ID.")] 

23 

24 

25class Refund(MetadataOutputMixin, IDSchema, TimestampedSchema): 

26 status: RefundStatus 

27 reason: RefundReason 

28 amount: int 

29 tax_amount: int 

30 currency: str 

31 organization_id: UUID4 

32 order_id: UUID4 

33 subscription_id: UUID4 | None 

34 customer_id: UUID4 

35 revoke_benefits: bool 

36 

37 def get_amount_display(self) -> str: 

38 return f"{ 

39 format_currency( 

40 self.amount / 100, 

41 self.currency.upper(), 

42 locale='en_US', 

43 ) 

44 }" 

45 

46 

47class RefundCreate(MetadataInputMixin, Schema): 

48 order_id: UUID4 

49 reason: RefundReason 

50 amount: int = Field(description="Amount to refund in cents. Minimum is 1.", gt=0) 

51 comment: str | None = Field( 

52 None, description="An internal comment about the refund." 

53 ) 

54 revoke_benefits: bool = Field( 

55 False, 

56 description=inspect.cleandoc( 

57 """ 

58 Should this refund trigger the associated customer benefits to be revoked? 

59 

60 **Note:** 

61 Only allowed in case the `order` is a one-time purchase. 

62 Subscriptions automatically revoke customer benefits once the 

63 subscription itself is revoked, i.e fully canceled. 

64 """ 

65 ), 

66 ) 

67 

68 

69class InternalRefundCreate(MetadataInputMixin, Schema): 

70 status: RefundStatus 

71 reason: RefundReason 

72 amount: int 

73 tax_amount: int 

74 currency: str 

75 comment: str | None = None 

76 failure_reason: RefundFailureReason | None 

77 destination_details: dict[str, Any] = {} 

78 order_id: UUID | None 

79 subscription_id: UUID | None 

80 customer_id: UUID | None 

81 organization_id: UUID | None 

82 pledge_id: UUID | None 

83 processor: PaymentProcessor 

84 processor_id: str 

85 processor_receipt_number: str | None 

86 processor_reason: str 

87 processor_balance_transaction_id: str | None 

88 revoke_benefits: bool = False 

89 

90 @classmethod 

91 def from_stripe( 

92 cls, 

93 stripe_refund: stripe_lib.Refund, 

94 *, 

95 refunded_amount: int, 

96 refunded_tax_amount: int, 

97 order_id: UUID | None = None, 

98 subscription_id: UUID | None = None, 

99 customer_id: UUID | None = None, 

100 organization_id: UUID | None = None, 

101 pledge_id: UUID | None = None, 

102 ) -> "InternalRefundCreate": 

103 failure_reason = getattr(stripe_refund, "failure_reason", None) 

104 failure_reason = RefundFailureReason.from_stripe(failure_reason) 

105 stripe_reason = stripe_refund.reason if stripe_refund.reason else "other" 

106 reason = RefundReason.from_stripe(stripe_refund.reason) 

107 

108 destination_details: dict[str, Any] = getattr( 

109 stripe_refund, "destination_details", {} 

110 ) 

111 

112 status = RefundStatus.pending 

113 if stripe_refund.status: 

114 status = RefundStatus(stripe_refund.status) 

115 

116 balance_transaction_id = None 

117 if stripe_refund.balance_transaction: 

118 balance_transaction_id = str(stripe_refund.balance_transaction) 

119 

120 # Skip validation from trusted source (Stripe) 

121 return cls.model_construct( 

122 status=status, 

123 reason=reason, 

124 amount=refunded_amount, 

125 tax_amount=refunded_tax_amount, 

126 currency=stripe_refund.currency, 

127 failure_reason=failure_reason, 

128 destination_details=destination_details, 

129 order_id=order_id, 

130 subscription_id=subscription_id, 

131 customer_id=customer_id, 

132 organization_id=organization_id, 

133 pledge_id=pledge_id, 

134 revoke_benefits=False, 

135 processor=PaymentProcessor.stripe, 

136 processor_id=stripe_refund.id, 

137 processor_receipt_number=stripe_refund.receipt_number, 

138 processor_reason=stripe_reason, 

139 processor_balance_transaction_id=balance_transaction_id, 

140 )