Coverage for src/local_deep_research/news/core/storage.py: 71%

1""" 

2Base storage interfaces for the news system. 

3""" 

4 

5from abc import ABC, abstractmethod 

6from typing import List, Optional, Dict, Any 

7import uuid 

8 

9 

10class BaseStorage(ABC): 

11 """Abstract base class for all storage interfaces""" 

12 

13 @abstractmethod 

14 def create(self, data: Dict[str, Any]) -> str: 

15 """Create a new record and return its ID""" 

16 pass 

17 

18 @abstractmethod 

19 def get(self, id: str) -> Optional[Dict[str, Any]]: 

20 """Get a record by ID""" 

21 pass 

22 

23 @abstractmethod 

24 def update(self, id: str, data: Dict[str, Any]) -> bool: 

25 """Update a record, return True if successful""" 

26 pass 

27 

28 @abstractmethod 

29 def delete(self, id: str) -> bool: 

30 """Delete a record, return True if successful""" 

31 pass 

32 

33 @abstractmethod 

34 def list( 

35 self, 

36 filters: Optional[Dict[str, Any]] = None, 

37 limit: int = 100, 

38 offset: int = 0, 

39 ) -> List[Dict[str, Any]]: 

40 """List records with optional filtering""" 

41 pass 

42 

43 def generate_id(self) -> str: 

44 """Generate a unique ID""" 

45 return str(uuid.uuid4()) 

46 

47 

48class CardStorage(BaseStorage): 

49 """Interface for news card storage""" 

50 

51 @abstractmethod 

52 def get_by_user( 

53 self, user_id: str, limit: int = 50, offset: int = 0 

54 ) -> List[Dict[str, Any]]: 

55 """Get cards for a specific user""" 

56 pass 

57 

58 @abstractmethod 

59 def get_latest_version(self, card_id: str) -> Optional[Dict[str, Any]]: 

60 """Get the latest version of a card""" 

61 pass 

62 

63 @abstractmethod 

64 def add_version(self, card_id: str, version_data: Dict[str, Any]) -> str: 

65 """Add a new version to a card""" 

66 pass 

67 

68 @abstractmethod 

69 def update_latest_info( 

70 self, card_id: str, version_data: Dict[str, Any] 

71 ) -> bool: 

72 """Update the denormalized latest version info on the card""" 

73 pass 

74 

75 @abstractmethod 

76 def archive_card(self, card_id: str) -> bool: 

77 """Archive a card""" 

78 pass 

79 

80 @abstractmethod 

81 def pin_card(self, card_id: str, pinned: bool = True) -> bool: 

82 """Pin or unpin a card""" 

83 pass 

84 

85 

86class RatingStorage(BaseStorage): 

87 """Interface for rating storage""" 

88 

89 @abstractmethod 

90 def get_user_rating( 

91 self, user_id: str, item_id: str, rating_type: str 

92 ) -> Optional[Dict[str, Any]]: 

93 """Get a user's rating for a specific item""" 

94 pass 

95 

96 @abstractmethod 

97 def upsert_rating( 

98 self, 

99 user_id: str, 

100 item_id: str, 

101 rating_type: str, 

102 rating_value: str, 

103 item_type: str = "card", 

104 ) -> str: 

105 """Create or update a rating""" 

106 pass 

107 

108 @abstractmethod 

109 def get_ratings_summary( 

110 self, item_id: str, item_type: str = "card" 

111 ) -> Dict[str, Any]: 

112 """Get aggregated ratings for an item""" 

113 pass 

114 

115 @abstractmethod 

116 def get_user_ratings( 

117 self, user_id: str, rating_type: Optional[str] = None, limit: int = 100 

118 ) -> List[Dict[str, Any]]: 

119 """Get all ratings by a user""" 

120 pass 

121 

122 

123class PreferenceStorage(BaseStorage): 

124 """Interface for user preference storage""" 

125 

126 @abstractmethod 

127 def get_user_preferences(self, user_id: str) -> Optional[Dict[str, Any]]: 

128 """Get preferences for a user""" 

129 pass 

130 

131 @abstractmethod 

132 def upsert_preferences( 

133 self, user_id: str, preferences: Dict[str, Any] 

134 ) -> str: 

135 """Create or update user preferences""" 

136 pass 

137 

138 @abstractmethod 

139 def add_liked_item( 

140 self, user_id: str, item_id: str, item_type: str = "news" 

141 ) -> bool: 

142 """Add an item to liked list""" 

143 pass 

144 

145 @abstractmethod 

146 def add_disliked_item( 

147 self, user_id: str, item_id: str, item_type: str = "news" 

148 ) -> bool: 

149 """Add an item to disliked list""" 

150 pass 

151 

152 @abstractmethod 

153 def update_preference_embedding( 

154 self, user_id: str, embedding: List[float] 

155 ) -> bool: 

156 """Update the user's preference embedding""" 

157 pass