Coverage for src/local_deep_research/notifications/queue_helpers.py: 97%
127 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +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 # DEBUG on purpose: notifications.on_research_queued defaults
74 # to off, so this drop is the steady state for every queued
75 # research — WARNING here would spam once per submission.
76 # Actionable causes (server gate, delivery failure) already
77 # WARN inside send_notification itself.
78 logger.debug(
79 f"Queue notification not sent for {research_id} "
80 f"{_format_reason(result)}"
81 )
82 return bool(getattr(result, "sent", result))
84 except RateLimitError:
85 logger.warning(
86 f"Rate limited sending queued notification for {research_id}"
87 )
88 return False
89 except Exception:
90 logger.warning(f"Failed to send queued notification for {research_id}")
91 return False
94def send_queue_failed_notification(
95 username: str,
96 research_id: str,
97 query: str,
98 error_message: Optional[str] = None,
99 settings_snapshot: Optional[Dict[str, Any]] = None,
100) -> bool:
101 """
102 Send a research failed notification from queue operations.
104 Args:
105 username: User who owns the research
106 research_id: UUID of the research
107 query: Research query string
108 error_message: Optional error message
109 settings_snapshot: Settings snapshot for thread-safe access
111 Returns:
112 True if notification was sent successfully, False otherwise.
113 Returned as ``bool`` for backward compatibility; the precise
114 reason for a falsy return is logged via ``NotificationResult``.
115 """
116 if not settings_snapshot:
117 logger.debug("No settings snapshot provided for failed notification")
118 return False
120 try:
121 notification_manager = NotificationManager(
122 settings_snapshot=settings_snapshot, user_id=username
123 )
125 # Build notification context
126 context = {
127 "query": query,
128 "research_id": research_id,
129 }
131 if error_message:
132 context["error"] = error_message
134 result = notification_manager.send_notification(
135 event_type=EventType.RESEARCH_FAILED,
136 context=context,
137 )
138 if not result:
139 # DEBUG on purpose: the from_session wrapper around this
140 # helper already emits its own WARNING on a falsy return, so
141 # a WARNING here would double-log every drop. The reason
142 # string is still available at debug level.
143 logger.debug(
144 f"Failed-notification not sent for {research_id} "
145 f"{_format_reason(result)}"
146 )
147 return bool(getattr(result, "sent", result))
149 except RateLimitError:
150 logger.warning(
151 f"Rate limited sending failed notification for {research_id}"
152 )
153 return False
154 except Exception:
155 logger.warning(f"Failed to send failed notification for {research_id}")
156 return False
159def send_queue_failed_notification_from_session(
160 username: str,
161 research_id: str,
162 query: str,
163 error_message: str,
164 db_session: Session,
165) -> None:
166 """
167 Send a research failed notification, fetching settings from db_session.
169 This is a convenience wrapper for the queue processor that handles
170 settings snapshot retrieval, logging, and error handling internally.
171 All notification logic is contained within this function.
173 Args:
174 username: User who owns the research
175 research_id: UUID of the research
176 query: Research query string
177 error_message: Error message to include in notification
178 db_session: Database session to fetch settings from
179 """
180 try:
181 from ..settings import SettingsManager
183 # Get settings snapshot from database session
184 settings_manager = SettingsManager(db_session)
185 settings_snapshot = settings_manager.get_settings_snapshot()
187 # Send notification (handles RateLimitError internally, returns False)
188 success = send_queue_failed_notification(
189 username=username,
190 research_id=research_id,
191 query=query,
192 error_message=error_message,
193 settings_snapshot=settings_snapshot,
194 )
196 if success:
197 logger.info(f"Sent failure notification for research {research_id}")
198 else:
199 logger.warning(
200 f"Failed to send failure notification for {research_id} (not sent)"
201 )
202 except Exception:
203 logger.exception(
204 f"Failed to send failure notification for {research_id}"
205 )
208def send_research_completed_notification_from_session(
209 username: str,
210 research_id: str,
211 db_session: Session,
212) -> None:
213 """
214 Send research completed notification with summary and URL.
216 This is a convenience wrapper for the queue processor that handles
217 all notification logic for completed research, including:
218 - Research database lookup
219 - Report content retrieval
220 - URL building
221 - Context building with summary
222 - All logging and error handling
224 Args:
225 username: User who owns the research
226 research_id: UUID of the research
227 db_session: Database session to fetch research and settings from
228 """
229 try:
230 logger.info(
231 f"Starting completed notification process for research {research_id}, "
232 f"user {username}"
233 )
235 # Import here to avoid circular dependencies
236 from ..database.models import ResearchHistory
237 from ..settings import SettingsManager
238 from .manager import NotificationManager
239 from .url_builder import build_notification_url
241 # Get research details for notification
242 research = (
243 db_session.query(ResearchHistory).filter_by(id=research_id).first()
244 )
246 # Get settings snapshot for thread-safe notification sending
247 settings_manager = SettingsManager(db_session)
248 settings_snapshot = settings_manager.get_settings_snapshot()
250 if research:
251 logger.info(
252 f"Found research record, creating NotificationManager "
253 f"for user {username}"
254 )
256 # Create notification manager with settings snapshot
257 notification_manager = NotificationManager(
258 settings_snapshot=settings_snapshot, user_id=username
259 )
261 # Build full URL for notification
262 full_url = build_notification_url(
263 f"/research/{research_id}",
264 settings_manager=settings_manager,
265 )
267 # Build notification context with required fields
268 context = {
269 "query": research.query or "Unknown query",
270 "research_id": research_id,
271 "summary": "No summary available",
272 "url": full_url,
273 }
275 # Get report content for notification
276 from ..storage import get_report_storage
278 storage = get_report_storage(session=db_session)
279 report_content = storage.get_report(research_id)
281 if report_content:
282 # Truncate summary if too long
283 context["summary"] = (
284 report_content[:200] + "..."
285 if len(report_content) > 200
286 else report_content
287 )
289 logger.info(
290 f"Sending RESEARCH_COMPLETED notification for research "
291 f"{research_id} to user {username}"
292 )
293 logger.debug(f"Notification context: {context}")
295 # Send notification using the manager
296 result = notification_manager.send_notification(
297 event_type=EventType.RESEARCH_COMPLETED,
298 context=context,
299 )
301 if result:
302 logger.info(
303 f"Successfully sent completion notification for research {research_id}"
304 )
305 else:
306 logger.warning(
307 f"Completion notification not sent for {research_id} "
308 f"{_format_reason(result)}"
309 )
311 else:
312 logger.warning(
313 f"Could not find research {research_id} in database, "
314 f"sending notification with minimal details"
315 )
317 # Create notification manager with settings snapshot
318 notification_manager = NotificationManager(
319 settings_snapshot=settings_snapshot, user_id=username
320 )
322 # Build minimal context
323 context = {
324 "query": f"Research {research_id}",
325 "research_id": research_id,
326 "summary": "Research completed but details unavailable",
327 "url": f"/research/{research_id}",
328 }
330 minimal_result = notification_manager.send_notification(
331 event_type=EventType.RESEARCH_COMPLETED,
332 context=context,
333 )
334 if minimal_result:
335 logger.info(
336 f"Sent completion notification for research {research_id} "
337 f"(minimal details)"
338 )
339 else:
340 logger.warning(
341 f"Completion notification not sent for {research_id} "
342 f"(minimal details) {_format_reason(minimal_result)}"
343 )
345 except RateLimitError:
346 logger.warning(
347 f"Rate limited sending completion notification for {research_id}"
348 )
349 except Exception:
350 logger.exception(
351 f"Failed to send completion notification for {research_id}"
352 )
355def send_research_failed_notification_from_session(
356 username: str,
357 research_id: str,
358 error_message: str,
359 db_session: Session,
360) -> None:
361 """
362 Send research failed notification (research-specific version).
364 This is a convenience wrapper for the queue processor that handles
365 all notification logic for failed research, including:
366 - Research database lookup to get query
367 - Context building with sanitized error message
368 - All logging and error handling
370 Args:
371 username: User who owns the research
372 research_id: UUID of the research
373 error_message: Error message (will be sanitized for security)
374 db_session: Database session to fetch research and settings from
375 """
376 try:
377 logger.info(
378 f"Starting failed notification process for research {research_id}, "
379 f"user {username}"
380 )
382 # Import here to avoid circular dependencies
383 from ..database.models import ResearchHistory
384 from ..settings import SettingsManager
385 from .manager import NotificationManager
387 # Get research details for notification
388 research = (
389 db_session.query(ResearchHistory).filter_by(id=research_id).first()
390 )
392 # Get settings snapshot for thread-safe notification sending
393 settings_manager = SettingsManager(db_session)
394 settings_snapshot = settings_manager.get_settings_snapshot()
396 # Sanitize error message for notification to avoid exposing
397 # sensitive information (as noted by github-advanced-security)
398 safe_error = "Research failed. Check logs for details."
400 if research:
401 logger.info(
402 f"Found research record, creating NotificationManager "
403 f"for user {username}"
404 )
406 # Create notification manager with settings snapshot
407 notification_manager = NotificationManager(
408 settings_snapshot=settings_snapshot, user_id=username
409 )
411 # Build notification context
412 context = {
413 "query": research.query or "Unknown query",
414 "research_id": research_id,
415 "error": safe_error,
416 }
418 logger.info(
419 f"Sending RESEARCH_FAILED notification for research "
420 f"{research_id} to user {username}"
421 )
422 logger.debug(f"Notification context: {context}")
424 # Send notification using the manager
425 result = notification_manager.send_notification(
426 event_type=EventType.RESEARCH_FAILED,
427 context=context,
428 )
430 if result:
431 logger.info(
432 f"Successfully sent failure notification for research {research_id}"
433 )
434 else:
435 logger.warning(
436 f"Failure notification not sent for {research_id} "
437 f"{_format_reason(result)}"
438 )
440 else:
441 logger.warning(
442 f"Could not find research {research_id} in database, "
443 f"sending notification with minimal details"
444 )
446 # Create notification manager with settings snapshot
447 notification_manager = NotificationManager(
448 settings_snapshot=settings_snapshot, user_id=username
449 )
451 # Build minimal context
452 context = {
453 "query": f"Research {research_id}",
454 "research_id": research_id,
455 "error": safe_error,
456 }
458 minimal_result = notification_manager.send_notification(
459 event_type=EventType.RESEARCH_FAILED,
460 context=context,
461 )
462 if minimal_result:
463 logger.info(
464 f"Sent failure notification for research {research_id} "
465 f"(minimal details)"
466 )
467 else:
468 logger.warning(
469 f"Failure notification not sent for {research_id} "
470 f"(minimal details) {_format_reason(minimal_result)}"
471 )
473 except RateLimitError:
474 logger.warning(
475 f"Rate limited sending failure notification for {research_id}"
476 )
477 except Exception:
478 logger.exception(
479 f"Failed to send failure notification for {research_id}"
480 )