Coverage for functions \ flipdare \ generated \ model \ payment \ payment_model.py: 100%
0 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-05-08 12:22 +1000
1#!/usr/bin/env python
2#
3# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved.
4#
5# This file is part of Flipdare's proprietary software and contains
6# confidential and copyrighted material. Unauthorised copying,
7# modification, distribution, or use of this file is strictly
8# prohibited without prior written permission from Flipdare Pty Ltd.
9#
10# This software includes third-party components licensed under MIT,
11# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details.
12#
13# NOTE: THIS FILE IS AUTO GENERATED. DO NOT EDIT.
14#
15# Generated by codegen_models.py
16#
17# Modify 'codegen_models.py'
18# and re-run the script above to update.
19#
20# pragma: no cover
21from __future__ import annotations
22from datetime import datetime
23from google.cloud.firestore_v1.transforms import Sentinel
24from flipdare.core.firestore_field import FirestoreField
25from flipdare.util.time_util import FirestoreTime
26from typing import Any, TypedDict, cast, Unpack
27from enum import StrEnum
28from pydantic import Field, ConfigDict, TypeAdapter
29from flipdare.firestore.core.app_base_model import AppBaseModel
30from flipdare.generated.model.payment.customer_info_model import (
31 CustomerInfoModel,
32 CustomerInfoDict,
33)
34from flipdare.generated.model.payment.account_info_model import AccountInfoModel, AccountInfoDict
35from flipdare.generated.model.payment.audit_info_model import AuditInfoModel, AuditInfoDict
36from flipdare.generated.model.payment.risk_assessment_model import (
37 RiskAssessmentModel,
38 RiskAssessmentDict,
39)
40from flipdare.generated.shared.payment.payment_status import PaymentStatus
41from flipdare.generated.shared.stripe.stripe_intent_status import StripeIntentStatus
42from flipdare.generated.model.payment.payment_event_model import (
43 PaymentEventModel,
44 PaymentEventDict,
45)
46from flipdare.generated.model.payment.payment_result_model import (
47 PaymentResultModel,
48 PaymentResultDict,
49)
50from flipdare.generated.model.payment.payment_schedule_model import (
51 PaymentScheduleModel,
52 PaymentScheduleDict,
53)
54from flipdare.app_log import LOG
55from flipdare.util.time_util import TimeUtil
56from flipdare.payments.dto.payment_intent_dto import PaymentIntentDTO
57from flipdare.generated.shared.payment.payment_event_status import PaymentEventStatus
58from flipdare.generated.shared.model.user.app_fee_type import AppFeeType
59from flipdare.generated.shared.stripe.stripe_currency_code import StripeCurrencyCode
62class PaymentKeys(StrEnum):
63 CREATED_AT = "created_at"
64 UPDATED_AT = "updated_at"
65 CUSTOMER_INFO = "customer_info"
66 ACCOUNT_INFO = "account_info"
67 AUDIT_INFO = "audit_info"
68 RISK_ASSESSMENT = "risk_assessment"
69 STATUS = "status"
70 INTENT_STATUS = "intent_status"
71 PAYMENT_INTENT_ID = "payment_intent_id"
72 PAYMENT_METHOD_ID = "payment_method_id"
73 AMOUNT = "amount"
74 LAST_EVENT = "last_event"
75 RESULT_TOTAL = "result_total"
76 CAPTURE_BEFORE = "capture_before"
77 SCHEDULE = "schedule"
78 STRIPE_DISPUTE_ID = "stripe_dispute_id"
81# !! IMPORTANT !!
82# !!
83# !! this should only be used in the database to query.
84# !!
85class PaymentInternalKeys(StrEnum):
86 CREATED_AT = "created_at"
87 UPDATED_AT = "updated_at"
90class PaymentModel(AppBaseModel):
91 """Represents a payment for a pledge, including charge information and status."""
93 model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True)
95 created_at: FirestoreField = Field(
96 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp())
97 )
98 updated_at: FirestoreField = Field(
99 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp())
100 )
101 customer_info: CustomerInfoModel
102 account_info: AccountInfoModel
103 audit_info: AuditInfoModel
104 risk_assessment: RiskAssessmentModel
105 status: PaymentStatus = Field(default=PaymentStatus.PENDING)
106 intent_status: StripeIntentStatus | None = None
107 payment_intent_id: str
108 payment_method_id: str | None = None
109 amount: int = Field(default=0)
110 last_event: PaymentEventModel | None = None
111 result_total: PaymentResultModel = Field(default_factory=lambda: PaymentResultModel())
112 capture_before: int | None = None
113 schedule: PaymentScheduleModel = Field(default_factory=lambda: PaymentScheduleModel())
114 stripe_dispute_id: str | None = None
116 @classmethod
117 def validate_partial(cls, **data: Unpack[PaymentDict]) -> dict[str, Any]:
118 """
119 Uses Unpack to give you autocomplete and static warnings
120 if you pass an invalid key or type in your code.
122 Returns a dict with Firestore field names (aliases) for use with batch.update().
123 """
124 result: dict[str, Any] = {}
125 for k, v in data.items():
126 if k in cls.__pydantic_fields__:
127 field_info = cls.__pydantic_fields__[k]
128 validated_value = cast(
129 "Any",
130 TypeAdapter(field_info.annotation).validate_python(v),
131 )
132 # Use alias if defined, otherwise use field name
133 output_key = field_info.alias or k
134 result[output_key] = validated_value
135 return result
137 # ---- Convenience factories -----------------------------------------
139 @classmethod
140 def create(
141 cls,
142 audit_info: AuditInfoModel,
143 risk_assessment: RiskAssessmentModel,
144 customer_info: CustomerInfoModel,
145 account_info: AccountInfoModel,
146 amount: int,
147 payment_intent_id: str,
148 intent_status: StripeIntentStatus | None = None,
149 payment_method_id: str | None = None,
150 capture_before: int | None = None,
151 ) -> PaymentModel:
152 return PaymentModel(
153 audit_info=audit_info,
154 risk_assessment=risk_assessment,
155 customer_info=customer_info,
156 account_info=account_info,
157 amount=amount,
158 intent_status=intent_status,
159 capture_before=capture_before,
160 payment_intent_id=payment_intent_id,
161 payment_method_id=payment_method_id,
162 )
164 # ---- Convenience predicates -----------------------------------------
166 @property
167 def is_captured(self) -> bool:
168 return (
169 self.last_event is not None and self.last_event.status == PaymentEventStatus.CAPTURED
170 )
172 @property
173 def is_transferred(self) -> bool:
174 return (
175 self.last_event is not None
176 and self.last_event.status == PaymentEventStatus.TRANSFERRED
177 )
179 @property
180 def capture_before_dt(self) -> datetime | None:
181 # NOTE: stripe store timestamps as epoch ints (hence UTC)
182 epoch = self.capture_before
183 if epoch is None:
184 return None
186 return TimeUtil.simple_epoch_to_utc_dt(epoch)
188 @capture_before_dt.setter
189 def capture_before_dt(self, value: datetime) -> None:
190 # convert to epoch seconds for stripe
191 epoch = TimeUtil.dt_to_simple_epoch(value)
192 self.capture_before = epoch
194 # ---- Nested Properties -----------------------------------------
196 @property
197 def amount_captured(self) -> int:
198 return self.result_total.amount_captured
200 @property
201 def amount_transferred(self) -> int:
202 return self.result_total.amount_transferred
204 @property
205 def stripe_fee_amount(self) -> int:
206 return self.result_total.stripe_fee_amount
208 @property
209 def app_fee_amount(self) -> int:
210 return self.result_total.app_fee_amount
212 @property
213 def customer_id(self) -> str:
214 return self.customer_info.customer_id
216 @property
217 def customer_currency_code(self) -> StripeCurrencyCode:
218 return self.customer_info.currency_code
220 @property
221 def account_id(self) -> str:
222 return self.account_info.account_id
224 @property
225 def account_currency_code(self) -> StripeCurrencyCode:
226 return self.account_info.currency_code
228 @property
229 def fee_type(self) -> AppFeeType:
230 return self.account_info.fee_type
232 @property
233 def latest_charge_id(self) -> str | None:
234 if self.last_event is None:
235 return None
236 return self.last_event.stripe_charge_id
238 @property
239 def event_status(self) -> PaymentEventStatus | None:
240 if self.last_event is None:
241 return None
242 return self.last_event.status
244 # ---- Processing Logic -----------------------------------------
246 def set_capture_on(self, intent_status: StripeIntentStatus, value: float) -> None:
247 if not intent_status.is_processable:
248 msg = (
249 f"Cannot set capture_on for payment intent {self.payment_intent_id} because "
250 f"intent status {intent_status} is not in a processable state."
251 )
252 LOG().error(msg)
253 raise ValueError(msg)
255 self.intent_status = intent_status
256 self.schedule.capture_on = value
257 self.status = PaymentStatus.CAPTURE
259 def set_transfer_on(self, intent_status: StripeIntentStatus, value: float) -> None:
260 if not intent_status.is_processable:
261 msg = (
262 f"Cannot set transfer_on for payment intent {self.payment_intent_id} because "
263 f"intent status {intent_status} is not in a processable state."
264 )
265 LOG().error(msg)
266 raise ValueError(msg)
268 self.intent_status = intent_status
269 self.schedule.transfer_on = value
270 self.status = PaymentStatus.TRANSFER
272 def set_refund_on(self, intent_status: StripeIntentStatus, value: float) -> None:
273 if not intent_status.is_processable:
274 msg = (
275 f"Cannot set refund_on for payment intent {self.payment_intent_id} because "
276 f"intent status {intent_status} is not in a processable state."
277 )
278 LOG().error(msg)
279 raise ValueError(msg)
281 self.intent_status = intent_status
282 self.schedule.refund_on = value
283 self.status = PaymentStatus.REFUND
285 def set_captured(
286 self,
287 intent_status: StripeIntentStatus,
288 amount_captured: int = 0,
289 stripe_fee_amount: int = 0,
290 app_fee_amount: int = 0,
291 ) -> None:
292 if not intent_status.is_completed:
293 msg = (
294 f"Cannot set captured for payment intent {self.payment_intent_id} because "
295 f"intent status {intent_status} is not in a completed state."
296 )
297 LOG().error(msg)
298 raise ValueError(msg)
300 self.intent_status = intent_status
301 result = self.result_total
302 result.amount_captured += amount_captured
303 result.stripe_fee_amount += stripe_fee_amount
304 result.app_fee_amount += app_fee_amount
306 self.status = PaymentStatus.COMPLETE
308 def update_with_intent(self, intent: PaymentIntentDTO) -> None:
309 payment_method_id = intent.payment_method_id
310 if payment_method_id is not None:
311 self.payment_method_id = payment_method_id
313 capture_before = intent.capture_before
314 if capture_before is not None:
315 epoch = TimeUtil.dt_to_simple_epoch(capture_before)
316 self.capture_before = epoch
318 intent_status = intent.intent_status
319 if intent_status != self.intent_status:
320 self.intent_status = intent_status
322 amount_captured = intent.amount_captured
323 if amount_captured is not None and amount_captured != self.amount_captured:
324 self.result_total.amount_captured += amount_captured
326 app_fee_amount = intent.app_fee_amount
327 if app_fee_amount is not None and app_fee_amount != self.app_fee_amount:
328 self.result_total.app_fee_amount += app_fee_amount
330 stripe_fee_amount = intent.stripe_fee_amount
331 if stripe_fee_amount is not None and stripe_fee_amount != self.stripe_fee_amount:
332 self.result_total.stripe_fee_amount += stripe_fee_amount
335PAYMENT_FIELD_NAMES: list[str] = list(PaymentModel.model_fields.keys())
338class PaymentDict(TypedDict, total=False):
339 created_at: Sentinel | datetime | str
340 updated_at: Sentinel | datetime | str
341 customer_info: CustomerInfoDict
342 account_info: AccountInfoDict
343 audit_info: AuditInfoDict
344 risk_assessment: RiskAssessmentDict
345 status: PaymentStatus | None
346 intent_status: StripeIntentStatus | None
347 payment_intent_id: str
348 payment_method_id: str | None
349 amount: int | None
350 last_event: PaymentEventDict | None
351 result_total: PaymentResultDict
352 capture_before: int | None
353 schedule: PaymentScheduleDict
354 stripe_dispute_id: str | None