Coverage for src / local_deep_research / news / preference_manager / storage.py: 0%
98 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-01-11 00:51 +0000
1"""
2SQLAlchemy storage implementation for user preferences.
3"""
5from typing import List, Optional, Dict, Any
6from sqlalchemy.orm import Session
8from ..core.storage import PreferenceStorage
9from ...database.models.news import UserPreference
12class SQLPreferenceStorage(PreferenceStorage):
13 """SQLAlchemy implementation of preference storage"""
15 def __init__(self, session: Session):
16 """Initialize with a database session from the user's encrypted database"""
17 if not session:
18 raise ValueError("Session is required for SQLPreferenceStorage")
19 self._session = session
21 @property
22 def session(self):
23 """Get database session"""
24 return self._session
26 def create(self, data: Dict[str, Any]) -> str:
27 """Create new user preferences"""
28 with self.session as session:
29 prefs = UserPreference(
30 user_id=data["user_id"],
31 liked_categories=data.get("liked_categories", []),
32 disliked_categories=data.get("disliked_categories", []),
33 liked_topics=data.get("liked_topics", []),
34 disliked_topics=data.get("disliked_topics", []),
35 impact_threshold=data.get("impact_threshold", 5),
36 focus_preferences=data.get("focus_preferences", {}),
37 custom_prompt=data.get("custom_prompt"),
38 custom_search_terms=data.get("custom_search_terms"),
39 preference_embedding=data.get("preference_embedding"),
40 liked_news_ids=data.get("liked_news_ids", []),
41 disliked_news_ids=data.get("disliked_news_ids", []),
42 )
44 session.add(prefs)
45 session.commit()
47 return str(prefs.id)
49 def get(self, id: str) -> Optional[Dict[str, Any]]:
50 """Get preferences by ID"""
51 with self.session as session:
52 prefs = session.query(UserPreference).filter_by(id=int(id)).first()
53 return prefs.to_dict() if prefs else None
55 def update(self, id: str, data: Dict[str, Any]) -> bool:
56 """Update preferences"""
57 with self.session as session:
58 prefs = session.query(UserPreference).filter_by(id=int(id)).first()
59 if not prefs:
60 return False
62 # Update fields
63 for field, value in data.items():
64 if hasattr(prefs, field):
65 setattr(prefs, field, value)
67 session.commit()
68 return True
70 def delete(self, id: str) -> bool:
71 """Delete preferences"""
72 with self.session as session:
73 prefs = session.query(UserPreference).filter_by(id=int(id)).first()
74 if not prefs:
75 return False
77 session.delete(prefs)
78 session.commit()
79 return True
81 def list(
82 self,
83 filters: Optional[Dict[str, Any]] = None,
84 limit: int = 100,
85 offset: int = 0,
86 ) -> List[Dict[str, Any]]:
87 """List preferences (usually just one per user)"""
88 with self.session as session:
89 query = session.query(UserPreference)
91 if filters and "user_id" in filters:
92 query = query.filter_by(user_id=filters["user_id"])
94 prefs = query.limit(limit).offset(offset).all()
95 return [p.to_dict() for p in prefs]
97 def get_user_preferences(self, user_id: str) -> Optional[Dict[str, Any]]:
98 """Get preferences for a user"""
99 with self.session as session:
100 prefs = (
101 session.query(UserPreference).filter_by(user_id=user_id).first()
102 )
103 return prefs.to_dict() if prefs else None
105 def upsert_preferences(
106 self, user_id: str, preferences: Dict[str, Any]
107 ) -> str:
108 """Create or update user preferences"""
109 existing = self.get_user_preferences(user_id)
111 if existing:
112 # Update existing preferences
113 with self.session as session:
114 prefs = (
115 session.query(UserPreference)
116 .filter_by(user_id=user_id)
117 .first()
118 )
120 # Update fields
121 for field, value in preferences.items():
122 if hasattr(prefs, field):
123 setattr(prefs, field, value)
125 session.commit()
126 return str(prefs.id)
127 else:
128 # Create new preferences
129 preferences["user_id"] = user_id
130 return self.create(preferences)
132 def add_liked_item(
133 self, user_id: str, item_id: str, item_type: str = "news"
134 ) -> bool:
135 """Add an item to liked list"""
136 with self.session as session:
137 prefs = (
138 session.query(UserPreference).filter_by(user_id=user_id).first()
139 )
141 if not prefs:
142 # Create new preferences
143 prefs = UserPreference(
144 user_id=user_id,
145 liked_news_ids=[item_id] if item_type == "news" else [],
146 disliked_news_ids=[],
147 )
148 session.add(prefs)
149 else:
150 # Update existing preferences
151 if item_type == "news":
152 liked_ids = prefs.liked_news_ids or []
153 if item_id not in liked_ids:
154 liked_ids.append(item_id)
155 # Keep last 100 items
156 prefs.liked_news_ids = liked_ids[-100:]
158 session.commit()
159 return True
161 def add_disliked_item(
162 self, user_id: str, item_id: str, item_type: str = "news"
163 ) -> bool:
164 """Add an item to disliked list"""
165 with self.session as session:
166 prefs = (
167 session.query(UserPreference).filter_by(user_id=user_id).first()
168 )
170 if not prefs:
171 # Create new preferences
172 prefs = UserPreference(
173 user_id=user_id,
174 liked_news_ids=[],
175 disliked_news_ids=[item_id] if item_type == "news" else [],
176 )
177 session.add(prefs)
178 else:
179 # Update existing preferences
180 if item_type == "news":
181 disliked_ids = prefs.disliked_news_ids or []
182 if item_id not in disliked_ids:
183 disliked_ids.append(item_id)
184 # Keep last 100 items
185 prefs.disliked_news_ids = disliked_ids[-100:]
187 session.commit()
188 return True
190 def update_preference_embedding(
191 self, user_id: str, embedding: List[float]
192 ) -> bool:
193 """Update the user's preference embedding"""
194 with self.session as session:
195 prefs = (
196 session.query(UserPreference).filter_by(user_id=user_id).first()
197 )
199 if not prefs:
200 # Create new preferences with embedding
201 prefs = UserPreference(
202 user_id=user_id, preference_embedding=embedding
203 )
204 session.add(prefs)
205 else:
206 prefs.preference_embedding = embedding
208 session.commit()
209 return True