From da739fbaba3d32ec4ff6070ee41f5f0cdb8b94ad Mon Sep 17 00:00:00 2001 From: Simon Willison Date: Tue, 16 Dec 2025 16:51:58 -0800 Subject: [PATCH] Fix remaining type warnings in sqlite_utils package MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add assert for sniff_buffer not being None - Handle cursor.fetchone() potentially returning None - Use db.table() for counts_table and index_foreign_keys - Add type: ignore for cursor union type in raw mode 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- sqlite_utils/cli.py | 4 +++- sqlite_utils/db.py | 4 ++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/sqlite_utils/cli.py b/sqlite_utils/cli.py index 18d29f9..f2c6f35 100644 --- a/sqlite_utils/cli.py +++ b/sqlite_utils/cli.py @@ -1037,6 +1037,7 @@ def insert_upsert_implementation( if csv or tsv: if sniff: # Read first 2048 bytes and use that to detect + assert sniff_buffer is not None first_bytes = sniff_buffer.peek(2048) dialect = csv_std.Sniffer().sniff( first_bytes.decode(encoding, "ignore") @@ -2120,7 +2121,8 @@ def _execute_query( else: headers = [c[0] for c in cursor.description] if raw: - data = cursor.fetchone()[0] + row = cursor.fetchone() # type: ignore[union-attr] + data = row[0] if row else None if isinstance(data, bytes): sys.stdout.buffer.write(data) else: diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index b53b027..0f720ef 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -819,7 +819,7 @@ class Database: tables = [table for table in self.tables if table.has_counts_triggers] with self.conn: self._ensure_counts_table() - counts_table = self[self._counts_table_name] + counts_table = self.table(self._counts_table_name) counts_table.delete_where() counts_table.insert_all( {"table": table.name, "count": table.execute_count()} @@ -1275,7 +1275,7 @@ class Database: def index_foreign_keys(self): "Create indexes for every foreign key column on every table in the database." for table_name in self.table_names(): - table = self[table_name] + table = self.table(table_name) existing_indexes = { i.columns[0] for i in table.indexes if len(i.columns) == 1 }