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
« prev ^ index » next coverage.py v7.11.3, created at 2026-03-24 05:23 +0000
1"""Read-only NexusLIMS database browser TUI.
3Provides :class:`NexusLIMSDBApp`, a Textual app that wraps Squall's
4``SQLiteClientApp`` to give a stripped-down, read-only view of the database:
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.
11Usage
12-----
14.. code-block:: python
16 from argparse import Namespace
17 from nexusLIMS.tui.apps.db_browser import NexusLIMSDBApp
19 app = NexusLIMSDBApp(Namespace(filepath="/path/to/nexuslims.db"))
20 app.run()
21"""
23from __future__ import annotations
25import sqlite3
26from pathlib import Path
27from typing import TYPE_CHECKING
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
36if TYPE_CHECKING:
37 from textual.app import ComposeResult
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
45_SQUALL_TCSS = str(Path(_squall_pkg.__file__).parent / "squall.tcss")
48class NexusLIMSTableViewerPane(TabPane):
49 """Table viewer with live full-text filter and sortable column headers.
51 All filtering and sorting is performed via parameterised SQL queries
52 so they compose naturally and the database is never modified.
53 """
55 DEFAULT_CSS = """
56 NexusLIMSTableViewerPane {
57 Select {
58 margin: 1;
59 border: round gold;
60 }
62 #filter_input {
63 margin: 1;
64 border: round $accent;
65 width: 100%;
66 }
68 DataTable {
69 margin: 1;
70 border: round gold;
71 height: 1fr;
72 }
73 }
74 """
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] = []
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")
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()
102 # ------------------------------------------------------------------
103 # Event handlers
104 # ------------------------------------------------------------------
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()
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()
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()
139 # ------------------------------------------------------------------
140 # Internal helpers
141 # ------------------------------------------------------------------
143 def _load_table(self) -> None:
144 """Fetch rows from SQLite and populate the DataTable.
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()
153 conn = sqlite3.connect(self.db_path)
154 cursor = conn.cursor()
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
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]
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}"
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()
181 # Build display labels: add ↑/↓ to the sorted column header
182 display_labels = [self._sort_label(name) for name in col_names]
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)
189 if rows:
190 dt.add_rows(rows)
191 else:
192 dt.add_rows([tuple("" for _ in col_names)])
194 dt.cursor_type = "row"
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
204class NexusLIMSDBApp(SQLiteClientApp):
205 """Read-only NexusLIMS database browser.
207 Extends Squall's ``SQLiteClientApp`` with:
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 """
215 CSS_PATH = _SQUALL_TCSS
216 DEFAULT_CSS = "#center { display: none; }"
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
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}"