Coverage for src/local_deep_research/web/models/database.py: 98%

1from datetime import datetime, UTC 

2 

3from loguru import logger 

4 

5from ...config.paths import get_data_directory 

6from ...database.models import ResearchLog 

7from ...database.session_context import get_user_db_session 

8 

9# Database paths using new centralized configuration 

10_raw_data_dir = get_data_directory() 

11DATA_DIR: str | None = None 

12if _raw_data_dir: 12 ↛ 23line 12 didn't jump to line 23 because the condition on line 12 was always true

13 DATA_DIR = str(_raw_data_dir) 

14 # Lazy import: this module executes at import time (early bootstrap), 

15 # so avoid a top-level security -> ... -> web import cycle. 

16 from ...security.directory_creation import create_directory 

17 

18 create_directory(DATA_DIR, context="application data directory") 

19 

20# DB_PATH removed - use per-user encrypted databases instead 

21 

22 

23def get_db_connection(): 

24 """ 

25 Get a connection to the SQLite database. 

26 DEPRECATED: This uses the shared database which should not be used. 

27 Use get_db_session() instead for per-user databases. 

28 """ 

29 raise RuntimeError( 

30 "Shared database access is deprecated. Use get_db_session() for per-user databases." 

31 ) 

32 

33 

34def calculate_duration(created_at_str, completed_at_str=None): 

35 """ 

36 Calculate duration in seconds between created_at timestamp and completed_at or now. 

37 Handles various timestamp formats and returns None if calculation fails. 

38 

39 Args: 

40 created_at_str: The start timestamp 

41 completed_at_str: Optional end timestamp, defaults to current time if None 

42 

43 Returns: 

44 Duration in seconds or None if calculation fails 

45 """ 

46 if not created_at_str: 

47 return None 

48 

49 end_time = None 

50 if completed_at_str: 

51 # Use completed_at time if provided 

52 try: 

53 if "T" in completed_at_str: # ISO format with T separator 

54 end_time = datetime.fromisoformat(completed_at_str) 

55 else: # Older format without T 

56 # Try different formats 

57 try: 

58 end_time = datetime.strptime( 

59 completed_at_str, "%Y-%m-%d %H:%M:%S.%f" 

60 ) 

61 except ValueError: 

62 try: 

63 end_time = datetime.strptime( 

64 completed_at_str, "%Y-%m-%d %H:%M:%S" 

65 ) 

66 except ValueError: 

67 # Last resort fallback 

68 end_time = datetime.fromisoformat( 

69 completed_at_str.replace(" ", "T") 

70 ) 

71 except Exception: 

72 logger.exception("Error parsing completed_at timestamp") 

73 try: 

74 from dateutil import parser # type: ignore[import-untyped] 

75 

76 end_time = parser.parse(completed_at_str) 

77 except Exception: 

78 logger.exception( 

79 f"Fallback parsing also failed for completed_at: {completed_at_str}" 

80 ) 

81 # Fall back to current time 

82 end_time = datetime.now(UTC) 

83 else: 

84 # Use current time if no completed_at provided 

85 end_time = datetime.now(UTC) 

86 # Ensure end_time is UTC. 

87 end_time = end_time.astimezone(UTC) 

88 

89 start_time = None 

90 try: 

91 # Proper parsing of ISO format 

92 if "T" in created_at_str: # ISO format with T separator 

93 start_time = datetime.fromisoformat(created_at_str) 

94 else: # Older format without T 

95 # Try different formats 

96 try: 

97 start_time = datetime.strptime( 

98 created_at_str, "%Y-%m-%d %H:%M:%S.%f" 

99 ) 

100 except ValueError: 

101 try: 

102 start_time = datetime.strptime( 

103 created_at_str, "%Y-%m-%d %H:%M:%S" 

104 ) 

105 except ValueError: 

106 # Last resort fallback 

107 start_time = datetime.fromisoformat( 

108 created_at_str.replace(" ", "T") 

109 ) 

110 except Exception: 

111 logger.exception("Error parsing created_at timestamp") 

112 # Fallback method if parsing fails 

113 try: 

114 from dateutil import parser 

115 

116 start_time = parser.parse(created_at_str) 

117 except Exception: 

118 logger.exception( 

119 f"Fallback parsing also failed for created_at: {created_at_str}" 

120 ) 

121 return None 

122 

123 # Calculate duration if both timestamps are valid 

124 if start_time and end_time: 124 ↛ 130line 124 didn't jump to line 130 because the condition on line 124 was always true

125 try: 

126 return int((end_time - start_time).total_seconds()) 

127 except Exception: 

128 logger.exception("Error calculating duration") 

129 

130 return None 

131 

132 

133def get_logs_for_research(research_id, username, limit: int | None = None): 

134 """ 

135 Retrieve logs for a specific research ID, oldest-first. 

136 

137 Args: 

138 research_id: ID of the research 

139 username: Authenticated user whose encrypted DB to open. 

140 limit: If set, return only the most recent ``limit`` entries 

141 (still oldest-first in the returned list). Used to bound the 

142 response size of ``/history/logs/<id>`` — long langgraph 

143 researches can produce thousands of 10 KB rows that the 

144 frontend prunes to 500 anyway. 

145 

146 Returns: 

147 List of log entries as dictionaries 

148 """ 

149 try: 

150 with get_user_db_session(username) as session: 

151 query = session.query(ResearchLog).filter( 

152 ResearchLog.research_id == research_id 

153 ) 

154 if limit is not None: 

155 # Take the newest ``limit`` rows from the DB, then flip back 

156 # to oldest-first ordering for the caller. Avoids loading 

157 # the entire table when only the tail is wanted. ``id`` is the 

158 # tie-break: timestamps are not unique, so without it the rows 

159 # surviving ``.limit()`` at a shared-timestamp boundary would 

160 # be SQL-undefined. 

161 log_results = list( 

162 reversed( 

163 query.order_by( 

164 ResearchLog.timestamp.desc(), 

165 ResearchLog.id.desc(), 

166 ) 

167 .limit(limit) 

168 .all() 

169 ) 

170 ) 

171 else: 

172 log_results = query.order_by( 

173 ResearchLog.timestamp.asc(), ResearchLog.id.asc() 

174 ).all() 

175 

176 logs = [] 

177 for result in log_results: 

178 # Convert entry for frontend consumption 

179 formatted_entry = { 

180 "time": result.timestamp, 

181 "message": result.message, 

182 "type": result.level, 

183 "module": result.module, 

184 "line_no": result.line_no, 

185 } 

186 logs.append(formatted_entry) 

187 

188 return logs 

189 except Exception: 

190 logger.exception("Error retrieving logs from database") 

191 return [] 

192 

193 

194@logger.catch 

195def get_total_logs_for_research(research_id, username): 

196 """ 

197 Returns the total number of logs for a given `research_id`. 

198 

199 Args: 

200 research_id (int): The ID of the research. 

201 username (str): Authenticated user whose encrypted DB to open. 

202 

203 Returns: 

204 int: Total number of logs for the specified research ID. 

205 """ 

206 with get_user_db_session(username) as session: 

207 return ( 

208 session.query(ResearchLog) 

209 .filter(ResearchLog.research_id == research_id) 

210 .count() 

211 )