Coverage for polar/refund/schemas.py: 75%
65 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1import inspect 1a
2from typing import Annotated, Any 1a
3from uuid import UUID 1a
5import stripe as stripe_lib 1a
6from babel.numbers import format_currency 1a
7from fastapi import Path 1a
8from pydantic import UUID4, Field 1a
10from polar.enums import PaymentProcessor 1a
11from polar.kit.metadata import ( 1a
12 MetadataInputMixin,
13 MetadataOutputMixin,
14)
15from polar.kit.schemas import IDSchema, Schema, TimestampedSchema 1a
16from polar.models.refund import ( 1a
17 RefundFailureReason,
18 RefundReason,
19 RefundStatus,
20)
22RefundID = Annotated[UUID4, Path(description="The refund ID.")] 1a
25class Refund(MetadataOutputMixin, IDSchema, TimestampedSchema): 1a
26 status: RefundStatus 1a
27 reason: RefundReason 1a
28 amount: int 1a
29 tax_amount: int 1a
30 currency: str 1a
31 organization_id: UUID4 1a
32 order_id: UUID4 1a
33 subscription_id: UUID4 | None 1a
34 customer_id: UUID4 1a
35 revoke_benefits: bool 1a
37 def get_amount_display(self) -> str: 1a
38 return f"{
39 format_currency(
40 self.amount / 100,
41 self.currency.upper(),
42 locale='en_US',
43 )
44 }"
47class RefundCreate(MetadataInputMixin, Schema): 1a
48 order_id: UUID4 1a
49 reason: RefundReason 1a
50 amount: int = Field(description="Amount to refund in cents. Minimum is 1.", gt=0) 1a
51 comment: str | None = Field( 1a
52 None, description="An internal comment about the refund."
53 )
54 revoke_benefits: bool = Field( 1a
55 False,
56 description=inspect.cleandoc(
57 """
58 Should this refund trigger the associated customer benefits to be revoked?
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 )
69class InternalRefundCreate(MetadataInputMixin, Schema): 1a
70 status: RefundStatus 1a
71 reason: RefundReason 1a
72 amount: int 1a
73 tax_amount: int 1a
74 currency: str 1a
75 comment: str | None = None 1a
76 failure_reason: RefundFailureReason | None 1a
77 destination_details: dict[str, Any] = {} 1a
78 order_id: UUID | None 1a
79 subscription_id: UUID | None 1a
80 customer_id: UUID | None 1a
81 organization_id: UUID | None 1a
82 pledge_id: UUID | None 1a
83 processor: PaymentProcessor 1a
84 processor_id: str 1a
85 processor_receipt_number: str | None 1a
86 processor_reason: str 1a
87 processor_balance_transaction_id: str | None 1a
88 revoke_benefits: bool = False 1a
90 @classmethod 1a
91 def from_stripe( 1a
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)
108 destination_details: dict[str, Any] = getattr(
109 stripe_refund, "destination_details", {}
110 )
112 status = RefundStatus.pending
113 if stripe_refund.status:
114 status = RefundStatus(stripe_refund.status)
116 balance_transaction_id = None
117 if stripe_refund.balance_transaction:
118 balance_transaction_id = str(stripe_refund.balance_transaction)
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 )