Coverage for src / local_deep_research / storage / factory.py: 100%

1"""Factory for creating report storage instances.""" 

2 

3from typing import Optional 

4from sqlalchemy.orm import Session 

5from loguru import logger 

6 

7from .base import ReportStorage 

8from .database_with_file_backup import DatabaseWithFileBackupStorage 

9from ..config.thread_settings import ( 

10 get_setting_from_snapshot, 

11 NoSettingsContextError, 

12) 

13 

14 

15def get_report_storage( 

16 session: Optional[Session] = None, 

17 settings_snapshot: Optional[dict] = None, 

18 enable_file_backup: Optional[bool] = None, 

19) -> ReportStorage: 

20 """Get a report storage instance that always uses database with optional file backup. 

21 

22 Args: 

23 session: Database session (required) 

24 settings_snapshot: Settings snapshot for thread-safe access 

25 enable_file_backup: Whether to enable file backup. If None, uses setting. 

26 

27 Returns: 

28 ReportStorage instance (DatabaseWithFileBackupStorage) 

29 

30 Raises: 

31 ValueError: If database session is not provided 

32 """ 

33 if session is None: 

34 raise ValueError("Database session is required for report storage") 

35 

36 # Determine if file backup should be enabled 

37 if enable_file_backup is None: 

38 try: 

39 enable_file_backup = get_setting_from_snapshot( 

40 "report.enable_file_backup", 

41 settings_snapshot=settings_snapshot, 

42 ) 

43 except NoSettingsContextError: 

44 # Fall back to default if no settings context 

45 enable_file_backup = False 

46 

47 logger.info( 

48 f"Report storage: Database (primary) with file backup {'enabled' if enable_file_backup else 'disabled'}" 

49 ) 

50 

51 return DatabaseWithFileBackupStorage( 

52 session=session, enable_file_storage=enable_file_backup 

53 ) 

54 

55 

56# Global singleton for request context 

57_request_storage: Optional[ReportStorage] = None 

58 

59 

60def get_request_report_storage() -> Optional[ReportStorage]: 

61 """Get the report storage instance for the current request context.""" 

62 return _request_storage 

63 

64 

65def set_request_report_storage(storage: ReportStorage) -> None: 

66 """Set the report storage instance for the current request context.""" 

67 global _request_storage 

68 _request_storage = storage 

69 

70 

71def clear_request_report_storage() -> None: 

72 """Clear the report storage instance for the current request context.""" 

73 global _request_storage 

74 _request_storage = None