remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-09-11 19:14:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fix remaining type warnings in sqlite_utils package

Browse source 
- 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 <noreply@anthropic.com>
This commit is contained in:
Simon Willison 2025-12-16 16:51:58 -08:00
commit da739fbaba
2 changed files with 5 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

4
sqlite_utils/cli.py
View file

@ -1037,6 +1037,7 @@ def insert_upsert_implementation(
if csv or tsv: if csv or tsv:
if sniff: if sniff:
# Read first 2048 bytes and use that to detect # Read first 2048 bytes and use that to detect
assert sniff_buffer is not None
first_bytes = sniff_buffer.peek(2048) first_bytes = sniff_buffer.peek(2048)
dialect = csv_std.Sniffer().sniff( dialect = csv_std.Sniffer().sniff(
first_bytes.decode(encoding, "ignore") first_bytes.decode(encoding, "ignore")
@ -2120,7 +2121,8 @@ def _execute_query(
else: else:
headers = [c[0] for c in cursor.description] headers = [c[0] for c in cursor.description]
if raw: if raw:
data = cursor.fetchone()[0] row = cursor.fetchone() # type: ignore[union-attr]
data = row[0] if row else None
if isinstance(data, bytes): if isinstance(data, bytes):
sys.stdout.buffer.write(data) sys.stdout.buffer.write(data)
else: else:

4
sqlite_utils/db.py
View file

@ -819,7 +819,7 @@ class Database:
tables = [table for table in self.tables if table.has_counts_triggers] tables = [table for table in self.tables if table.has_counts_triggers]
with self.conn: with self.conn:
self._ensure_counts_table() 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.delete_where()
counts_table.insert_all( counts_table.insert_all(
{"table": table.name, "count": table.execute_count()} {"table": table.name, "count": table.execute_count()}
@ -1275,7 +1275,7 @@ class Database:
def index_foreign_keys(self): def index_foreign_keys(self):
"Create indexes for every foreign key column on every table in the database." "Create indexes for every foreign key column on every table in the database."
for table_name in self.table_names(): for table_name in self.table_names():
table = self[table_name] table = self.table(table_name)
existing_indexes = { existing_indexes = {
i.columns[0] for i in table.indexes if len(i.columns) == 1 i.columns[0] for i in table.indexes if len(i.columns) == 1
} }