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

1"""Storage that always uses database as primary storage with optional file backup.""" 

2 

3from typing import Dict, Any, Optional 

4from loguru import logger 

5from sqlalchemy.orm import Session 

6 

7from .base import ReportStorage 

8from .database import DatabaseReportStorage 

9from .file import FileReportStorage 

10from ..config.paths import ( 

11 REPORT_BACKUP_REMEDY, 

12 describe_private_directory_refusal, 

13) 

14 

15 

16class DatabaseWithFileBackupStorage(ReportStorage): 

17 """ 

18 Storage that always saves to database and optionally backs up to file system. 

19 

20 Database is the primary storage and is always used. 

21 File storage is optional for external access/backup purposes. 

22 """ 

23 

24 def __init__(self, session: Session, enable_file_storage: bool = False): 

25 """ 

26 Initialize combined storage. 

27 

28 Args: 

29 session: SQLAlchemy database session 

30 enable_file_storage: Whether to also save reports to file system. 

31 If FileReportStorage refuses its directory with an OSError 

32 (for example one that is not private), ``file_storage`` 

33 stays None, a warning names the cause, and reports are saved 

34 to the database only. 

35 """ 

36 self.db_storage = DatabaseReportStorage(session) 

37 self.file_storage = None 

38 if enable_file_storage: 

39 try: 

40 self.file_storage = FileReportStorage() 

41 except OSError as exc: 

42 cause = describe_private_directory_refusal(exc) 

43 logger.warning( 

44 "Report file backup is disabled because the research " 

45 f"outputs directory was refused: {cause}. Reports are " 

46 f"still saved in the database. {REPORT_BACKUP_REMEDY}." 

47 ) 

48 self.enable_file_storage = enable_file_storage 

49 

50 def save_report( 

51 self, 

52 research_id: str, 

53 content: str, 

54 metadata: Optional[Dict[str, Any]] = None, 

55 username: Optional[str] = None, 

56 ) -> bool: 

57 """Save report to database and optionally to file.""" 

58 # Always save to database first (primary storage) 

59 success = self.db_storage.save_report( 

60 research_id, content, metadata, username 

61 ) 

62 

63 if not success: 

64 logger.error(f"Failed to save report {research_id} to database") 

65 return False 

66 

67 # Optionally save to file system 

68 if self.file_storage: 

69 try: 

70 file_success = self.file_storage.save_report( 

71 research_id, content, metadata, username 

72 ) 

73 if not file_success: 

74 logger.warning( 

75 f"Failed to save report {research_id} to file, " 

76 "but database save was successful" 

77 ) 

78 except Exception: 

79 logger.warning( 

80 f"Error saving report {research_id} to file" 

81 "but database save was successful" 

82 ) 

83 

84 return success 

85 

86 def get_report( 

87 self, research_id: str, username: Optional[str] = None 

88 ) -> Optional[str]: 

89 """Get report from database (never from file).""" 

90 # Always read from database for consistency 

91 return self.db_storage.get_report(research_id, username) 

92 

93 def delete_report( 

94 self, research_id: str, username: Optional[str] = None 

95 ) -> bool: 

96 """Delete report from database and file if it exists.""" 

97 # Delete from database 

98 db_success = self.db_storage.delete_report(research_id, username) 

99 

100 # Also delete from file if enabled 

101 if self.file_storage and db_success: 

102 try: 

103 self.file_storage.delete_report(research_id, username) 

104 except Exception: 

105 logger.warning(f"Error deleting report file {research_id}") 

106 

107 return db_success 

108 

109 def list_reports(self, username: Optional[str] = None) -> list: 

110 """List reports from database only.""" 

111 # Always list from database for consistency 

112 return self.db_storage.list_reports(username) 

113 

114 def get_report_with_metadata( 

115 self, research_id: str, username: Optional[str] = None 

116 ) -> Optional[Dict[str, Any]]: 

117 """Get report with metadata from database.""" 

118 # Always read from database for consistency 

119 return self.db_storage.get_report_with_metadata(research_id, username) 

120 

121 def report_exists( 

122 self, research_id: str, username: Optional[str] = None 

123 ) -> bool: 

124 """Check if report exists in database.""" 

125 # Always check database for consistency 

126 return self.db_storage.report_exists(research_id, username)