Coverage for functions \ flipdare \ firestore \ restriction_db.py: 74%

1#!/usr/bin/env python 

2 

3 

4from google.cloud.firestore import Client as FirestoreClient 

5from flipdare.app_log import LOG 

6from flipdare.constants import IS_DEBUG 

7from flipdare.firestore._app_db import AppDb 

8from flipdare.firestore.core.db_query import DbQuery, FieldOp, WhereField 

9from flipdare.generated import ( 

10 RestrictionCategory, 

11 StopwatchDuration, 

12 RestrictionKeys, 

13 RestrictionStatus, 

14) 

15from flipdare.generated.model.issue.restriction_model import ( 

16 RestrictionInternalKeys, 

17 RestrictionModel, 

18) 

19from flipdare.generated.shared.firestore_collections import FirestoreCollections 

20from flipdare.util.time_util import TimeUtil 

21from flipdare.wrapper import RestrictionWrapper 

22 

23_RESTRICTION: str = FirestoreCollections.RESTRICTION.value 

24 

25__all__ = ["RestrictionDb"] 

26 

27_K = RestrictionKeys 

28_OP = FieldOp 

29_I = RestrictionInternalKeys 

30 

31 

32class RestrictionDb(AppDb[RestrictionWrapper, RestrictionModel]): 

33 """Class for managing flag action-related database operations.""" 

34 

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

36 super().__init__( 

37 client=client, 

38 collection_name=FirestoreCollections.RESTRICTION, 

39 model_class=RestrictionModel, 

40 wrapper_class=RestrictionWrapper, 

41 ) 

42 

43 def get_inactive_actions(self) -> list[RestrictionWrapper]: 

44 """Get all inactive flag actions from Firestore.""" 

45 query = DbQuery.and_( 

46 [ 

47 WhereField[_K](_K.STOPWATCH, _OP.EQUAL, None), 

48 WhereField[_K](_K.STATUS, _OP.EQUAL, RestrictionStatus.PENDING.value), 

49 ], 

50 ) 

51 

52 results = query.get_query(self.client, _RESTRICTION).get() 

53 if len(results) == 0: 

54 return [] 

55 

56 entries = [ 

57 action for doc in results if (action := self._cvt_snap_to_model(doc)) is not None 

58 ] 

59 if IS_DEBUG: 

60 LOG().debug(f"Retrieved {len(entries)} inactive actions.") 

61 return entries 

62 

63 def get_expired_active_actions(self) -> list[RestrictionWrapper]: 

64 """CompleteAt < now - get all completed flag actions from Firestore.""" 

65 now = TimeUtil.get_current_utc_dt() 

66 query = DbQuery.and_( 

67 [ 

68 WhereField[_K](_K.STOPWATCH, _OP.LESS_THAN_OR_EQUAL, now), 

69 WhereField[_K](_K.STATUS, _OP.NOT_EQUAL, RestrictionStatus.PENDING.value), 

70 ], 

71 ) 

72 

73 results = query.get_query(self.client, _RESTRICTION).get() 

74 if len(results) == 0: 

75 return [] 

76 

77 entries = [ 

78 action for doc in results if (action := self._cvt_snap_to_model(doc)) is not None 

79 ] 

80 if IS_DEBUG: 

81 LOG().debug(f"Retrieved {len(entries)} expired active actions as of {now.isoformat()}") 

82 return entries 

83 

84 def get_auto_permanent(self) -> list[RestrictionWrapper]: 

85 """Get all auto restrictions that are permanent, so they can be reviewed.""" 

86 query = DbQuery.and_( 

87 [ 

88 WhereField[_K](_K.STOPWATCH, _OP.NOT_EQUAL, None), 

89 WhereField[_K](_K.CATEGORY, _OP.EQUAL, RestrictionCategory.AUTO.value), 

90 WhereField[_K](_K.DURATION, _OP.EQUAL, StopwatchDuration.PERMANENT.value), 

91 ], 

92 ) 

93 

94 results = query.get_query(self.client, _RESTRICTION).get() 

95 if len(results) == 0: 

96 return [] 

97 

98 entries = [ 

99 action for doc in results if (action := self._cvt_snap_to_model(doc)) is not None 

100 ] 

101 if IS_DEBUG: 

102 msg = f"Retrieved {len(entries)} permanent restrictions auto generated (not reviewed)." 

103 LOG().debug(msg) 

104 return entries 

105 

106 def get_auto_not_permanent(self) -> list[RestrictionWrapper]: 

107 """Get all auto restrictions that are not permanent, so they can be reviewed.""" 

108 query = DbQuery.and_( 

109 [ 

110 WhereField[_K](_K.STOPWATCH, _OP.NOT_EQUAL, None), 

111 WhereField[_K](_K.CATEGORY, _OP.EQUAL, RestrictionCategory.AUTO.value), 

112 WhereField[_K](_K.DURATION, _OP.NOT_EQUAL, StopwatchDuration.PERMANENT.value), 

113 ], 

114 ) 

115 

116 results = query.get_query(self.client, _RESTRICTION).get() 

117 if len(results) == 0: 

118 return [] 

119 

120 entries = [ 

121 action for doc in results if (action := self._cvt_snap_to_model(doc)) is not None 

122 ] 

123 if IS_DEBUG: 

124 msg = f"Retrieved {len(entries)} non-permanent restrictions auto generated (not reviewed)." 

125 LOG().debug(msg) 

126 return entries