Coverage for functions \ flipdare \ firestore \ payment_issue_db.py: 90%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14from google.cloud.firestore import Client as FirestoreClient 

15from flipdare.app_log import LOG 

16from flipdare.constants import IS_DEBUG 

17from flipdare.error.app_error import DatabaseError 

18from flipdare.firestore._app_db import AppDb 

19from flipdare.firestore._app_sub_db import AppSubDb 

20from flipdare.firestore.core.db_query import DbQuery, FieldOp, OrderByField, WhereField 

21from flipdare.generated import AppErrorCode, IssueProgress, PaymentIssueKeys 

22from flipdare.generated.model.issue.issue_comment_model import IssueCommentModel 

23from flipdare.generated.model.issue.payment_issue_model import ( 

24 PaymentIssueInternalKeys, 

25 PaymentIssueModel, 

26) 

27from flipdare.generated.shared.firestore_collections import FirestoreCollections 

28from flipdare.util.time_util import TimeUtil 

29from flipdare.wrapper import IssueCommentWrapper, PaymentIssueWrapper 

30 

31_PAYMENT_ISSUE: str = FirestoreCollections.PAYMENT_ISSUE.value 

32 

33__all__ = ["PaymentIssueDb"] 

34 

35_K = PaymentIssueKeys 

36_OP = FieldOp 

37 

38_I = PaymentIssueInternalKeys 

39 

40type _KeyType = PaymentIssueKeys | PaymentIssueInternalKeys # for type checking. 

41 

42 

43class PaymentIssueDb(AppDb[PaymentIssueWrapper, PaymentIssueModel]): 

44 """Class for managing payment issue-related database operations.""" 

45 

46 def __init__(self, client: FirestoreClient) -> None: 

47 super().__init__( 

48 client=client, 

49 collection_name=FirestoreCollections.PAYMENT_ISSUE, 

50 model_class=PaymentIssueModel, 

51 wrapper_class=PaymentIssueWrapper, 

52 ) 

53 

54 self.comments = AppSubDb[IssueCommentWrapper, IssueCommentModel]( 

55 client=client, 

56 collection_name=FirestoreCollections.PAYMENT_ISSUE, 

57 sub_collection_name=FirestoreCollections.ISSUE_COMMENTS, 

58 wrapper_class=IssueCommentWrapper, 

59 model_class=IssueCommentModel, 

60 ) 

61 

62 def get_recent_waiting_admin(self, hours: int = 4) -> list[PaymentIssueWrapper]: 

63 """ 

64 Get issues that are waiting for admin review in the last 4 hours 

65 """ 

66 hours_ago = TimeUtil.get_utc_time_hours_ago(hours) 

67 debug_str = ( 

68 f"Getting issues waiting for admin review within last " 

69 f"{TimeUtil.formatted_dt(hours_ago)}" 

70 ) 

71 LOG().info(debug_str) 

72 

73 try: 

74 and_ = [ 

75 WhereField[_KeyType](_I.CREATED_AT, _OP.GREATER_THAN_OR_EQUAL, hours_ago), 

76 WhereField[_KeyType](_K.PROGRESS, _OP.EQUAL, IssueProgress.WAITING_ADMIN.value), 

77 ] 

78 order_by = OrderByField.created_at(descending=False) 

79 query = DbQuery.and_(and_, order_by=order_by) 

80 

81 results = query.get_query(self.client, _PAYMENT_ISSUE).stream() 

82 

83 entries = [ 

84 issue for doc in results if (issue := self._cvt_snap_to_model(doc)) is not None 

85 ] 

86 if IS_DEBUG: 

87 LOG().debug(f"Retrieved {len(entries)} payment issues waiting for admin review.") 

88 return entries 

89 except Exception as e: 

90 msg = f"Failed to get issues waiting for admin review: {e}" 

91 raise DatabaseError( 

92 message=msg, 

93 error_code=AppErrorCode.DATABASE, 

94 collection_name=self.collection_name, 

95 ) from e