Coverage for functions \ flipdare \ generated \ model \ payment \ payment_event_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.audit_info_model import AuditInfoModel, AuditInfoDict 

31from flipdare.generated.shared.stripe.stripe_intent_status import StripeIntentStatus 

32from flipdare.generated.shared.payment.payment_event_status import PaymentEventStatus 

33from flipdare.generated.model.payment.payment_result_model import ( 

34 PaymentResultModel, 

35 PaymentResultDict, 

36) 

37from flipdare.generated.shared.stripe.stripe_refund_reason import StripeRefundReason 

38from typing import Literal 

39 

40 

41class PaymentEventKeys(StrEnum): 

42 ID = "id" 

43 UPDATED_AT = "updated_at" 

44 CREATED_AT = "created_at" 

45 AUDIT_INFO = "audit_info" 

46 STRIPE_CHARGE_ID = "stripe_charge_id" 

47 STRIPE_REFUND_ID = "stripe_refund_id" 

48 STRIPE_TRANSFER_ID = "stripe_transfer_id" 

49 INTENT_STATUS = "intent_status" 

50 STATUS = "status" 

51 RESULT = "result" 

52 STRIPE_ERROR_CODE = "stripe_error_code" 

53 REFUND_REASON = "refund_reason" 

54 ERROR_MESSAGE = "error_message" 

55 ACTION_MESSAGE = "action_message" 

56 

57 

58# !! IMPORTANT !! 

59# !! 

60# !! this should only be used in the database to query. 

61# !! 

62class PaymentEventInternalKeys(StrEnum): 

63 UPDATED_AT = "updated_at" 

64 CREATED_AT = "created_at" 

65 

66 

67class PaymentEventModel(AppBaseModel): 

68 """Internal Stripe Payment Event Schema""" 

69 

70 model_config = ConfigDict(populate_by_name=True, arbitrary_types_allowed=True) 

71 

72 id: str | None = Field(None, alias="id") 

73 updated_at: FirestoreField = Field( 

74 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp()) 

75 ) 

76 created_at: FirestoreField = Field( 

77 default_factory=cast("Any", lambda: FirestoreTime.server_timestamp()) 

78 ) 

79 audit_info: AuditInfoModel 

80 # The ID of the Stripe charge associated with this event. 

81 stripe_charge_id: str | None = None 

82 # The ID of the Stripe refund associated with this event, if applicable. 

83 stripe_refund_id: str | None = None 

84 # The ID of the Stripe transfer associated with this event, if applicable. 

85 stripe_transfer_id: str | None = None 

86 intent_status: StripeIntentStatus 

87 status: PaymentEventStatus 

88 result: PaymentResultModel 

89 stripe_error_code: str | None = None 

90 # Messages related to the charge event, such as errors or actions taken. 

91 refund_reason: StripeRefundReason | None = None 

92 error_message: str | None = None 

93 action_message: str | None = None 

94 

95 @classmethod 

96 def validate_partial(cls, **data: Unpack[PaymentEventDict]) -> dict[str, Any]: 

97 """ 

98 Uses Unpack to give you autocomplete and static warnings 

99 if you pass an invalid key or type in your code. 

100 

101 Returns a dict with Firestore field names (aliases) for use with batch.update(). 

102 """ 

103 result: dict[str, Any] = {} 

104 for k, v in data.items(): 

105 if k in cls.__pydantic_fields__: 

106 field_info = cls.__pydantic_fields__[k] 

107 validated_value = cast( 

108 "Any", 

109 TypeAdapter(field_info.annotation).validate_python(v), 

110 ) 

111 # Use alias if defined, otherwise use field name 

112 output_key = field_info.alias or k 

113 result[output_key] = validated_value 

114 return result 

115 

116 # ---- Convenience factories ----------------------------------------- 

117 

118 @classmethod 

119 def success( 

120 cls, 

121 audit_info: AuditInfoModel, 

122 stripe_id: str, 

123 intent_status: StripeIntentStatus, 

124 result: PaymentResultModel, 

125 success_type: Literal["captured", "transferred", "refunded"] = "captured", 

126 ) -> PaymentEventModel: 

127 status: PaymentEventStatus 

128 stripe_charge_id: str | None = None 

129 stripe_refund_id: str | None = None 

130 stripe_transfer_id: str | None = None 

131 

132 match success_type: 

133 case "captured": 

134 status = PaymentEventStatus.CAPTURED 

135 stripe_charge_id = stripe_id 

136 case "transferred": 

137 status = PaymentEventStatus.TRANSFERRED 

138 stripe_transfer_id = stripe_id 

139 case "refunded": 

140 status = PaymentEventStatus.REFUNDED 

141 stripe_refund_id = stripe_id 

142 

143 return cls( 

144 id=None, 

145 audit_info=audit_info, 

146 stripe_charge_id=stripe_charge_id, 

147 stripe_refund_id=stripe_refund_id, 

148 stripe_transfer_id=stripe_transfer_id, 

149 intent_status=intent_status, 

150 status=status, 

151 result=result, 

152 ) 

153 

154 @classmethod 

155 def error( 

156 cls, 

157 audit_info: AuditInfoModel, 

158 result: PaymentResultModel, 

159 error_message: str, 

160 stripe_error_code: str | None = None, 

161 stripe_charge_id: str | None = None, 

162 action_message: str | None = None, 

163 status: PaymentEventStatus = PaymentEventStatus.ERROR, 

164 intent_status: StripeIntentStatus = StripeIntentStatus.ERROR, 

165 ) -> PaymentEventModel: 

166 """Factory method for creating an error charge event.""" 

167 return cls( 

168 id=None, 

169 audit_info=audit_info, 

170 stripe_charge_id=stripe_charge_id, 

171 intent_status=intent_status, 

172 status=status, 

173 stripe_error_code=stripe_error_code, 

174 error_message=error_message, 

175 action_message=action_message, 

176 result=result, 

177 ) 

178 

179 

180PAYMENTEVENT_FIELD_NAMES: list[str] = list(PaymentEventModel.model_fields.keys()) 

181 

182 

183class PaymentEventDict(TypedDict, total=False): 

184 id: str | None 

185 updated_at: Sentinel | datetime | str 

186 created_at: Sentinel | datetime | str 

187 audit_info: AuditInfoDict 

188 stripe_charge_id: str | None 

189 stripe_refund_id: str | None 

190 stripe_transfer_id: str | None 

191 intent_status: StripeIntentStatus 

192 status: PaymentEventStatus 

193 result: PaymentResultDict 

194 stripe_error_code: str | None 

195 refund_reason: StripeRefundReason | None 

196 error_message: str | None 

197 action_message: str | None