Fix SQL query handling to allow trailing semicolons (bug from copilot fix) and improve error checking

Author: jairus-m

src/dbt_mcp/dbt_admin/run_artifacts/store.py

@@ -337,10 +337,12 @@ def describe_table(self, table_name: str) -> list[dict[str, str]]:
 
     def query(self, sql: str) -> list[dict[str, Any]]:
         """Execute a read-only SQL query. Results capped at 500 rows."""
-        sanitized = _strip_sql_comments(sql)
+        sanitized = _strip_sql_comments(sql).strip()
+        if sanitized.endswith(";"):
+            sanitized = sanitized[:-1]
         if ";" in sanitized:
             raise ArtifactQueryError("Multi-statement queries are not allowed.")
-        tokens = sanitized.strip().upper().split()
+        tokens = sanitized.upper().split()
         for token in tokens:
             if token in READONLY_BLOCKED:
                 raise ArtifactQueryError(
@@ -371,14 +373,19 @@ def search(
 
         fts_table = f"fts_main_{table_name}"
         sql = f"""
-            SELECT t.*, {fts_table}.match_bm25(t.id, ?) AS fts_score
+            WITH scored AS (
+                SELECT id, {fts_table}.match_bm25(id, ?) AS fts_score
+                FROM "{table_name}"
+            )
+            SELECT t.*, s.fts_score
             FROM "{table_name}" t
-            WHERE {fts_table}.match_bm25(t.id, ?) IS NOT NULL
-            ORDER BY fts_score DESC
+            JOIN scored s ON t.id = s.id
+            WHERE s.fts_score IS NOT NULL
+            ORDER BY s.fts_score DESC
             LIMIT ?
         """
         try:
-            result = self.conn.execute(sql, [query_text, query_text, limit])
+            result = self.conn.execute(sql, [query_text, limit])
             columns = [desc[0] for desc in result.description]
             rows = result.fetchall()
             return [

tests/unit/dbt_admin/run_artifacts/test_store.py

@@ -141,6 +141,11 @@ def test_comment_bypass_blocked(self, store: ArtifactStore) -> None:
         with pytest.raises(ArtifactQueryError, match="Multi-statement"):
             store.query("SELECT 1;/**/DROP TABLE nodes")
 
+    def test_trailing_semicolon_is_allowed(self, store: ArtifactStore) -> None:
+        store._ensure_tables_created()
+        result = store.query("SELECT 1 AS n;")
+        assert result == [{"n": 1}]
+
     def test_invalid_sql_raises(self, store: ArtifactStore) -> None:
         with pytest.raises(ArtifactQueryError, match="Query failed"):
             store.query("SELECT * FROM table_that_does_not_exist_xyz")