Coverage for src/local_deep_research/web/auth/queue_middleware.py: 95%

1""" 

2Middleware to process pending queue operations when user has active session. 

3""" 

4 

5from flask import g 

6from loguru import logger 

7 

8from ...database.encrypted_db import db_manager 

9from ...database.session_context import get_user_db_session 

10 

11# Keep the singleton import lazy so disabled pytest app startup does not 

12# initialize the queue processor. Tests can still replace this module-level 

13# seam with a lightweight mock. 

14queue_processor = None 

15 

16 

17def process_pending_queue_operations(): 

18 """ 

19 Process pending queue operations for the current user. 

20 This runs in request context where we have access to the encrypted database. 

21 """ 

22 # Check if user is authenticated and has a database session 

23 if not hasattr(g, "current_user") or not g.current_user: 

24 return 

25 

26 # g.current_user might be a string (username) or an object 

27 username = ( 

28 g.current_user 

29 if isinstance(g.current_user, str) 

30 else g.current_user.username 

31 ) 

32 

33 # Check if user has an open database connection 

34 if not db_manager.is_user_connected(username): 

35 return 

36 

37 try: 

38 processor = queue_processor 

39 if processor is None: 

40 from ..queue.processor_v2 import queue_processor as processor 

41 

42 # Use the session context manager to properly handle the session 

43 with get_user_db_session(username) as db_session: 

44 if not db_session: 

45 logger.warning( 

46 f"get_user_db_session returned None for {username} during queue processing" 

47 ) 

48 return 

49 

50 # Process any pending operations for this user 

51 started_count = processor.process_pending_operations_for_user( 

52 username, db_session 

53 ) 

54 

55 if started_count > 0: 

56 logger.info( 

57 f"Started {started_count} queued researches for {username}" 

58 ) 

59 

60 except Exception: 

61 logger.exception( 

62 f"Error processing pending queue operations for {username}" 

63 ) 

64 # Rollback g.db_session to prevent PendingRollbackError 

65 # cascading to the view function that uses the same session 

66 if hasattr(g, "db_session") and g.db_session: 

67 try: 

68 g.db_session.rollback() 

69 except Exception: 

70 logger.debug( 

71 "Failed to rollback g.db_session after queue error" 

72 )