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 )