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

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 

10 

11 

12class DatabaseWithFileBackupStorage(ReportStorage): 

13 """ 

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

15 

16 Database is the primary storage and is always used. 

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

18 """ 

19 

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

21 """ 

22 Initialize combined storage. 

23 

24 Args: 

25 session: SQLAlchemy database session 

26 enable_file_storage: Whether to also save reports to file system 

27 """ 

28 self.db_storage = DatabaseReportStorage(session) 

29 self.file_storage = FileReportStorage() if enable_file_storage else None 

30 self.enable_file_storage = enable_file_storage 

31 

32 def save_report( 

33 self, 

34 research_id: str, 

35 content: str, 

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

37 username: Optional[str] = None, 

38 ) -> bool: 

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

40 # Always save to database first (primary storage) 

41 success = self.db_storage.save_report( 

42 research_id, content, metadata, username 

43 ) 

44 

45 if not success: 

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

47 return False 

48 

49 # Optionally save to file system 

50 if self.file_storage: 

51 try: 

52 file_success = self.file_storage.save_report( 

53 research_id, content, metadata, username 

54 ) 

55 if not file_success: 

56 logger.warning( 

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

58 "but database save was successful" 

59 ) 

60 except Exception as e: 

61 logger.warning( 

62 f"Error saving report {research_id} to file: {e}, " 

63 "but database save was successful" 

64 ) 

65 

66 return success 

67 

68 def get_report( 

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

70 ) -> Optional[str]: 

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

72 # Always read from database for consistency 

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

74 

75 def delete_report( 

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

77 ) -> bool: 

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

79 # Delete from database 

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

81 

82 # Also delete from file if enabled 

83 if self.file_storage and db_success: 

84 try: 

85 self.file_storage.delete_report(research_id, username) 

86 except Exception as e: 

87 logger.warning(f"Error deleting report file {research_id}: {e}") 

88 

89 return db_success 

90 

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

92 """List reports from database only.""" 

93 # Always list from database for consistency 

94 return self.db_storage.list_reports(username) 

95 

96 def get_report_with_metadata( 

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

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

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

100 # Always read from database for consistency 

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

102 

103 def report_exists( 

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

105 ) -> bool: 

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

107 # Always check database for consistency 

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