Coverage for functions \ flipdare \ firestore \ issue_db.py: 92%

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, IssueKeys, IssueProgress 

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

23from flipdare.generated.model.issue.issue_model import IssueModel 

24from flipdare.generated.shared.firestore_collections import FirestoreCollections 

25from flipdare.util.time_util import TimeUtil 

26from flipdare.wrapper import IssueCommentWrapper, IssueWrapper 

27 

28_ISSUE: str = FirestoreCollections.ISSUE.value 

29 

30__all__ = ["IssueDb"] 

31 

32_K = IssueKeys 

33_OP = FieldOp 

34 

35 

36class IssueDb(AppDb[IssueWrapper, IssueModel]): 

37 """Class for managing issue-related database operations.""" 

38 

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

40 super().__init__( 

41 client=client, 

42 collection_name=FirestoreCollections.ISSUE, 

43 model_class=IssueModel, 

44 wrapper_class=IssueWrapper, 

45 ) 

46 

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

48 client=client, 

49 collection_name=FirestoreCollections.ISSUE, 

50 sub_collection_name=FirestoreCollections.ISSUE_COMMENTS, 

51 wrapper_class=IssueCommentWrapper, 

52 model_class=IssueCommentModel, 

53 ) 

54 

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

56 """ 

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

58 """ 

59 hours_ago = TimeUtil.get_utc_time_hours_ago(hours) 

60 debug_str = ( 

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

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

63 ) 

64 LOG().info(debug_str) 

65 

66 progresses = [IssueProgress.OPEN.value, IssueProgress.WAITING_ADMIN.value] 

67 progresses_str = list(progresses) 

68 

69 try: 

70 where_fields = [ 

71 WhereField[_K](_K.CREATED_AT, _OP.GREATER_THAN_OR_EQUAL, hours_ago), 

72 WhereField[_K](_K.PROGRESS, _OP.IN, progresses_str), 

73 ] 

74 order_by = OrderByField.created_at(descending=False) 

75 query = DbQuery.and_(where_fields, order_by=order_by) 

76 

77 results = query.get_query(self.client, _ISSUE).stream() 

78 issues = [ 

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

80 ] 

81 if IS_DEBUG: 

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

83 

84 return issues 

85 except Exception as e: 

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

87 raise DatabaseError( 

88 message=msg, 

89 error_code=AppErrorCode.DATABASE, 

90 collection_name=self.collection_name, 

91 ) from e