Coverage for functions \ flipdare \ firestore \ context \ group_context.py: 63%

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 

13from __future__ import annotations 

14 

15from typing import TYPE_CHECKING, Any, override 

16 

17from flipdare.app_log import LOG 

18from flipdare.app_types import DatabaseDict 

19from flipdare.constants import MAX_SEARCH_GROUP_MEMBERS 

20from flipdare.firestore import GroupDb 

21from flipdare.firestore.context._model_context import ModelContext 

22from flipdare.firestore.context._model_context_factory import ModelContextFactory 

23from flipdare.generated.model.group_model import GroupModel 

24from flipdare.wrapper import ( 

25 GroupMemberWrapper, 

26 GroupWrapper, 

27 UserWrapper, 

28) 

29 

30if TYPE_CHECKING: 

31 from flipdare.manager.db_manager import DbManager 

32 

33__all__ = ["GroupContextFactory", "GroupContext"] 

34 

35 

36class GroupContextFactory(ModelContextFactory[GroupWrapper, "GroupContext"]): 

37 

38 def __init__( 

39 self, db_manager: DbManager | None = None, limit: int = MAX_SEARCH_GROUP_MEMBERS 

40 ) -> None: 

41 super().__init__(db_manager=db_manager) 

42 self._limit = limit 

43 

44 @property 

45 def limit(self) -> int: 

46 return self._limit 

47 

48 @override 

49 def create(self, obj: Any) -> GroupContext | None: 

50 if isinstance(obj, GroupContext): 

51 return obj 

52 if isinstance(obj, GroupWrapper): 

53 return self._from_model(obj) 

54 if isinstance(obj, str): 

55 return self._from_id(obj) 

56 obj_data: DatabaseDict = obj # set explicity type for type checkers. 

57 return self._from_data(obj_data) 

58 

59 @override 

60 def _from_id(self, doc_id: str) -> GroupContext | None: 

61 group_db = self.group_db 

62 group: GroupWrapper | None = None 

63 try: 

64 group = group_db.get(doc_id) 

65 if group is None: 

66 # need at least the group 

67 LOG().error(f"Group {doc_id} not found in db.") 

68 return None 

69 

70 owner: UserWrapper | None = self.get_user(group.model.uid) 

71 return GroupContext(group=group, group_db=group_db, owner=owner, limit=self.limit) 

72 except Exception: 

73 LOG().error(f"Error retrieving group {doc_id} from db.") 

74 return None 

75 

76 @override 

77 def _from_data(self, data: DatabaseDict) -> GroupContext | None: 

78 try: 

79 group = GroupWrapper.from_dict(data) 

80 return self._from_model(group) 

81 except Exception: 

82 return None 

83 

84 @override 

85 def _from_model(self, model: GroupWrapper | GroupModel) -> GroupContext | None: 

86 owner = self.get_user(model.uid) # Returns UserWrapper | None 

87 

88 if isinstance(model, GroupModel): 

89 model = GroupWrapper.from_model(model) 

90 

91 return GroupContext( 

92 group=model, 

93 group_db=self.group_db, 

94 owner=owner, 

95 limit=self.limit, 

96 ) 

97 

98 

99class GroupContext(ModelContext): 

100 def __init__( 

101 self, 

102 group: GroupWrapper, 

103 owner: UserWrapper | None = None, 

104 group_db: GroupDb | None = None, 

105 members: list[GroupMemberWrapper] | None = None, 

106 limit: int = MAX_SEARCH_GROUP_MEMBERS, 

107 ) -> None: 

108 self._group = group 

109 self._owner = owner 

110 self._group_db = group_db 

111 self._limit = limit 

112 self._members = members 

113 # Call super().__init__() LAST - it calls validate() 

114 super().__init__() 

115 

116 @property 

117 def group_db(self) -> GroupDb: 

118 

119 from flipdare.services import get_group_db 

120 

121 if self._group_db is None: 

122 self._group_db = get_group_db() 

123 return self._group_db 

124 

125 @property 

126 def group(self) -> GroupWrapper: 

127 self._require_valid("access group") 

128 return self._group 

129 

130 @property 

131 def owner(self) -> UserWrapper: 

132 self._require_valid("access owner") 

133 return self._owner # type: ignore 

134 

135 @property 

136 def group_id(self) -> str: 

137 self._require_valid("access group_id") 

138 assert self._group is not None # for mypy 

139 return self._group.doc_id 

140 

141 def members(self) -> list[GroupMemberWrapper] | None: 

142 if not self.validate(): 

143 raise ValueError("Cannot get members from invalid GroupContext.") 

144 

145 if self._members is not None: 

146 return self._members 

147 

148 try: 

149 members = self.group_db.get_members(group_id=self.group_id, limit=self._limit) 

150 self._members = members 

151 return self._members 

152 except Exception: 

153 LOG().error(f"Error retrieving members for group {self.group_id} from db.") 

154 return None 

155 

156 def users(self) -> list[UserWrapper] | None: 

157 if not self.validate(): 

158 raise ValueError("Cannot get users from invalid GroupContext.") 

159 

160 try: 

161 return self.group_db.get_users(group_id=self.group_id, limit=self._limit) 

162 except Exception: 

163 LOG().error(f"Error retrieving users for group {self.group_id} from db.") 

164 return None 

165 

166 def __len__(self) -> int: 

167 """Return number of members in the group.""" 

168 members = self.members() 

169 if members is None: 

170 return 0 

171 return len(members) 

172 

173 def __getitem__(self, index: int) -> GroupMemberWrapper | None: 

174 """Get member at given index.""" 

175 members = self.members() 

176 if members is None: 

177 return None 

178 if index < 0 or index >= len(members): 

179 return None 

180 return members[index] 

181 

182 @property 

183 @override 

184 def doc_id(self) -> str: 

185 return self.group.doc_id 

186 

187 @override 

188 def validate(self) -> bool: 

189 """Validate that all required models exist and have doc_ids.""" 

190 return self._is_model_valid(self._group) and self._is_model_valid(self._owner) 

191 

192 @property 

193 @override 

194 def _error_messages(self) -> list[str]: 

195 """Build list of validation errors.""" 

196 errors: list[str] = [] 

197 if err := self._validate_model(self._group, "group"): 

198 errors.append(err) 

199 if err := self._validate_model(self._owner, "owner"): 

200 errors.append(err) 

201 return errors