Coverage for functions \ flipdare \ search \ doc \ general_document.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 

13from __future__ import annotations 

14 

15from typing import Any, cast, override 

16 

17from flipdare.app_types import SearchDict 

18from flipdare.generated.schema.search.general_document_schema import ( 

19 GeneralDocumentKey, 

20 GeneralDocumentSchema, 

21) 

22from flipdare.generated.shared.model.token_score import TokenScore 

23from flipdare.generated.shared.search.search_obj_type import SearchObjType 

24from flipdare.search.doc._search_document import TS_C, SearchDocument 

25from flipdare.util.time_util import TypesenseTime 

26 

27__all__ = ["GeneralDocument"] 

28 

29 

30class GeneralDocument(SearchDocument[GeneralDocumentSchema]): 

31 

32 SCHEMA_CLASS = GeneralDocumentSchema 

33 

34 def __init__(self, data: GeneralDocumentSchema, doc_id: str | None = None) -> None: 

35 super().__init__(data=data, doc_id=doc_id) 

36 

37 @classmethod 

38 def create( 

39 cls, 

40 keywords: list[str], 

41 tags: list[str], 

42 obj_id: str, 

43 uid: str, 

44 obj_type: SearchObjType, 

45 updated_at: TS_C, 

46 created_at: TS_C, 

47 group_id: str | None = None, 

48 score: float = 1.0, 

49 tag_score: float = TokenScore.LOW.score, 

50 views: int = 0, 

51 ) -> GeneralDocument: 

52 

53 data: GeneralDocumentSchema = { 

54 "updated_at": TypesenseTime.from_any(updated_at), 

55 "created_at": TypesenseTime.from_any(created_at), 

56 "score": score, 

57 "uid": uid, 

58 "obj_id": obj_id, 

59 "obj_type": obj_type, 

60 "keywords": keywords, 

61 "views": views, 

62 "tags": tags, 

63 "tag_score": tag_score, 

64 } 

65 

66 if group_id is not None: 

67 data["group_id"] = group_id 

68 

69 return cls(data=data) 

70 

71 @classmethod 

72 @override 

73 def from_payload( 

74 cls, 

75 doc_id: str, 

76 payload: SearchDict, 

77 original: SearchDocument[Any] | None = None, 

78 ) -> GeneralDocument: 

79 schema_data = cls.parse(cls.SCHEMA_CLASS, payload, original) 

80 return cls(data=schema_data, doc_id=doc_id) 

81 

82 @property 

83 @override 

84 def uid(self) -> str: 

85 return self._data["uid"] 

86 

87 @property 

88 @override 

89 def created_at(self) -> int: 

90 return self._data["created_at"] 

91 

92 @property 

93 @override 

94 def updated_at(self) -> int: 

95 return self._data["updated_at"] 

96 

97 @property 

98 def obj_id(self) -> str: 

99 return self._data["obj_id"] 

100 

101 @property 

102 @override 

103 def obj_type(self) -> SearchObjType: 

104 return self._data["obj_type"] 

105 

106 @property 

107 def keywords(self) -> str | list[str]: 

108 return self._data["keywords"] 

109 

110 @property 

111 def views(self) -> int: 

112 return self._data["views"] 

113 

114 @property 

115 def group_id(self) -> str | None: 

116 return self._data.get("group_id") 

117 

118 @property 

119 def tags(self) -> str | list[str]: 

120 return self._data["tags"] 

121 

122 @property 

123 def score(self) -> float: 

124 return self._data["score"] 

125 

126 @property 

127 def tag_score(self) -> float: 

128 return self._data["tag_score"] 

129 

130 def update_views(self, views: int, score: float, updated_at: int) -> None: 

131 values = { 

132 GeneralDocumentKey.VIEWS: views, 

133 GeneralDocumentKey.SCORE: score, 

134 } 

135 self.update_payload( 

136 payload_values=cast("SearchDict", values), 

137 updated_at=updated_at, 

138 ) 

139 

140 def update_keywords( 

141 self, 

142 keywords: list[str], 

143 score: float, 

144 updated_at: int, 

145 tags: list[str] | None = None, 

146 tag_score: float | None = None, 

147 ) -> None: 

148 data = { 

149 GeneralDocumentKey.KEYWORDS: keywords, 

150 GeneralDocumentKey.SCORE: score, 

151 } 

152 if tags is not None: 

153 data[GeneralDocumentKey.TAGS] = tags 

154 if tag_score is not None: 

155 data[GeneralDocumentKey.TAG_SCORE] = tag_score 

156 

157 self.update_payload( 

158 payload_values=cast("SearchDict", data), 

159 updated_at=updated_at, 

160 ) 

161 

162 @override 

163 def __str__(self) -> str: 

164 return ( 

165 f"GeneralDocument(doc_id={self.doc_id}, uid={self.uid}, " 

166 f"obj_id={self.obj_id}, obj_type={self.obj_type})" 

167 ) 

168 

169 @override 

170 def __repr__(self) -> str: 

171 return self.__str__()