Coverage for nexusLIMS/tui/apps/db_browser.py: 100%

101 statements  

« prev     ^ index     » next       coverage.py v7.11.3, created at 2026-03-24 05:23 +0000

1"""Read-only NexusLIMS database browser TUI. 

2 

3Provides :class:`NexusLIMSDBApp`, a Textual app that wraps Squall's 

4``SQLiteClientApp`` to give a stripped-down, read-only view of the database: 

5 

6* The "Open Database" button and "Execute SQL" tab are removed. 

7* The Table Viewer tab gains a live filter input and click-to-sort column 

8 headers, both implemented via parameterised SQL queries so they compose 

9 correctly and never expose the database to modification. 

10 

11Usage 

12----- 

13 

14.. code-block:: python 

15 

16 from argparse import Namespace 

17 from nexusLIMS.tui.apps.db_browser import NexusLIMSDBApp 

18 

19 app = NexusLIMSDBApp(Namespace(filepath="/path/to/nexuslims.db")) 

20 app.run() 

21""" 

22 

23from __future__ import annotations 

24 

25import sqlite3 

26from pathlib import Path 

27from typing import TYPE_CHECKING 

28 

29from squall import db_utility 

30from squall.database_structure_tree import DatabaseStructurePane 

31from squall.squall import SQLiteClientApp 

32from textual import on 

33from textual.widgets import DataTable, Input, Select, TabbedContent, TabPane 

34from textual.widgets._select import NULL as SELECT_BLANK 

35 

36if TYPE_CHECKING: 

37 from textual.app import ComposeResult 

38 

39# --------------------------------------------------------------------------- 

40# Resolve squall's stylesheet at import time so the subclass can reference it 

41# as a class variable regardless of where *this* module lives on disk. 

42# --------------------------------------------------------------------------- 

43import squall as _squall_pkg 

44 

45_SQUALL_TCSS = str(Path(_squall_pkg.__file__).parent / "squall.tcss") 

46 

47 

48class NexusLIMSTableViewerPane(TabPane): 

49 """Table viewer with live full-text filter and sortable column headers. 

50 

51 All filtering and sorting is performed via parameterised SQL queries 

52 so they compose naturally and the database is never modified. 

53 """ 

54 

55 DEFAULT_CSS = """ 

56 NexusLIMSTableViewerPane { 

57 Select { 

58 margin: 1; 

59 border: round gold; 

60 } 

61 

62 #filter_input { 

63 margin: 1; 

64 border: round $accent; 

65 width: 100%; 

66 } 

67 

68 DataTable { 

69 margin: 1; 

70 border: round gold; 

71 height: 1fr; 

72 } 

73 } 

74 """ 

75 

76 def __init__(self, db_path: Path, *args, **kwargs) -> None: 

77 super().__init__(*args, **kwargs) 

78 self.db_path = db_path 

79 self.tables: list[str] = sorted(db_utility.get_table_names(db_path)) 

80 self._filter_text: str = "" 

81 self._sort_col: str | None = None 

82 self._sort_asc: bool = True 

83 self._col_names: list[str] = [] 

84 

85 def compose(self) -> ComposeResult: 

86 """Build the table viewer layout.""" 

87 yield Select.from_values( 

88 self.tables, id="table_names_select", value=self.tables[0] 

89 ) 

90 yield Input( 

91 placeholder="Filter rows… (searches all columns)", 

92 id="filter_input", 

93 ) 

94 yield DataTable(id="sqlite_table_data") 

95 

96 def on_mount(self) -> None: 

97 """Load the initial table on mount.""" 

98 # Inline style wins over squall's CSS_PATH rule (Input { width: 80% }) 

99 self.query_one("#filter_input", Input).styles.width = "100%" 

100 self._load_table() 

101 

102 # ------------------------------------------------------------------ 

103 # Event handlers 

104 # ------------------------------------------------------------------ 

105 

106 @on(Select.Changed, "#table_names_select") 

107 def _on_table_changed(self, event: Select.Changed) -> None: 

108 """Reset filter and sort state when the selected table changes.""" 

109 if event.value is SELECT_BLANK: 

110 return 

111 self._filter_text = "" 

112 self._sort_col = None 

113 self._sort_asc = True 

114 self.query_one("#filter_input", Input).value = "" 

115 self._load_table() 

116 

117 @on(Input.Changed, "#filter_input") 

118 def _on_filter_changed(self, event: Input.Changed) -> None: 

119 """Re-query the database whenever the filter text changes.""" 

120 self._filter_text = event.value 

121 self._load_table() 

122 

123 @on(DataTable.HeaderSelected, "#sqlite_table_data") 

124 def _on_header_selected(self, event: DataTable.HeaderSelected) -> None: 

125 """Cycle sort direction on repeated clicks; change column otherwise.""" 

126 clicked_col = event.column_key.value # str key we supplied to add_column 

127 if self._sort_col == clicked_col: 

128 if self._sort_asc: 

129 self._sort_asc = False 

130 else: 

131 # Third click on the same column clears sorting 

132 self._sort_col = None 

133 self._sort_asc = True 

134 else: 

135 self._sort_col = clicked_col 

136 self._sort_asc = True 

137 self._load_table() 

138 

139 # ------------------------------------------------------------------ 

140 # Internal helpers 

141 # ------------------------------------------------------------------ 

142 

143 def _load_table(self) -> None: 

144 """Fetch rows from SQLite and populate the DataTable. 

145 

146 Builds a single parameterised ``SELECT`` statement that combines the 

147 current filter text (``WHERE … LIKE ?``) and sort column 

148 (``ORDER BY … ASC/DESC``) so both features compose correctly. 

149 """ 

150 table_name = str(self.app.query_one("#table_names_select", Select).value) 

151 filter_text = self._filter_text.strip() 

152 

153 conn = sqlite3.connect(self.db_path) 

154 cursor = conn.cursor() 

155 

156 # Discover column names without fetching any data 

157 cursor.execute(f"SELECT * FROM {table_name} LIMIT 0") # noqa: S608 

158 col_names = [d[0] for d in cursor.description] 

159 self._col_names = col_names 

160 

161 # Build WHERE clause — cast every column to TEXT so LIKE works on 

162 # integers, datetimes, etc. 

163 params: list[str] = [] 

164 where = "" 

165 if filter_text: 

166 conditions = [f"CAST({col} AS TEXT) LIKE ?" for col in col_names] 

167 where = "WHERE " + " OR ".join(conditions) 

168 params = [f"%{filter_text}%" for _ in col_names] 

169 

170 # Build ORDER BY clause 

171 order = "" 

172 if self._sort_col and self._sort_col in col_names: 

173 direction = "ASC" if self._sort_asc else "DESC" 

174 order = f"ORDER BY {self._sort_col} {direction}" 

175 

176 sql = f"SELECT * FROM {table_name} {where} {order} LIMIT 1000" # noqa: S608 

177 cursor.execute(sql, params) 

178 rows = cursor.fetchall() 

179 conn.close() 

180 

181 # Build display labels: add ↑/↓ to the sorted column header 

182 display_labels = [self._sort_label(name) for name in col_names] 

183 

184 dt = self.query_one("#sqlite_table_data", DataTable) 

185 dt.clear(columns=True) 

186 for name, label in zip(col_names, display_labels): 

187 dt.add_column(label, key=name) 

188 

189 if rows: 

190 dt.add_rows(rows) 

191 else: 

192 dt.add_rows([tuple("" for _ in col_names)]) 

193 

194 dt.cursor_type = "row" 

195 

196 def _sort_label(self, col_name: str) -> str: 

197 """Return the display label for a column header, with sort indicator.""" 

198 if col_name == self._sort_col: 

199 indicator = " ↑" if self._sort_asc else " ↓" 

200 return f"{col_name}{indicator}" 

201 return col_name 

202 

203 

204class NexusLIMSDBApp(SQLiteClientApp): 

205 """Read-only NexusLIMS database browser. 

206 

207 Extends Squall's ``SQLiteClientApp`` with: 

208 

209 * The "Open Database" button hidden (path always comes from ``NX_DB_PATH``) 

210 * The "Execute SQL" tab removed to prevent accidental modifications 

211 * The Table Viewer replaced with :class:`NexusLIMSTableViewerPane`, which 

212 adds a live filter input and sortable column headers 

213 """ 

214 

215 CSS_PATH = _SQUALL_TCSS 

216 DEFAULT_CSS = "#center { display: none; }" 

217 

218 async def update_ui(self, db_file_path: Path) -> None: 

219 """Load the database into the read-only browser tabs.""" 

220 if not Path(db_file_path).exists(): 

221 self.notify("Database not found") 

222 return 

223 

224 tabbed_content = self.query_one("#tabbed_ui", TabbedContent) 

225 await tabbed_content.clear_panes() 

226 await tabbed_content.add_pane( 

227 NexusLIMSTableViewerPane(db_file_path, title="Table Viewer") 

228 ) 

229 await tabbed_content.add_pane( 

230 DatabaseStructurePane( 

231 db_file_path, 

232 title="Database Structure", 

233 id="db_structure", 

234 ) 

235 ) 

236 self.title = f"NexusLIMS DB — {db_file_path}"