Fix type errors caught by ty check

- Add type: ignore comments for external library type stub limitations
  (csv.reader, click.progressbar, IOBase.name, Callable.__name__)
- Change Iterable to Sequence for SQL where_args parameters
- Use db.table() instead of db[name] for proper Table return type
- Fix rebuild_fts return type from None to Table
- Update test_tracer to expect fewer queries (optimization side effect)

🤖 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 21:38:54 -08:00
commit 9b1daa346a
4 changed files with 19 additions and 20 deletions

View file

@ -1051,7 +1051,7 @@ def insert_upsert_implementation(
csv_reader_args["delimiter"] = delimiter csv_reader_args["delimiter"] = delimiter
if quotechar: if quotechar:
csv_reader_args["quotechar"] = quotechar csv_reader_args["quotechar"] = quotechar
reader = csv_std.reader(decoded, **csv_reader_args) reader = csv_std.reader(decoded, **csv_reader_args) # type: ignore
first_row = next(reader) first_row = next(reader)
if no_headers: if no_headers:
headers = ["untitled_{}".format(i + 1) for i in range(len(first_row))] headers = ["untitled_{}".format(i + 1) for i in range(len(first_row))]

View file

@ -489,7 +489,7 @@ class Database:
""" """
def register(fn: Callable) -> Callable: def register(fn: Callable) -> Callable:
fn_name = name or fn.__name__ fn_name = name or fn.__name__ # type: ignore
arity = len(inspect.signature(fn).parameters) arity = len(inspect.signature(fn).parameters)
if not replace and (fn_name, arity) in self._registered_functions: if not replace and (fn_name, arity) in self._registered_functions:
return fn return fn
@ -1450,11 +1450,11 @@ class Queryable:
def pks_and_rows_where( def pks_and_rows_where(
self, self,
where: Optional[str] = None, where: Optional[str] = None,
where_args: Optional[Union[Iterable, dict]] = None, where_args: Optional[Union[Sequence, Dict[str, Any]]] = None,
order_by: Optional[str] = None, order_by: Optional[str] = None,
limit: Optional[int] = None, limit: Optional[int] = None,
offset: Optional[int] = None, offset: Optional[int] = None,
) -> Generator[Tuple[Any, Dict], None, None]: ) -> Generator[Tuple[Any, Dict[str, Any]], None, None]:
""" """
Like ``.rows_where()`` but returns ``(pk, row)`` pairs - ``pk`` can be a single value or tuple. Like ``.rows_where()`` but returns ``(pk, row)`` pairs - ``pk`` can be a single value or tuple.
@ -1848,7 +1848,7 @@ class Table(Queryable):
quote_identifier(self.name), quote_identifier(self.name),
) )
self.db.execute(sql) self.db.execute(sql)
return self.db[new_name] return self.db.table(new_name)
def transform( def transform(
self, self,
@ -2153,7 +2153,7 @@ class Table(Queryable):
) )
) )
table = table or "_".join(columns) table = table or "_".join(columns)
lookup_table = self.db[table] lookup_table = self.db.table(table)
fk_column = fk_column or "{}_id".format(table) fk_column = fk_column or "{}_id".format(table)
magic_lookup_column = "{}_{}".format(fk_column, os.urandom(6).hex()) magic_lookup_column = "{}_{}".format(fk_column, os.urandom(6).hex())
@ -2680,7 +2680,7 @@ class Table(Queryable):
) )
return self return self
def rebuild_fts(self) -> None: def rebuild_fts(self) -> "Table":
"Run the ``rebuild`` operation against the associated full-text search index table." "Run the ``rebuild`` operation against the associated full-text search index table."
fts_table = self.detect_fts() fts_table = self.detect_fts()
if fts_table is None: if fts_table is None:
@ -2767,7 +2767,7 @@ class Table(Queryable):
self.name self.name
) )
fts_table_quoted = quote_identifier(fts_table) fts_table_quoted = quote_identifier(fts_table)
virtual_table_using = self.db[fts_table].virtual_table_using virtual_table_using = self.db.table(fts_table).virtual_table_using
sql = textwrap.dedent( sql = textwrap.dedent(
""" """
with {original} as ( with {original} as (
@ -2887,7 +2887,7 @@ class Table(Queryable):
def delete_where( def delete_where(
self, self,
where: Optional[str] = None, where: Optional[str] = None,
where_args: Optional[Union[Iterable, dict]] = None, where_args: Optional[Union[Sequence, Dict[str, Any]]] = None,
analyze: bool = False, analyze: bool = False,
) -> "Table": ) -> "Table":
""" """
@ -2978,9 +2978,9 @@ class Table(Queryable):
drop: bool = False, drop: bool = False,
multi: bool = False, multi: bool = False,
where: Optional[str] = None, where: Optional[str] = None,
where_args: Optional[Union[Iterable, dict]] = None, where_args: Optional[Union[Sequence, Dict[str, Any]]] = None,
show_progress: bool = False, show_progress: bool = False,
): ) -> "Table":
""" """
Apply conversion function ``fn`` to every value in the specified columns. Apply conversion function ``fn`` to every value in the specified columns.
@ -3143,7 +3143,7 @@ class Table(Queryable):
if has_extracts: if has_extracts:
for i, key in enumerate(all_columns): for i, key in enumerate(all_columns):
if key in extracts: if key in extracts:
record_values[i] = self.db[extracts[key]].lookup( record_values[i] = self.db.table(extracts[key]).lookup(
{"value": record_values[i]} {"value": record_values[i]}
) )
values.append(record_values) values.append(record_values)
@ -3164,7 +3164,7 @@ class Table(Queryable):
) )
if key in extracts: if key in extracts:
extract_table = extracts[key] extract_table = extracts[key]
value = self.db[extract_table].lookup({"value": value}) value = self.db.table(extract_table).lookup({"value": value})
record_values.append(value) record_values.append(value)
values.append(record_values) values.append(record_values)
@ -3874,7 +3874,7 @@ class Table(Queryable):
already exists. already exists.
""" """
if isinstance(other_table, str): if isinstance(other_table, str):
other_table = cast(Table, self.db.table(other_table, pk=pk)) other_table = self.db.table(other_table, pk=pk)
our_id = self.last_pk our_id = self.last_pk
if lookup is not None: if lookup is not None:
assert record_or_iterable is None, "Provide lookup= or record, not both" assert record_or_iterable is None, "Provide lookup= or record, not both"

View file

@ -237,8 +237,8 @@ def file_progress(
if fileno == 0: # 0 means stdin if fileno == 0: # 0 means stdin
yield file yield file
else: else:
file_length = os.path.getsize(file.name) file_length = os.path.getsize(file.name) # type: ignore
with click.progressbar(length=file_length, **kwargs) as bar: with click.progressbar(length=file_length, **kwargs) as bar: # type: ignore
yield UpdateWrapper(file, bar.update) yield UpdateWrapper(file, bar.update)
@ -516,7 +516,7 @@ def progressbar(*args: Iterable[T], **kwargs: Any) -> Generator[Any, None, None]
if silent: if silent:
yield NullProgressBar(*args) yield NullProgressBar(*args)
else: else:
with click.progressbar(*args, **kwargs) as bar: with click.progressbar(*args, **kwargs) as bar: # type: ignore
yield bar yield bar

View file

@ -50,7 +50,7 @@ def test_with_tracer():
with db.tracer(tracer): with db.tracer(tracer):
list(dogs.search("Cleopaws")) list(dogs.search("Cleopaws"))
assert len(collected) == 5 assert len(collected) == 4
assert collected == [ assert collected == [
( (
"SELECT name FROM sqlite_master\n" "SELECT name FROM sqlite_master\n"
@ -70,7 +70,6 @@ def test_with_tracer():
}, },
), ),
("select name from sqlite_master where type = 'view'", None), ("select name from sqlite_master where type = 'view'", None),
("select name from sqlite_master where type = 'view'", None),
("select sql from sqlite_master where name = ?", ("dogs_fts",)), ("select sql from sqlite_master where name = ?", ("dogs_fts",)),
( (
'with "original" as (\n' 'with "original" as (\n'
@ -94,4 +93,4 @@ def test_with_tracer():
# Outside the with block collected should not be appended to # Outside the with block collected should not be appended to
dogs.insert({"name": "Cleopaws"}) dogs.insert({"name": "Cleopaws"})
assert len(collected) == 5 assert len(collected) == 4