Coverage for src/local_deep_research/news/core/card_storage.py: 97%
149 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-20 01:24 +0000
1"""
2SQLAlchemy storage implementation for news cards.
4Note: This module aligns with the NewsCard SQLAlchemy model in database/models/news.py.
5The NewsCard model has these relevant fields:
6- id, title, summary, content, url
7- source_name, source_type, source_id
8- category, tags, card_type
9- published_at, discovered_at
10- is_read, read_at, is_saved, saved_at
11- extra_data, subscription_id
12"""
14from datetime import datetime, timedelta, timezone
15from typing import List, Optional, Dict, Any
16from sqlalchemy.orm import Session
17from sqlalchemy import desc
18from loguru import logger
20from .storage import CardStorage
21from ...database.models.news import NewsCard
24class SQLCardStorage(CardStorage):
25 """SQLAlchemy implementation of card storage.
27 Maps between the card system's data model and the NewsCard database model.
28 Some fields from the card system are stored in extra_data JSON field.
29 """
31 def __init__(self, session: Session):
32 """Initialize with a database session from the user's encrypted database"""
33 if not session:
34 raise ValueError("Session is required for SQLCardStorage")
35 self._session = session
37 @property
38 def session(self):
39 """Get database session"""
40 return self._session
42 def create(self, data: Dict[str, Any]) -> str:
43 """Create a new card.
45 Maps card system fields to NewsCard model:
46 - topic → title
47 - user_id, parent_card_id, created_from → stored in extra_data
48 """
49 card_id = data.get("id") or self.generate_id()
51 # Extract source info if it's nested under "source"; otherwise
52 # fall back to the flat source_type / source_id / created_from
53 # fields. Note: `data.get("source", {})` returns {} when missing,
54 # which is itself a dict — so check truthiness, not type, to
55 # decide which branch applies.
56 source_info = data.get("source")
57 if isinstance(source_info, dict) and source_info:
58 source_type = source_info.get("type")
59 source_id = source_info.get("source_id")
60 created_from = source_info.get("created_from")
61 else:
62 source_type = data.get("source_type")
63 source_id = data.get("source_id")
64 created_from = data.get("created_from")
66 # Map card_type enum properly
67 card_type_str = data.get("card_type", data.get("type", "news"))
69 # Store extended fields in extra_data
70 extra_data = data.get("extra_data", {}) or {}
71 extra_data.update(
72 {
73 "user_id": data.get("user_id"),
74 "parent_card_id": data.get("parent_card_id"),
75 "created_from": created_from,
76 "metadata": data.get("metadata", {}),
77 "interaction": data.get("interaction", {}),
78 }
79 )
81 with self.session as session:
82 card = NewsCard(
83 id=card_id,
84 title=data.get("topic", data.get("title", "Untitled")),
85 summary=data.get("summary"),
86 content=data.get("content"),
87 url=data.get("url", data.get("source_url")),
88 source_name=data.get("source_name"),
89 source_type=source_type,
90 source_id=source_id,
91 category=data.get("category"),
92 tags=data.get("tags"),
93 card_type=card_type_str,
94 extra_data=extra_data,
95 )
97 session.add(card)
98 session.commit()
100 user_id = data.get("user_id", "unknown")
101 logger.info(f"Created card {card_id} for user {user_id}")
102 return card_id
104 def get(self, id: str) -> Optional[Dict[str, Any]]:
105 """Get a card by ID"""
106 with self.session as session:
107 card = session.query(NewsCard).filter_by(id=id).first()
108 if not card:
109 return None
110 return self._card_to_dict(card)
112 def update(self, id: str, data: Dict[str, Any]) -> bool:
113 """Update a card.
115 Maps card system fields to NewsCard model:
116 - is_archived → stored in extra_data
117 - is_pinned → is_saved
118 - last_viewed → read_at (and sets is_read=True)
119 """
120 with self.session as session:
121 card = session.query(NewsCard).filter_by(id=id).first()
122 if not card:
123 return False
125 # Map is_pinned to is_saved
126 if "is_pinned" in data:
127 card.is_saved = data["is_pinned"]
128 if data["is_pinned"]:
129 card.saved_at = datetime.now(timezone.utc)
131 # Map last_viewed to read_at
132 if "last_viewed" in data:
133 card.is_read = True
134 card.read_at = data["last_viewed"]
136 # Store is_archived and other custom fields in extra_data
137 extra_data = card.extra_data or {}
138 if "is_archived" in data:
139 extra_data["is_archived"] = data["is_archived"]
140 if "interaction" in data:
141 extra_data["interaction"] = data["interaction"]
142 card.extra_data = extra_data
144 session.commit()
145 return True
147 def delete(self, id: str) -> bool:
148 """Delete a card"""
149 with self.session as session:
150 card = session.query(NewsCard).filter_by(id=id).first()
151 if not card:
152 return False
154 session.delete(card)
155 session.commit()
156 return True
158 def list(
159 self,
160 filters: Optional[Dict[str, Any]] = None,
161 limit: int = 100,
162 offset: int = 0,
163 ) -> List[Dict[str, Any]]:
164 """List cards with optional filtering.
166 Supported filters:
167 - user_id: Filter by user (stored in extra_data)
168 - card_type: Filter by card type
169 - is_archived: Filter by archived status (in extra_data)
170 - is_pinned: Filter by pinned/saved status
171 - category: Filter by category
172 """
173 with self.session as session:
174 query = session.query(NewsCard)
176 if filters:
177 if "card_type" in filters:
178 card_type_val = filters["card_type"]
179 # Handle both string and list of types
180 if isinstance(card_type_val, list):
181 query = query.filter(
182 NewsCard.card_type.in_(card_type_val)
183 )
184 else:
185 query = query.filter_by(card_type=card_type_val)
186 if "is_pinned" in filters:
187 query = query.filter_by(is_saved=filters["is_pinned"])
188 if "category" in filters:
189 query = query.filter_by(category=filters["category"])
190 # Note: user_id and is_archived filtering would require
191 # JSON querying which varies by database backend
193 # Order by discovery date (newest first)
194 query = query.order_by(desc(NewsCard.discovered_at))
196 cards = query.limit(limit).offset(offset).all()
197 return [self._card_to_dict(card) for card in cards]
199 def get_recent(
200 self,
201 hours: int = 24,
202 card_types: Optional[List[str]] = None,
203 limit: int = 50,
204 ) -> List[Dict[str, Any]]:
205 """Get recent cards within the specified time window.
207 Args:
208 hours: How many hours back to look (default 24)
209 card_types: Optional list of card types to filter
210 limit: Maximum number of cards to return
212 Returns:
213 List of card dictionaries
214 """
215 cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
217 with self.session as session:
218 query = session.query(NewsCard).filter(
219 NewsCard.discovered_at >= cutoff
220 )
222 if card_types:
223 query = query.filter(NewsCard.card_type.in_(card_types))
225 query = query.order_by(desc(NewsCard.discovered_at))
226 cards = query.limit(limit).all()
228 return [self._card_to_dict(card) for card in cards]
230 def _card_to_dict(self, card: NewsCard) -> Dict[str, Any]:
231 """Convert a NewsCard model to the dictionary format expected by the card system.
233 Maps NewsCard model fields back to card system format:
234 - title → topic
235 - is_saved → is_pinned
236 - extra_data fields → top-level fields
237 """
238 extra_data: Dict[str, Any] = (
239 dict(card.extra_data) if card.extra_data else {}
240 )
242 return {
243 "id": card.id,
244 "topic": card.title, # Map title back to topic
245 "title": card.title,
246 "summary": card.summary,
247 "content": card.content,
248 "url": card.url,
249 "source_name": card.source_name,
250 "source_type": card.source_type,
251 "source_id": card.source_id,
252 "category": card.category,
253 "tags": card.tags,
254 "card_type": card.card_type,
255 "published_at": card.published_at.isoformat()
256 if card.published_at
257 else None,
258 "discovered_at": card.discovered_at.isoformat()
259 if card.discovered_at
260 else None,
261 "created_at": card.discovered_at.isoformat()
262 if card.discovered_at
263 else None, # Alias for compatibility
264 "updated_at": card.discovered_at.isoformat()
265 if card.discovered_at
266 else None, # Best approximation
267 "is_read": card.is_read,
268 "read_at": card.read_at.isoformat() if card.read_at else None,
269 "is_saved": card.is_saved,
270 "is_pinned": card.is_saved, # Alias for compatibility
271 "saved_at": card.saved_at.isoformat() if card.saved_at else None,
272 # Fields from extra_data
273 "user_id": extra_data.get("user_id"),
274 "parent_card_id": extra_data.get("parent_card_id"),
275 "created_from": extra_data.get("created_from"),
276 "is_archived": extra_data.get("is_archived", False),
277 "metadata": extra_data.get("metadata", {}),
278 "interaction": extra_data.get("interaction", {}),
279 "source": {
280 "type": card.source_type,
281 "source_id": card.source_id,
282 "created_from": extra_data.get("created_from", ""),
283 "metadata": extra_data.get("metadata", {}),
284 },
285 }
287 def get_by_user(
288 self, user_id: str, limit: int = 50, offset: int = 0
289 ) -> List[Dict[str, Any]]:
290 """Get cards for a specific user.
292 Note: Since user_id is stored in extra_data JSON, this does a
293 post-filter. For better performance with large datasets,
294 consider adding a proper user_id column.
295 """
296 # Get more cards than needed to account for filtering
297 all_cards = self.list(filters=None, limit=limit * 3, offset=0)
299 # Filter by user_id from extra_data
300 user_cards = [
301 card
302 for card in all_cards
303 if card.get("user_id") == user_id
304 and not card.get("is_archived", False)
305 ]
307 # Apply pagination
308 return user_cards[offset : offset + limit]
310 def get_latest_version(self, card_id: str) -> Optional[Dict[str, Any]]:
311 """Get the latest version of a card.
313 Note: The versioning system is not yet implemented at the database level.
314 CardVersion is a Python dataclass for in-memory use, not a SQLAlchemy model.
315 This method returns version info stored in extra_data if available.
316 """
317 card_data = self.get(card_id)
318 if not card_data:
319 return None
321 # Check if version info is stored in extra_data
322 extra_data = card_data.get("metadata", {})
323 if "latest_version" in extra_data: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 result: Dict[str, Any] = extra_data["latest_version"]
325 return result
327 # Return card's current state as version 1
328 return {
329 "version_id": f"{card_id}_v1",
330 "version_number": 1,
331 "headline": card_data.get("title"),
332 "summary": card_data.get("summary"),
333 "card_id": card_id,
334 }
336 def add_version(self, card_id: str, version_data: Dict[str, Any]) -> str:
337 """Add a new version to a card.
339 Note: The versioning system stores version data in the card's extra_data
340 field since CardVersion is not a database model. For full versioning
341 support, a CardVersion SQLAlchemy model would need to be created.
342 """
343 version_id = version_data.get("id") or self.generate_id()
345 with self.session as session:
346 card = session.query(NewsCard).filter_by(id=card_id).first()
347 if not card:
348 raise ValueError(f"Card {card_id} not found")
350 # Get current version count from extra_data
351 extra_data = card.extra_data or {}
352 versions = extra_data.get("versions", [])
353 version_number = len(versions) + 1
355 # Create version record
356 version_record = {
357 "id": version_id,
358 "version_number": version_number,
359 "search_query": version_data.get("search_query"),
360 "headline": version_data.get("headline"),
361 "summary": version_data.get("summary"),
362 "findings": version_data.get("findings"),
363 "sources": version_data.get("sources"),
364 "impact_score": version_data.get("impact_score"),
365 "topics": version_data.get("topics"),
366 "entities": version_data.get("entities"),
367 "created_at": datetime.now(timezone.utc).isoformat(),
368 }
370 versions.append(version_record)
371 extra_data["versions"] = versions
372 extra_data["latest_version"] = version_record
374 # Update card fields with latest version info
375 if version_data.get("headline"): 375 ↛ 377line 375 didn't jump to line 377 because the condition on line 375 was always true
376 card.title = version_data["headline"]
377 if version_data.get("summary"):
378 card.summary = version_data["summary"]
380 card.extra_data = extra_data
381 session.commit()
383 logger.info(f"Added version {version_number} to card {card_id}")
384 return version_id
386 def update_latest_info(
387 self, card_id: str, version_data: Dict[str, Any]
388 ) -> bool:
389 """Update the denormalized latest version info on the card.
391 Updates the card's main fields with the latest version data.
392 """
393 with self.session as session:
394 card = session.query(NewsCard).filter_by(id=card_id).first()
395 if not card:
396 return False
398 # Update card's display fields
399 if version_data.get("headline"): 399 ↛ 401line 399 didn't jump to line 401 because the condition on line 399 was always true
400 card.title = version_data["headline"]
401 if version_data.get("summary"): 401 ↛ 405line 401 didn't jump to line 405 because the condition on line 401 was always true
402 card.summary = version_data["summary"]
404 # Store version metadata in extra_data
405 extra_data = card.extra_data or {}
406 extra_data["latest_version"] = {
407 "id": version_data.get("id"),
408 "headline": version_data.get("headline"),
409 "summary": version_data.get("summary"),
410 "impact_score": version_data.get("impact_score"),
411 }
412 card.extra_data = extra_data
414 session.commit()
415 return True
417 def archive_card(self, card_id: str) -> bool:
418 """Archive a card"""
419 return self.update(card_id, {"is_archived": True})
421 def pin_card(self, card_id: str, pinned: bool = True) -> bool:
422 """Pin or unpin a card"""
423 return self.update(card_id, {"is_pinned": pinned})