Coverage for functions \ flipdare \ app_service.py: 68%

117 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-05-08 12:22 +1000

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 

15import stripe 

16from typing import Any, override 

17from firebase_admin import App, credentials, firestore, get_app, initialize_app 

18from google.cloud.firestore import Client as FirestoreClient 

19from firebase_admin._auth_client import Client as AuthClient 

20from flipdare.constants import IS_DEBUG 

21from flipdare.core.singleton import Singleton 

22from flipdare.app_env import get_app_environment 

23from flipdare.app_log import LOG 

24from flipdare.generated.shared.app_error_code import AppErrorCode 

25from flipdare.error.app_error import ServerError 

26 

27from flipdare.manager.service_manager import ServiceManager 

28from flipdare.manager.db_manager import DbManager 

29from flipdare.manager.search_manager import SearchManager 

30from flipdare.manager.backend_manager import BackendManager 

31from flipdare.manager.task_manager import TaskManager 

32 

33__all__ = [ 

34 "AppService", 

35 "get_app_service", 

36] 

37 

38 

39def get_app_service() -> AppService: 

40 return AppService.instance() 

41 

42 

43# ----------------------------------------------------------------------------- 

44# Service singleton 

45# ----------------------------------------------------------------------------- 

46 

47# NOTE: we need to create everything here so that any config 

48# NOTE: is set beforehand. 

49 

50 

51class AppService(Singleton): 

52 

53 def __init__( 

54 self, 

55 auth: AuthClient | None = None, 

56 database_client: FirestoreClient | None = None, 

57 db_manager: DbManager | None = None, 

58 service_manager: ServiceManager | None = None, 

59 search_manager: SearchManager | None = None, 

60 backend_manager: BackendManager | None = None, 

61 task_manager: TaskManager | None = None, 

62 *args: Any, 

63 **kwargs: Any, 

64 ) -> None: 

65 

66 super().__init__(*args, **kwargs) 

67 

68 if not get_app_environment().in_cloud: 

69 LOG().warning("Skipping Firebase initialization...") 

70 else: 

71 AppService.setup() 

72 

73 self._auth = auth 

74 self._database_client = database_client 

75 

76 # managers 

77 self._db_manager = db_manager 

78 self._service_manager = service_manager 

79 self._search_manager = search_manager 

80 self._backend_manager = backend_manager 

81 self._task_manager = task_manager 

82 

83 @staticmethod 

84 def setup() -> None: 

85 from flipdare.app_config import get_app_config 

86 

87 LOG().info("Initializing Firebase AppService...") 

88 config = get_app_config() 

89 try: 

90 config.validate() 

91 except Exception as e: 

92 LOG().error(f"Configuration validation failed: {e}") 

93 raise ServerError( 

94 message=f"Configuration error: {e}", 

95 error_code=AppErrorCode.SERVER_CONFIG, 

96 ) from e 

97 

98 stripe.api_key = config.stripe_secret_key 

99 

100 # Check if Firebase is already initialized 

101 try: 

102 get_app() # This will raise ValueError if no app exists 

103 if IS_DEBUG: 

104 LOG().debug("Firebase app already initialized, skipping initialization") 

105 except ValueError: 

106 # No app exists, initialize it 

107 cred = credentials.Certificate(config.credential) 

108 initialize_app(cred) 

109 LOG().info("Firebase app initialized successfully") 

110 

111 # 

112 # misc 

113 # 

114 

115 @property 

116 def auth(self) -> AuthClient: 

117 if self._auth is None: 

118 app: App = get_app() 

119 self._auth = AuthClient(app) # type: ignore 

120 return self._auth 

121 

122 @auth.setter 

123 def auth(self, value: AuthClient) -> None: 

124 self._auth = value 

125 

126 # 

127 # managers 

128 # 

129 

130 @property 

131 def backend_manager(self) -> BackendManager: 

132 if self._backend_manager is None: 

133 self._backend_manager = BackendManager.instance() 

134 return self._backend_manager 

135 

136 @property 

137 def service_manager(self) -> ServiceManager: 

138 if self._service_manager is None: 

139 self._service_manager = ServiceManager.instance() 

140 return self._service_manager 

141 

142 @service_manager.setter 

143 def service_manager(self, value: ServiceManager) -> None: 

144 self._service_manager = value 

145 

146 @property 

147 def search_manager(self) -> SearchManager: 

148 if self._search_manager is None: 

149 self._search_manager = SearchManager.instance() 

150 return self._search_manager 

151 

152 @search_manager.setter 

153 def search_manager(self, value: SearchManager) -> None: 

154 self._search_manager = value 

155 

156 @property 

157 def task_manager(self) -> TaskManager: 

158 if self._task_manager is None: 

159 self._task_manager = TaskManager.instance() 

160 return self._task_manager 

161 

162 # 

163 # database 

164 # 

165 @property 

166 def firestore_client(self) -> FirestoreClient: 

167 if self._database_client is None: 

168 self._database_client = firestore.client() 

169 return self._database_client 

170 

171 @firestore_client.setter 

172 def firestore_client(self, value: FirestoreClient) -> None: 

173 self._database_client = value 

174 

175 @property 

176 def db_manager(self) -> DbManager: 

177 if self._db_manager is not None: 

178 return self._db_manager 

179 

180 database = self.firestore_client 

181 db_manager = DbManager.instance(database) 

182 self._db_manager = db_manager 

183 return self._db_manager 

184 

185 @db_manager.setter 

186 def db_manager(self, value: DbManager) -> None: 

187 from flipdare.app_env import get_app_environment 

188 

189 if get_app_environment().in_cloud: 

190 raise RuntimeError("DBManager cannot be overridden in production.") 

191 

192 self._db_manager = value 

193 

194 @override 

195 def __eq__(self, other: object) -> bool: 

196 if not isinstance(other, AppService): 

197 return NotImplemented 

198 return ( 

199 self.auth == other.auth 

200 and self.firestore_client == other.firestore_client 

201 and self.db_manager == other.db_manager 

202 and self.search_manager == other.search_manager 

203 and self.service_manager == other.service_manager 

204 and self.backend_manager == other.backend_manager 

205 ) 

206 

207 @override 

208 def __hash__(self) -> int: 

209 return hash( 

210 ( 

211 self.auth, 

212 self.firestore_client, 

213 self.db_manager, 

214 self.search_manager, 

215 self.service_manager, 

216 self.backend_manager, 

217 ), 

218 )