Coverage for src/local_deep_research/storage/database.py: 96%
72 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Database-based report storage implementation."""
3from typing import Dict, Any, List, Optional
4from loguru import logger
5from sqlalchemy.orm import Session
7from .base import ReportStorage
8from ..database.models import ResearchHistory
9from ..security import strip_settings_snapshot
12class DatabaseReportStorage(ReportStorage):
13 """Store reports in the database with caching support."""
15 def __init__(self, session: Session):
16 """Initialize database storage.
18 Args:
19 session: SQLAlchemy database session
20 """
21 self.session = session
23 def save_report(
24 self,
25 research_id: str,
26 content: str,
27 metadata: Optional[Dict[str, Any]] = None,
28 username: Optional[str] = None,
29 ) -> bool:
30 """Save report to database."""
31 try:
32 research = (
33 self.session.query(ResearchHistory)
34 .filter_by(id=research_id)
35 .first()
36 )
38 if not research:
39 logger.error(f"Research {research_id} not found")
40 return False
42 research.report_content = content # type: ignore[assignment]
44 if metadata:
45 if research.research_meta:
46 research.research_meta.update(metadata) # type: ignore[union-attr]
47 else:
48 research.research_meta = metadata # type: ignore[assignment]
50 self.session.commit()
51 logger.info(f"Saved report for research {research_id} to database")
52 return True
54 except Exception:
55 logger.exception("Error saving report to database")
56 self.session.rollback()
57 return False
59 def get_report(
60 self, research_id: str, username: Optional[str] = None
61 ) -> Optional[str]:
62 """Return raw ``report_content`` for a research row.
64 IMPORTANT: ``report_content`` is the answer body only — the
65 assembled "## Sources" / "## Research Metrics" sections live in
66 ``research_resources`` and the
67 per-research metrics tables and are stitched in by
68 ``web.services.report_assembly_service.assemble_full_report``.
70 Do **not** use this method for any user-facing display path.
71 Call ``assemble_full_report(research, db_session)`` instead so
72 legacy rows (which embed sources/metrics inline in
73 ``report_content``) and new rows render identically. Current
74 callers of this method use the truncated content for
75 notification teasers / summary previews where answer-only is
76 the desired shape.
77 """
78 try:
79 research = (
80 self.session.query(ResearchHistory)
81 .filter_by(id=research_id)
82 .first()
83 )
85 if not research or not research.report_content:
86 return None
88 return research.report_content # type: ignore[return-value]
90 except Exception:
91 logger.exception("Error getting report from database")
92 return None
94 def get_report_with_metadata(
95 self, research_id: str, username: Optional[str] = None
96 ) -> Optional[Dict[str, Any]]:
97 """Return ``report_content`` + research metadata for a row.
99 Same answer-only caveat as :meth:`get_report` — see that
100 docstring before using ``["content"]`` for any user-facing
101 rendering path; prefer ``assemble_full_report`` instead.
102 """
103 try:
104 research = (
105 self.session.query(ResearchHistory)
106 .filter_by(id=research_id)
107 .first()
108 )
110 if not research or not research.report_content:
111 return None
113 return {
114 "content": research.report_content,
115 # Strip settings_snapshot (API keys/tokens) before exposing
116 # research_meta — this method's contract is to hand back
117 # metadata, so a future route wiring it to a response must not
118 # leak the snapshot (CWE-200). Defence-in-depth at the source,
119 # mirroring the report/details routes. Other fields preserved.
120 "metadata": strip_settings_snapshot(research.research_meta),
121 "query": research.query,
122 "mode": research.mode,
123 "created_at": research.created_at,
124 "completed_at": research.completed_at,
125 "duration_seconds": research.duration_seconds,
126 }
128 except Exception:
129 logger.exception("Error getting report with metadata")
130 return None
132 def list_reports(
133 self, username: Optional[str] = None
134 ) -> List[Dict[str, Any]]:
135 """List reports from database."""
136 try:
137 # Select only the metadata columns the listing returns —
138 # never load the large ``report_content`` body, which would
139 # pull every report into memory at once (#4560). Filtering on
140 # the column does not load it.
141 query = self.session.query(
142 ResearchHistory.id,
143 ResearchHistory.query,
144 ResearchHistory.mode,
145 ResearchHistory.created_at,
146 ResearchHistory.completed_at,
147 ).filter(ResearchHistory.report_content.isnot(None))
148 results = query.all()
149 return [
150 {
151 "id": r.id,
152 "query": r.query,
153 "mode": r.mode,
154 "created_at": r.created_at,
155 "completed_at": r.completed_at,
156 }
157 for r in results
158 ]
159 except Exception:
160 logger.exception("Error listing reports from database")
161 return []
163 def delete_report(
164 self, research_id: str, username: Optional[str] = None
165 ) -> bool:
166 """Delete report from database."""
167 try:
168 research = (
169 self.session.query(ResearchHistory)
170 .filter_by(id=research_id)
171 .first()
172 )
174 if not research:
175 return False
177 research.report_content = None # type: ignore[assignment]
178 self.session.commit()
180 return True
182 except Exception:
183 logger.exception("Error deleting report")
184 self.session.rollback()
185 return False
187 def report_exists(
188 self, research_id: str, username: Optional[str] = None
189 ) -> bool:
190 """Check if report exists in database."""
191 try:
192 research = (
193 self.session.query(ResearchHistory)
194 .filter_by(id=research_id)
195 .first()
196 )
198 return research is not None and research.report_content is not None
200 except Exception:
201 logger.exception("Error checking if report exists")
202 return False