Coverage for functions \ flipdare \ firestore \ _app_sub_db.py: 77%

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 typing import Any 

15 

16from google.cloud.firestore import Client as FirestoreClient 

17from google.cloud.firestore_v1.base_document import DocumentSnapshot 

18from flipdare.app_types import DatabaseDict 

19from flipdare.constants import DEF_RETRIEVAL_WINDOW_HOURS 

20from flipdare.firestore._app_db import AppDb 

21from flipdare.firestore.core.app_base_model import AppBaseModel 

22from flipdare.firestore.core.db_query import OrderByField 

23from flipdare.generated.shared.firestore_collections import FirestoreCollections 

24from flipdare.wrapper._persisted_wrapper import PersistedWrapper 

25 

26__all__ = ["AppSubDb"] 

27 

28 

29class AppSubDb[W: PersistedWrapper[Any], T: AppBaseModel](AppDb[W, T]): 

30 def __init__( 

31 self, 

32 client: FirestoreClient, 

33 collection_name: FirestoreCollections, 

34 sub_collection_name: FirestoreCollections, 

35 wrapper_class: type[W], # The Sub Wrapper class e.g. CommentWrapper 

36 model_class: type[T], # The Sub Model class e.g. CommentModel 

37 def_window_hours: int = DEF_RETRIEVAL_WINDOW_HOURS, 

38 ) -> None: 

39 super().__init__( 

40 client=client, 

41 collection_name=collection_name, 

42 wrapper_class=wrapper_class, 

43 model_class=model_class, 

44 def_window_hours=def_window_hours, 

45 ) 

46 self._sub_collection_name = sub_collection_name 

47 

48 @property 

49 def sub_collection_name(self) -> str: 

50 """Get the name of the sub-collection.""" 

51 return self._sub_collection_name.value 

52 

53 def exists_sub(self, parent_id: str, sub_id: str) -> bool: 

54 """Check if a sub-collection document exists.""" 

55 return self._exists_sub(parent_id, self._sub_collection_name, sub_id) 

56 

57 def get_sub(self, parent_id: str, sub_id: str) -> W | None: 

58 """Get a sub-collection document by ID, returns PersistedWrapper.""" 

59 data = self._get_sub(parent_id, self._sub_collection_name, sub_id) 

60 if data is None: 

61 return None 

62 

63 return self.wrapper_class.from_dict(data) 

64 

65 def get_all_sub( 

66 self, 

67 parent_id: str, 

68 limit: int | None = None, 

69 order_by: OrderByField[Any] | None = None, 

70 ) -> list[W]: 

71 """Get all documents in a sub-collection, returns list of PersistedWrapper.""" 

72 data_items = self._get_all_sub( 

73 parent_id, 

74 self._sub_collection_name, 

75 order_by=order_by, 

76 limit=limit, 

77 ) 

78 if not data_items: 

79 return [] 

80 

81 results: list[W] = [] 

82 for item in data_items: 

83 model = self.wrapper_class.from_dict(item) 

84 results.append(model) 

85 return results 

86 

87 def get_bulk_sub( 

88 self, 

89 parent_id: str, 

90 sub_ids: list[str], 

91 ) -> list[W]: 

92 """Get multiple sub-collection documents by their IDs, returns list of PersistedWrapper.""" 

93 data_items = self._get_bulk_sub(parent_id, self._sub_collection_name, sub_ids) 

94 if not data_items: 

95 return [] 

96 

97 results: list[W] = [] 

98 for item in data_items: 

99 model = self.wrapper_class.from_dict(item) 

100 results.append(model) 

101 return results 

102 

103 def create_sub(self, parent_id: str, data: T | dict[str, Any]) -> W: 

104 """ 

105 Create a new document in the sub-collection, returns PersistedWrapper. 

106 

107 Args: 

108 parent_id: Parent document ID 

109 data: Model instance or dict to create from 

110 

111 Returns: 

112 PersistedWrapper for the created document 

113 

114 """ 

115 # Convert to dict for Firestore 

116 payload = data if isinstance(data, dict) else data.to_dict() 

117 created_data = self._create_sub(parent_id, self._sub_collection_name, payload) 

118 

119 return self.wrapper_class.from_dict(created_data) 

120 

121 def update_sub( 

122 self, 

123 parent_id: str, 

124 sub_id: str, 

125 updates: DatabaseDict, 

126 ) -> W | None: 

127 """Update a sub-collection document, returns PersistedWrapper.""" 

128 updated_data = self._update_sub(parent_id, self._sub_collection_name, sub_id, updates) 

129 

130 if updated_data is None: 

131 return None 

132 

133 return self.wrapper_class.from_dict(updated_data) 

134 

135 def _cvt_sub_snap_to_model(self, snap: DocumentSnapshot) -> W | None: 

136 """Convert DocumentSnapshot to PersistedWrapper[model].""" 

137 data = self._cvt_snap_to_data(snap) 

138 if data is None: 

139 return None 

140 

141 return self.wrapper_class.from_dict(data)