Coverage for functions \ flipdare \ firestore \ core \ pledge_event_transaction.py: 49%

1from __future__ import annotations 

2 

3from typing import Any 

4 

5from google.cloud.firestore import DocumentReference, Increment 

6 

7from flipdare.generated.model.payment.payment_event_model import PaymentEventModel 

8from flipdare.generated.model.payment.payment_model import PaymentKeys 

9from flipdare.generated.model.payment.payment_result_model import PaymentResultModel 

10from flipdare.generated.model.payment.pledge_model import PledgeKeys 

11from flipdare.generated.shared.firestore_collections import FirestoreCollections 

12from flipdare.generated.shared.payment.payment_status import PaymentStatus 

13from flipdare.util.time_util import FirestoreTime 

14from flipdare.wrapper.payment.payment_event_wrapper import PaymentEventWrapper 

15from flipdare.wrapper.payment.pledge_wrapper import PledgeWrapper 

16 

17_PK = PledgeKeys 

18_MK = PaymentKeys 

19_PLEDGE = FirestoreCollections.PLEDGE.value 

20_EVENTS = FirestoreCollections.PLEDGE_PAYMENT_EVENTS.value 

21 

22 

23class PledgeEventTransaction: 

24 

25 def __init__(self, parent_db: Any) -> None: 

26 self.parent_db = parent_db 

27 self.client = parent_db.client 

28 

29 def add_event( 

30 self, 

31 pledge: PledgeWrapper, 

32 event: PaymentEventModel, 

33 payment_status: PaymentStatus | None = None, 

34 result: PaymentResultModel | None = None, 

35 ) -> PaymentEventWrapper: 

36 """ 

37 Atomically add a payment event and update the pledge in a single batch. 

38 

39 Uses dot-notation field paths to update nested pledge.payment fields 

40 without rewriting the entire payment object. 

41 """ 

42 pledge_id = pledge.doc_id 

43 batch = self.client.batch() 

44 

45 # 1. Create event document in subcollection 

46 pledge_ref = self.client.collection(_PLEDGE).document(pledge_id) 

47 event_ref: DocumentReference = pledge_ref.collection(_EVENTS).document() 

48 event_payload = event.to_dict() 

49 batch.set(event_ref, event_payload) 

50 

51 # 2. update payment_event_count 

52 # 3. update last_event 

53 # 4. update payment status if provided 

54 # 5. update payment result total if provided 

55 pledge_updates: dict[str, Any] = { 

56 _PK.PAYMENT_EVENT_CT: Increment(1), 

57 f"{_PK.PAYMENT}.{_MK.LAST_EVENT}": event_payload, 

58 f"{_PK.PAYMENT}.{_MK.UPDATED_AT}": FirestoreTime.server_timestamp(), 

59 } 

60 if payment_status is not None: 

61 pledge_updates[f"{_PK.PAYMENT}.{_MK.STATUS}"] = payment_status 

62 

63 if result is not None: 

64 if pledge._model.payment is None: 

65 raise ValueError("Cannot update payment result when payment is None") 

66 

67 existing_result = pledge._model.payment.result_total 

68 existing_result.accumulate(result) 

69 

70 pledge_updates[f"{_PK.PAYMENT}.{_MK.RESULT_TOTAL}"] = existing_result.to_dict() 

71 

72 batch.update(pledge_ref, pledge_updates) 

73 batch.commit() 

74 

75 return PaymentEventWrapper.from_dict({**event_payload, "id": event_ref.id})