Coverage for src/local_deep_research/notifications/queue_helpers.py: 96%
127 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""
2Queue notification helpers for the notification system.
4Provides helper functions for sending queue-related notifications
5to keep the queue manager focused on queue logic.
6"""
8from typing import Any, Dict, Optional
9from loguru import logger
10from sqlalchemy.orm import Session
12from .exceptions import RateLimitError
13from .manager import NotificationManager, NotificationResult
14from .templates import EventType
17def _format_reason(result: Any) -> str:
18 """Render a ``send_notification`` outcome as ``"(reason: detail)"``,
19 falling back to ``bool(result)`` so legacy mocks / older callers that
20 still return a bare ``True``/``False`` keep working.
21 """
22 if isinstance(result, NotificationResult):
23 if result.detail: 23 ↛ 25line 23 didn't jump to line 25 because the condition on line 23 was always true
24 return f"({result.reason.value}: {result.detail})"
25 return f"({result.reason.value})"
26 return f"({'sent' if bool(result) else 'dropped'})"
29def send_queue_notification(
30 username: str,
31 research_id: str,
32 query: str,
33 settings_snapshot: Dict[str, Any],
34 position: Optional[int] = None,
35) -> bool:
36 """
37 Send a research queued notification.
39 Args:
40 username: User who owns the research
41 research_id: UUID of the research
42 query: Research query string
43 settings_snapshot: Settings snapshot for thread-safe access
44 position: Queue position (optional)
46 Returns:
47 True if notification was sent successfully, False otherwise.
48 Returned as ``bool`` for backward compatibility; the precise
49 reason for a falsy return is logged via ``NotificationResult``.
50 """
51 try:
52 notification_manager = NotificationManager(
53 settings_snapshot=settings_snapshot, user_id=username
54 )
56 # Build notification context
57 context: Dict[str, Any] = {
58 "query": query,
59 "research_id": research_id,
60 }
62 if position is not None:
63 context["position"] = position
64 context["wait_time"] = (
65 "Unknown" # Could estimate based on active researches
66 )
68 result = notification_manager.send_notification(
69 event_type=EventType.RESEARCH_QUEUED,
70 context=context,
71 )
72 if not result:
73 logger.debug(
74 f"Queue notification not sent for {research_id} "
75 f"{_format_reason(result)}"
76 )
77 return bool(getattr(result, "sent", result))
79 except RateLimitError:
80 logger.warning(
81 f"Rate limited sending queued notification for {research_id}"
82 )
83 return False
84 except Exception:
85 logger.warning(f"Failed to send queued notification for {research_id}")
86 return False
89def send_queue_failed_notification(
90 username: str,
91 research_id: str,
92 query: str,
93 error_message: Optional[str] = None,
94 settings_snapshot: Optional[Dict[str, Any]] = None,
95) -> bool:
96 """
97 Send a research failed notification from queue operations.
99 Args:
100 username: User who owns the research
101 research_id: UUID of the research
102 query: Research query string
103 error_message: Optional error message
104 settings_snapshot: Settings snapshot for thread-safe access
106 Returns:
107 True if notification was sent successfully, False otherwise.
108 Returned as ``bool`` for backward compatibility; the precise
109 reason for a falsy return is logged via ``NotificationResult``.
110 """
111 if not settings_snapshot:
112 logger.debug("No settings snapshot provided for failed notification")
113 return False
115 try:
116 notification_manager = NotificationManager(
117 settings_snapshot=settings_snapshot, user_id=username
118 )
120 # Build notification context
121 context = {
122 "query": query,
123 "research_id": research_id,
124 }
126 if error_message:
127 context["error"] = error_message
129 result = notification_manager.send_notification(
130 event_type=EventType.RESEARCH_FAILED,
131 context=context,
132 )
133 if not result:
134 logger.debug(
135 f"Failed-notification not sent for {research_id} "
136 f"{_format_reason(result)}"
137 )
138 return bool(getattr(result, "sent", result))
140 except RateLimitError:
141 logger.warning(
142 f"Rate limited sending failed notification for {research_id}"
143 )
144 return False
145 except Exception:
146 logger.warning(f"Failed to send failed notification for {research_id}")
147 return False
150def send_queue_failed_notification_from_session(
151 username: str,
152 research_id: str,
153 query: str,
154 error_message: str,
155 db_session: Session,
156) -> None:
157 """
158 Send a research failed notification, fetching settings from db_session.
160 This is a convenience wrapper for the queue processor that handles
161 settings snapshot retrieval, logging, and error handling internally.
162 All notification logic is contained within this function.
164 Args:
165 username: User who owns the research
166 research_id: UUID of the research
167 query: Research query string
168 error_message: Error message to include in notification
169 db_session: Database session to fetch settings from
170 """
171 try:
172 from ..settings import SettingsManager
174 # Get settings snapshot from database session
175 settings_manager = SettingsManager(db_session)
176 settings_snapshot = settings_manager.get_settings_snapshot()
178 # Send notification (handles RateLimitError internally, returns False)
179 success = send_queue_failed_notification(
180 username=username,
181 research_id=research_id,
182 query=query,
183 error_message=error_message,
184 settings_snapshot=settings_snapshot,
185 )
187 if success:
188 logger.info(f"Sent failure notification for research {research_id}")
189 else:
190 logger.warning(
191 f"Failed to send failure notification for {research_id} (not sent)"
192 )
193 except Exception:
194 logger.exception(
195 f"Failed to send failure notification for {research_id}"
196 )
199def send_research_completed_notification_from_session(
200 username: str,
201 research_id: str,
202 db_session: Session,
203) -> None:
204 """
205 Send research completed notification with summary and URL.
207 This is a convenience wrapper for the queue processor that handles
208 all notification logic for completed research, including:
209 - Research database lookup
210 - Report content retrieval
211 - URL building
212 - Context building with summary
213 - All logging and error handling
215 Args:
216 username: User who owns the research
217 research_id: UUID of the research
218 db_session: Database session to fetch research and settings from
219 """
220 try:
221 logger.info(
222 f"Starting completed notification process for research {research_id}, "
223 f"user {username}"
224 )
226 # Import here to avoid circular dependencies
227 from ..database.models import ResearchHistory
228 from ..settings import SettingsManager
229 from .manager import NotificationManager
230 from .url_builder import build_notification_url
232 # Get research details for notification
233 research = (
234 db_session.query(ResearchHistory).filter_by(id=research_id).first()
235 )
237 # Get settings snapshot for thread-safe notification sending
238 settings_manager = SettingsManager(db_session)
239 settings_snapshot = settings_manager.get_settings_snapshot()
241 if research:
242 logger.info(
243 f"Found research record, creating NotificationManager "
244 f"for user {username}"
245 )
247 # Create notification manager with settings snapshot
248 notification_manager = NotificationManager(
249 settings_snapshot=settings_snapshot, user_id=username
250 )
252 # Build full URL for notification
253 full_url = build_notification_url(
254 f"/research/{research_id}",
255 settings_manager=settings_manager,
256 )
258 # Build notification context with required fields
259 context = {
260 "query": research.query or "Unknown query",
261 "research_id": research_id,
262 "summary": "No summary available",
263 "url": full_url,
264 }
266 # Get report content for notification
267 from ..storage import get_report_storage
269 storage = get_report_storage(session=db_session)
270 report_content = storage.get_report(research_id)
272 if report_content:
273 # Truncate summary if too long
274 context["summary"] = (
275 report_content[:200] + "..."
276 if len(report_content) > 200
277 else report_content
278 )
280 logger.info(
281 f"Sending RESEARCH_COMPLETED notification for research "
282 f"{research_id} to user {username}"
283 )
284 logger.debug(f"Notification context: {context}")
286 # Send notification using the manager
287 result = notification_manager.send_notification(
288 event_type=EventType.RESEARCH_COMPLETED,
289 context=context,
290 )
292 if result:
293 logger.info(
294 f"Successfully sent completion notification for research {research_id}"
295 )
296 else:
297 logger.warning(
298 f"Completion notification not sent for {research_id} "
299 f"{_format_reason(result)}"
300 )
302 else:
303 logger.warning(
304 f"Could not find research {research_id} in database, "
305 f"sending notification with minimal details"
306 )
308 # Create notification manager with settings snapshot
309 notification_manager = NotificationManager(
310 settings_snapshot=settings_snapshot, user_id=username
311 )
313 # Build minimal context
314 context = {
315 "query": f"Research {research_id}",
316 "research_id": research_id,
317 "summary": "Research completed but details unavailable",
318 "url": f"/research/{research_id}",
319 }
321 minimal_result = notification_manager.send_notification(
322 event_type=EventType.RESEARCH_COMPLETED,
323 context=context,
324 )
325 if minimal_result:
326 logger.info(
327 f"Sent completion notification for research {research_id} "
328 f"(minimal details)"
329 )
330 else:
331 logger.warning(
332 f"Completion notification not sent for {research_id} "
333 f"(minimal details) {_format_reason(minimal_result)}"
334 )
336 except RateLimitError:
337 logger.warning(
338 f"Rate limited sending completion notification for {research_id}"
339 )
340 except Exception:
341 logger.exception(
342 f"Failed to send completion notification for {research_id}"
343 )
346def send_research_failed_notification_from_session(
347 username: str,
348 research_id: str,
349 error_message: str,
350 db_session: Session,
351) -> None:
352 """
353 Send research failed notification (research-specific version).
355 This is a convenience wrapper for the queue processor that handles
356 all notification logic for failed research, including:
357 - Research database lookup to get query
358 - Context building with sanitized error message
359 - All logging and error handling
361 Args:
362 username: User who owns the research
363 research_id: UUID of the research
364 error_message: Error message (will be sanitized for security)
365 db_session: Database session to fetch research and settings from
366 """
367 try:
368 logger.info(
369 f"Starting failed notification process for research {research_id}, "
370 f"user {username}"
371 )
373 # Import here to avoid circular dependencies
374 from ..database.models import ResearchHistory
375 from ..settings import SettingsManager
376 from .manager import NotificationManager
378 # Get research details for notification
379 research = (
380 db_session.query(ResearchHistory).filter_by(id=research_id).first()
381 )
383 # Get settings snapshot for thread-safe notification sending
384 settings_manager = SettingsManager(db_session)
385 settings_snapshot = settings_manager.get_settings_snapshot()
387 # Sanitize error message for notification to avoid exposing
388 # sensitive information (as noted by github-advanced-security)
389 safe_error = "Research failed. Check logs for details."
391 if research:
392 logger.info(
393 f"Found research record, creating NotificationManager "
394 f"for user {username}"
395 )
397 # Create notification manager with settings snapshot
398 notification_manager = NotificationManager(
399 settings_snapshot=settings_snapshot, user_id=username
400 )
402 # Build notification context
403 context = {
404 "query": research.query or "Unknown query",
405 "research_id": research_id,
406 "error": safe_error,
407 }
409 logger.info(
410 f"Sending RESEARCH_FAILED notification for research "
411 f"{research_id} to user {username}"
412 )
413 logger.debug(f"Notification context: {context}")
415 # Send notification using the manager
416 result = notification_manager.send_notification(
417 event_type=EventType.RESEARCH_FAILED,
418 context=context,
419 )
421 if result: 421 ↛ 426line 421 didn't jump to line 426 because the condition on line 421 was always true
422 logger.info(
423 f"Successfully sent failure notification for research {research_id}"
424 )
425 else:
426 logger.warning(
427 f"Failure notification not sent for {research_id} "
428 f"{_format_reason(result)}"
429 )
431 else:
432 logger.warning(
433 f"Could not find research {research_id} in database, "
434 f"sending notification with minimal details"
435 )
437 # Create notification manager with settings snapshot
438 notification_manager = NotificationManager(
439 settings_snapshot=settings_snapshot, user_id=username
440 )
442 # Build minimal context
443 context = {
444 "query": f"Research {research_id}",
445 "research_id": research_id,
446 "error": safe_error,
447 }
449 minimal_result = notification_manager.send_notification(
450 event_type=EventType.RESEARCH_FAILED,
451 context=context,
452 )
453 if minimal_result:
454 logger.info(
455 f"Sent failure notification for research {research_id} "
456 f"(minimal details)"
457 )
458 else:
459 logger.warning(
460 f"Failure notification not sent for {research_id} "
461 f"(minimal details) {_format_reason(minimal_result)}"
462 )
464 except RateLimitError:
465 logger.warning(
466 f"Rate limited sending failure notification for {research_id}"
467 )
468 except Exception:
469 logger.exception(
470 f"Failed to send failure notification for {research_id}"
471 )