remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-09-12 03:24:23 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fix remaining Ruff errors with GPT-5.6 Sol high

Browse source 
https://gist.github.com/simonw/6da7906a9fea6e90da131c21a9055199
This commit is contained in:
Simon Willison 2026-07-25 14:39:35 -07:00
commit 48ef55152c
16 changed files with 164 additions and 190 deletions
Download patch file Download diff file Expand all files Collapse all files

8
docs/conf.py
View file

@ -1,9 +1,7 @@
#!/usr/bin/env python3
import inspect
import sys
from pathlib import Path
from subprocess import PIPE, Popen, check_output
from subprocess import PIPE, CalledProcessError, Popen, check_output
# This file is execfile()d with the current directory set to its
# containing dir.
@ -49,7 +47,7 @@ extlinks = {
def _linkcode_git_ref():
try:
return check_output(["git", "rev-parse", "HEAD"]).decode("utf8").strip()
except Exception:
except (CalledProcessError, OSError):
return "main"
@ -78,7 +76,7 @@ def linkcode_resolve(domain, info):
obj = inspect.unwrap(obj)
source_file = inspect.getsourcefile(obj)
_, line_number = inspect.getsourcelines(obj)
except Exception:
except (OSError, TypeError, ValueError):
return None
if source_file is None:

42
sqlite_utils/cli.py
View file

@ -8,7 +8,7 @@ import itertools
import json
import os
import pathlib
import pdb
import pdb # noqa: T100
import sys
import textwrap
from datetime import datetime, timezone
@ -72,7 +72,7 @@ def _close_databases(ctx):
for db in ctx.meta.get("_databases_to_close", []):
try:
db.close()
except Exception:
except sqlite3.Error:
pass
@ -1840,9 +1840,7 @@ def rename_table(path, table, new_name, ignore, load_extension):
db.rename_table(table, new_name)
except sqlite3.OperationalError as ex:
if not ignore:
raise click.ClickException(
f'Table "{table}" could not be renamed. {ex!s}'
)
raise click.ClickException(f'Table "{table}" could not be renamed. {ex!s}')
@cli.command(name="drop-table")
@ -2378,9 +2376,7 @@ def search(
table_columns = table_obj.columns_dict
for c in column:
if c not in table_columns:
raise click.ClickException(
f"Table '{dbtable}' has no column '{c}"
)
raise click.ClickException(f"Table '{dbtable}' has no column '{c}")
sql = table_obj.search_sql(columns=column, order_by=order, limit=limit)
if show_sql:
click.echo(sql)
@ -2953,7 +2949,7 @@ def insert_files(
with progressbar(paths_and_relative_paths, silent=silent) as bar:
def to_insert():
for path, relative_path in bar:
for file_path, relative_path in bar:
row = {}
# content_text is special case as it considers 'encoding'
@ -2965,19 +2961,21 @@ def insert_files(
raise UnicodeDecodeErrorForPath(e, resolved)
lookups = dict(FILE_COLUMNS, content_text=_content_text)
if path == "-":
if file_path == "-":
stdin_data = sys.stdin.buffer.read()
# We only support a subset of columns for this case
lookups = {
"name": lambda p: name or "-",
"path": lambda p: name or "-",
"content": lambda p: stdin_data,
"content_text": lambda p: stdin_data.decode(
"content": lambda p, data=stdin_data: data,
"content_text": lambda p, data=stdin_data: data.decode(
encoding or "utf-8"
),
"sha256": lambda p: hashlib.sha256(stdin_data).hexdigest(),
"md5": lambda p: hashlib.md5(stdin_data).hexdigest(),
"size": lambda p: len(stdin_data),
"sha256": lambda p, data=stdin_data: hashlib.sha256(
data
).hexdigest(),
"md5": lambda p, data=stdin_data: hashlib.md5(data).hexdigest(),
"size": lambda p, data=stdin_data: len(data),
}
for coldef in column:
if ":" in coldef:
@ -2985,7 +2983,7 @@ def insert_files(
else:
colname, coltype = coldef, coldef
try:
value = lookups[coltype](path)
value = lookups[coltype](file_path)
row[colname] = value
except KeyError:
raise click.ClickException(
@ -3314,7 +3312,7 @@ def convert(
def wrapped_fn(value):
try:
return fn_(value)
except Exception as ex:
except Exception as ex: # noqa: BLE001
print("\nException raised, dropping into pdb...:", ex)
pdb.post_mortem(ex.__traceback__)
sys.exit(1)
@ -3477,7 +3475,7 @@ def _load_migration_sets(files):
"__file__": str(filepath),
"__name__": "__sqlite_utils_migration__",
}
exec(code, namespace)
exec(code, namespace) # noqa: S102
migration_sets.extend(
obj for obj in namespace.values() if _compatible_migration_set(obj)
)
@ -3587,9 +3585,7 @@ def migrate(db_path, migrations, stop_before, list_, verbose):
names = {m.name for m in migration_set.pending(db)}
names.update(m.name for m in migration_set.applied(db))
known_names.update(names)
known_names.update(
f"{migration_set.name}:{name}" for name in names
)
known_names.update(f"{migration_set.name}:{name}" for name in names)
unknown = [value for value in stop_before if value not in known_names]
if unknown:
raise click.ClickException(
@ -3766,7 +3762,7 @@ def _register_functions(db, functions):
sqlite3.enable_callback_tracebacks(True)
globals = {}
try:
exec(functions, globals)
exec(functions, globals) # noqa: S102
except SyntaxError as ex:
raise click.ClickException(f"Error in functions definition: {ex}")
# Register all callables in the locals dict:
@ -3792,7 +3788,7 @@ def _rows_from_code(code):
raise click.ClickException(f"File not found: {code}")
namespace = {}
try:
exec(code, namespace)
exec(code, namespace) # noqa: S102
except SyntaxError as ex:
raise click.ClickException(f"Error in --code: {ex}")
rows = namespace.get("rows")

121
sqlite_utils/db.py
View file

@ -15,6 +15,7 @@ import uuid
from collections import namedtuple
from collections.abc import Callable, Generator, Iterable, Mapping, Sequence
from dataclasses import dataclass, field
from types import TracebackType
from typing import (
Any,
Union,
@ -272,20 +273,20 @@ class TransformError(Exception):
# A single column name, or a tuple of columns for a compound foreign key
ForeignKeyColumns = Union[str, tuple[str, ...], list[str]]
ForeignKeyColumns = str | tuple[str, ...] | list[str]
# (table, column(s), other_table, other_column(s))
ForeignKeyTuple = tuple[str, ForeignKeyColumns, str, ForeignKeyColumns]
ForeignKeyIndicator = Union[
str,
ForeignKey,
tuple[ForeignKeyColumns, str],
tuple[ForeignKeyColumns, str, ForeignKeyColumns],
ForeignKeyTuple,
]
ForeignKeyIndicator = (
str
| ForeignKey
| tuple[ForeignKeyColumns, str]
| tuple[ForeignKeyColumns, str, ForeignKeyColumns]
| ForeignKeyTuple
)
ForeignKeysType = Union[Iterable[ForeignKeyIndicator], list[ForeignKeyIndicator]]
ForeignKeysType = Iterable[ForeignKeyIndicator] | list[ForeignKeyIndicator]
class Default:
@ -580,7 +581,7 @@ class Database:
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: object | None,
exc_tb: TracebackType | None,
) -> None:
self.close()
@ -689,9 +690,7 @@ class Database:
self.conn.isolation_level = old_isolation_level
@contextlib.contextmanager
def tracer(
self, tracer: Tracer | None = None
) -> Generator["Database", None, None]:
def tracer(self, tracer: Tracer | None = None) -> Generator["Database", None, None]:
"""
Context manager to temporarily set a tracer function - all executed SQL queries will
be passed to this.
@ -1003,9 +1002,7 @@ class Database:
query += '"'
bits = _quote_fts_re.split(query)
bits = [b for b in bits if b and b != '""']
return " ".join(
f'"{bit}"' if not bit.startswith('"') else bit for bit in bits
)
return " ".join(f'"{bit}"' if not bit.startswith('"') else bit for bit in bits)
def quote_default_value(self, value: str) -> str:
if any(
@ -1099,12 +1096,10 @@ class Database:
try:
table_name = f"t{secrets.token_hex(16)}"
with self.atomic():
self.conn.execute(
f"create table {table_name} (name text) strict"
)
self.conn.execute(f"create table {table_name} (name text) strict")
self.conn.execute(f"drop table {table_name}")
self._supports_strict = True
except Exception:
except sqlite3.OperationalError:
self._supports_strict = False
return self._supports_strict
@ -1122,13 +1117,11 @@ class Database:
f"insert into {table_name} (id, name) values (1, 'one')"
)
self.conn.execute(
f"insert into {table_name} (id, name) values (1, 'two') "
"on conflict do update set name = 'two'"
f"insert into {table_name} (id, name) values (1, 'two') "
"on conflict do update set name = 'two'"
)
self._supports_on_conflict = True
except Exception:
except sqlite3.OperationalError:
self._supports_on_conflict = False
finally:
self.conn.execute(f"drop table if exists {table_name}")
@ -1262,7 +1255,7 @@ class Database:
fks.append(ForeignKey(name, fk, other_table, other_column))
continue
if not isinstance(fk, (tuple, list)):
raise ValueError(
raise ValueError( # noqa: TRY004
"foreign_keys= should be a list of tuples, "
"ForeignKey objects or column name strings"
)
@ -1459,9 +1452,7 @@ class Database:
if other_column != "rowid" and not any(
c for c in self[fk.other_table].columns if c.name == other_column
):
raise AlterError(
f"No such column: {fk.other_table}.{other_column}"
)
raise AlterError(f"No such column: {fk.other_table}.{other_column}")
column_defs = []
# ensure pk is a tuple
@ -1704,16 +1695,15 @@ class Database:
if ignore and replace:
raise ValueError("Use one or the other of ignore/replace, not both")
create_sql = f"CREATE VIEW {quote_identifier(name)} AS {sql}"
if ignore or replace:
# Does view exist already?
if name in self.view_names():
if ignore:
if (ignore or replace) and name in self.view_names():
# View exists already
if ignore:
return self
elif replace:
# If SQL is the same, do nothing
if create_sql == self[name].schema:
return self
elif replace:
# If SQL is the same, do nothing
if create_sql == self[name].schema:
return self
self[name].drop()
self[name].drop()
self.execute(create_sql)
return self
@ -2231,7 +2221,7 @@ class Table(Queryable):
row = next(iter(rows))
self.last_pk = last_pk
return row
except IndexError:
except StopIteration:
raise NotFoundError
@property
@ -2298,9 +2288,7 @@ class Table(Queryable):
for row in self.db.execute_returning_dicts(sql):
index_name = row["name"]
index_name_quoted = (
f'"{index_name}"'
if not index_name.startswith('"')
else index_name
f'"{index_name}"' if not index_name.startswith('"') else index_name
)
column_sql = f"PRAGMA index_info({index_name_quoted})"
columns = []
@ -2322,9 +2310,7 @@ class Table(Queryable):
for row in self.db.execute_returning_dicts(sql):
index_name = row["name"]
index_name_quoted = (
f'"{index_name}"'
if not index_name.startswith('"')
else index_name
f'"{index_name}"' if not index_name.startswith('"') else index_name
)
column_sql = f"PRAGMA index_xinfo({index_name_quoted})"
index_columns = []
@ -2923,9 +2909,7 @@ class Table(Queryable):
else:
lookup_table.create(
{
"id": int
,
"id": int,
**lookup_columns_definition,
},
pk="id",
@ -3034,9 +3018,7 @@ class Table(Queryable):
suffix = None
created_index_name = None
while True:
created_index_name = (
f"{index_name}_{suffix}" if suffix else index_name
)
created_index_name = f"{index_name}_{suffix}" if suffix else index_name
sql = (
textwrap.dedent("""
CREATE {unique}INDEX {if_not_exists}{index_name}
@ -3083,9 +3065,7 @@ class Table(Queryable):
if index_name not in {index.name for index in self.indexes}:
if ignore:
return self
raise OperationalError(
f"No index named {index_name} on table {self.name}"
)
raise OperationalError(f"No index named {index_name} on table {self.name}")
self.db.execute(f"DROP INDEX {quote_identifier(index_name)}")
return self
@ -3132,7 +3112,9 @@ class Table(Queryable):
col_type = str
not_null_sql = None
if not_null_default is not None:
not_null_sql = f"NOT NULL DEFAULT {self.db.quote_default_value(not_null_default)}"
not_null_sql = (
f"NOT NULL DEFAULT {self.db.quote_default_value(not_null_default)}"
)
sql = "ALTER TABLE {} ADD COLUMN {} {col_type}{not_null_default};".format(
quote_identifier(self.name),
quote_identifier(col_name),
@ -3893,9 +3875,12 @@ class Table(Queryable):
self.add_column(column_name, column_type)
# Run the updates
with progressbar(
length=self.count, silent=not show_progress, label="2: Updating"
) as bar, self.db.atomic():
with (
progressbar(
length=self.count, silent=not show_progress, label="2: Updating"
) as bar,
self.db.atomic(),
):
for pk, updates in pk_to_values.items():
self.update(pk, updates)
bar.update(1)
@ -4100,9 +4085,7 @@ class Table(Queryable):
)
for col in set_cols
),
wheres=" AND ".join(
f"{quote_identifier(pk)} = ?" for pk in pks
),
wheres=" AND ".join(f"{quote_identifier(pk)} = ?" for pk in pks),
)
queries_and_params.append(
(
@ -4400,7 +4383,7 @@ class Table(Queryable):
except StopIteration:
return self # Only headers, no data
if not isinstance(first_record, (list, tuple)):
raise ValueError(
raise ValueError( # noqa: TRY004
"After column names list, all subsequent records must also be lists"
)
else:
@ -4414,9 +4397,7 @@ class Table(Queryable):
num_columns = len(first_record.keys())
if num_columns > SQLITE_MAX_VARS:
raise ValueError(
f"Rows can have a maximum of {SQLITE_MAX_VARS} columns"
)
raise ValueError(f"Rows can have a maximum of {SQLITE_MAX_VARS} columns")
batch_size = (
1
if num_columns == 0
@ -4579,7 +4560,9 @@ class Table(Queryable):
rowid_pk = isinstance(pk, str) and pk.lower() in ROWID_ALIASES
if (hash_id or (pk and not rowid_pk)) and self.last_rowid:
# Set self.last_pk to the pk(s) for that rowid
row = next(iter(self.rows_where("rowid = ?", [self.last_rowid])))
row = next(
iter(self.rows_where("rowid = ?", [self.last_rowid]))
)
if hash_id:
self.last_pk = row[hash_id]
elif isinstance(pk, str):
@ -4751,7 +4734,7 @@ class Table(Queryable):
:param strict: Boolean, apply STRICT mode if creating the table.
"""
if not isinstance(lookup_values, dict):
raise ValueError("lookup_values must be a dictionary")
raise ValueError("lookup_values must be a dictionary") # noqa: TRY004
if pk is None:
raise ValueError("pk cannot be None")
if extra_values is not None and not isinstance(extra_values, dict):
@ -4769,9 +4752,7 @@ class Table(Queryable):
} not in unique_column_sets:
self.create_index(lookup_values.keys(), unique=True)
# IS rather than = so that null values are matched correctly
wheres = [
f"{quote_identifier(column)} IS ?" for column in lookup_values
]
wheres = [f"{quote_identifier(column)} IS ?" for column in lookup_values]
rows = list(
self.rows_where(
" and ".join(wheres), [value for _, value in lookup_values.items()]

16
sqlite_utils/utils.py
View file

@ -52,7 +52,7 @@ SPATIALITE_PATHS = (
ORIGINAL_CSV_FIELD_SIZE_LIMIT = csv.field_size_limit()
# Type alias for row dictionaries - values can be various SQLite-compatible types
RowValue = Union[None, int, float, str, bytes, bool, list[str]]
RowValue = None | int | float | str | bytes | bool | list[str]
Row = dict[str, RowValue]
T = TypeVar("T")
@ -270,9 +270,7 @@ def _extra_key_strategy(
yield cast(Row, row)
elif not extras_key:
extras = row.pop(None)
raise RowError(
f"Row {row} contained these extra values: {extras}"
)
raise RowError(f"Row {row} contained these extra values: {extras}")
else:
extras_value = row.pop(None)
row_out = cast(Row, row)
@ -449,9 +447,7 @@ class ValueTracker:
@classmethod
def get_tests(cls) -> list[str]:
return [
key.split("test_")[-1]
for key in cls.__dict__
if key.startswith("test_")
key.split("test_")[-1] for key in cls.__dict__ if key.startswith("test_")
]
def test_integer(self, value: object) -> bool:
@ -522,7 +518,7 @@ def _compile_code(
# If user defined a convert() function, return that
try:
exec(code, globals_dict)
exec(code, globals_dict) # noqa: S102
return cast(Callable[..., object], globals_dict["convert"])
except (AttributeError, SyntaxError, NameError, KeyError, TypeError):
pass
@ -533,7 +529,7 @@ def _compile_code(
fn = eval(code, globals_dict)
if callable(fn):
return cast(Callable[..., object], fn)
except Exception:
except Exception: # noqa: BLE001, S110
pass
# Try compiling their code as a function instead
@ -557,7 +553,7 @@ def _compile_code(
if code_o is None:
raise SyntaxError("Could not compile code")
exec(code_o, globals_dict)
exec(code_o, globals_dict) # noqa: S102
return cast(Callable[..., object], globals_dict["fn"])

2
tests/conftest.py
View file

@ -56,7 +56,7 @@ def close_all_databases():
for db in databases:
try:
db.close()
except Exception:
except sqlite3.Error:
pass

37
tests/test_atomic.py
View file

@ -28,11 +28,13 @@ from sqlite_utils.utils import sqlite3
END;
""",
[
("CREATE TRIGGER t_ai AFTER INSERT ON t\n"
" BEGIN\n"
" UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
" INSERT INTO log VALUES ('x;y');\n"
" END;")
(
"CREATE TRIGGER t_ai AFTER INSERT ON t\n"
" BEGIN\n"
" UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
" INSERT INTO log VALUES ('x;y');\n"
" END;"
)
],
),
),
@ -83,9 +85,8 @@ def test_outer_atomic_rolls_back_released_savepoint(fresh_db):
def test_executescript_does_not_commit_open_atomic_block(fresh_db):
with pytest.raises(RuntimeError):
with fresh_db.atomic():
fresh_db.executescript("""
with pytest.raises(RuntimeError), fresh_db.atomic():
fresh_db.executescript("""
CREATE TABLE dogs(id INTEGER PRIMARY KEY, name TEXT);
CREATE TRIGGER dogs_ai AFTER INSERT ON dogs
BEGIN
@ -94,7 +95,7 @@ def test_executescript_does_not_commit_open_atomic_block(fresh_db):
-- This comment has a semicolon;
INSERT INTO dogs VALUES (1, 'Cleo; the first');
""")
raise RuntimeError("boom")
raise RuntimeError("boom")
assert not fresh_db["dogs"].exists()
@ -349,9 +350,11 @@ def test_atomic_preserves_error_from_transaction_destroying_trigger(fresh_db):
# with "cannot rollback - no transaction is active"
fresh_db.execute("create table t (id integer primary key, v text)")
fresh_db.execute(TRIGGER_SQL)
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
with fresh_db.atomic():
fresh_db.execute("insert into t (v) values ('bad')")
with (
pytest.raises(sqlite3.IntegrityError, match="trigger says no"),
fresh_db.atomic(),
):
fresh_db.execute("insert into t (v) values ('bad')")
assert not fresh_db.conn.in_transaction
@ -362,10 +365,12 @@ def test_nested_atomic_preserves_error_from_transaction_destroying_trigger(
# "no such savepoint" from ROLLBACK TO SAVEPOINT
fresh_db.execute("create table t (id integer primary key, v text)")
fresh_db.execute(TRIGGER_SQL)
with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
with fresh_db.atomic():
with fresh_db.atomic():
fresh_db.execute("insert into t (v) values ('bad')")
with (
pytest.raises(sqlite3.IntegrityError, match="trigger says no"),
fresh_db.atomic(),
fresh_db.atomic(),
):
fresh_db.execute("insert into t (v) values ('bad')")
assert not fresh_db.conn.in_transaction

40
tests/test_cli.py
View file

@ -23,7 +23,7 @@ def _supports_pragma_function_list():
try:
db.execute("select * from pragma_function_list()")
return True
except Exception:
except sqlite3.DatabaseError:
return False
finally:
db.close()
@ -1021,16 +1021,14 @@ def test_query_json_binary(db_path):
"data": {
"$base64": True,
"encoded": (
"eJzt0c1xAyEMBeC7q1ABHleR3HxNAQrIjmb4M0gelx+RTY7p4N2WBYT0vmufUknH"
"8kq5lz5pqRFXsTOl3pYkE/NJnHXoStruJEVjc0mOCyTqq/ZMJnXEZW1Js2ZvRm5U+"
"DPKk9hRWqjyvTFx0YfzhT6MpGmN2lR1fzxjyfVMD9dFrS+bnkleMpMam/ZGXgrX1I"
"/K+5Au3S/9lNQRh0k4Gq/RUz8GiKfsQm+7JLsJ6fTo5JhVG00ZU76kZZkxePx49uI"
"jnpNoJyYlWUsoaSl/CcVATje/Kxu13RANnrHweaH3V5Jh4jvGyKCnxJLiXPKhmW3f"
"iCnG7Jql7RR3UvFo8jJ4z039dtOkTFmWzL1be9lt8A5II471m6vXy+l0BR/4wAc+8"
"IEPfOADH/jABz7wgQ984AMf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984A"
"Mf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984PuP7xubBoN9"
"eJzt0c1xAyEMBeC7q1ABHleR3HxNAQrIjmb4M0gelx+RTY7p4N2WBYT0vmufUknH"
"8kq5lz5pqRFXsTOl3pYkE/NJnHXoStruJEVjc0mOCyTqq/ZMJnXEZW1Js2ZvRm5U+"
"DPKk9hRWqjyvTFx0YfzhT6MpGmN2lR1fzxjyfVMD9dFrS+bnkleMpMam/ZGXgrX1I"
"/K+5Au3S/9lNQRh0k4Gq/RUz8GiKfsQm+7JLsJ6fTo5JhVG00ZU76kZZkxePx49uI"
"jnpNoJyYlWUsoaSl/CcVATje/Kxu13RANnrHweaH3V5Jh4jvGyKCnxJLiXPKhmW3f"
"iCnG7Jql7RR3UvFo8jJ4z039dtOkTFmWzL1be9lt8A5II471m6vXy+l0BR/4wAc+8"
"IEPfOADH/jABz7wgQ984AMf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984A"
"Mf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984PuP7xubBoN9"
),
},
}
@ -2116,11 +2114,13 @@ _common_other_schema = (
),
(
["--rename", "name", "name2"],
('CREATE TABLE "trees" (\n'
' "id" INTEGER PRIMARY KEY,\n'
' "address" TEXT,\n'
' "species_id" INTEGER REFERENCES "species"("id")\n'
")"),
(
'CREATE TABLE "trees" (\n'
' "id" INTEGER PRIMARY KEY,\n'
' "address" TEXT,\n'
' "species_id" INTEGER REFERENCES "species"("id")\n'
")"
),
'CREATE TABLE "species" (\n "id" INTEGER PRIMARY KEY,\n "species" TEXT\n)',
),
],
@ -2139,7 +2139,9 @@ def test_extract(db_path, args, expected_table_schema, expected_other_schema):
assert result.exit_code == 0
schema = db["trees"].schema
assert schema == expected_table_schema
other_schema = next(t for t in db.tables if t.name not in ("trees", "Gosh", "Gosh2")).schema
other_schema = next(
t for t in db.tables if t.name not in ("trees", "Gosh", "Gosh2")
).schema
assert other_schema == expected_other_schema
@ -2690,7 +2692,9 @@ def test_integer_overflow_error(tmpdir):
def test_python_dash_m():
"Tool can be run using python -m sqlite_utils"
result = subprocess.run(
[sys.executable, "-m", "sqlite_utils", "--help"], stdout=subprocess.PIPE
[sys.executable, "-m", "sqlite_utils", "--help"],
stdout=subprocess.PIPE,
check=False,
)
assert result.returncode == 0
assert b"Commands for interacting with a SQLite database" in result.stdout

3
tests/test_cli_insert.py
View file

@ -142,8 +142,7 @@ def test_insert_multiple_with_compound_primary_key(db_path, tmpdir):
def test_insert_not_null_default(db_path, tmpdir):
json_path = str(tmpdir / "dogs.json")
dogs = [
{"id": i, "name": f"Cleo {i}", "age": i + 3, "score": 10}
for i in range(1, 21)
{"id": i, "name": f"Cleo {i}", "age": i + 3, "score": 10} for i in range(1, 21)
]
with open(json_path, "w") as fp:
fp.write(json.dumps(dogs))

16
tests/test_create.py
View file

@ -720,17 +720,9 @@ def test_columns_not_in_first_record_should_not_cause_batch_to_be_too_large(fres
records = [
{"c0": "first record"}, # one column in first record -> batch size = 999
# fill out the batch with 99 records with enough columns to exceed THRESHOLD
*[
{f"c{i}": j for i in range(extra_columns)}
for j in range(batch_size - 1)
],
*[{f"c{i}": j for i in range(extra_columns)} for j in range(batch_size - 1)],
]
try:
fresh_db["too_many_columns"].insert_all(
records, alter=True, batch_size=batch_size
)
except sqlite3.OperationalError:
raise
fresh_db["too_many_columns"].insert_all(records, alter=True, batch_size=batch_size)
@pytest.mark.parametrize(
@ -927,9 +919,7 @@ def test_insert_memoryview(fresh_db):
def test_insert_thousands_using_generator(fresh_db):
fresh_db["test"].insert_all(
{"i": i, "word": f"word_{i}"} for i in range(10000)
)
fresh_db["test"].insert_all({"i": i, "word": f"word_{i}"} for i in range(10000))
assert [{"name": "i", "type": "INTEGER"}, {"name": "word", "type": "TEXT"}] == [
{"name": col.name, "type": col.type} for col in fresh_db["test"].columns
]

2
tests/test_delete.py
View file

@ -3,7 +3,7 @@ import sqlite_utils
def test_delete_rowid_table(fresh_db):
table = fresh_db["table"]
table.insert({"foo": 1}).last_pk
table.insert({"foo": 1})
rowid = table.insert({"foo": 2}).last_pk
table.delete(rowid)
assert [{"foo": 1}] == list(table.rows)

2
tests/test_duplicate.py
View file

@ -14,7 +14,7 @@ def test_duplicate(fresh_db):
"bool_col" INTEGER,
"datetime_col" TEXT)""")
# Insert one row of mock data:
dt = datetime.datetime.now()
dt = datetime.datetime.now(datetime.timezone.utc)
data = {
"text_col": "Cleo",
"real_col": 3.14,

3
tests/test_extract.py
View file

@ -33,8 +33,7 @@ def test_extract_single_column(fresh_db, table, fk_column):
+ ")"
)
assert fresh_db[expected_table].schema == (
f'CREATE TABLE "{expected_table}" (\n'
+ ' "id" INTEGER PRIMARY KEY,\n'
f'CREATE TABLE "{expected_table}" (\n' + ' "id" INTEGER PRIMARY KEY,\n'
' "species" TEXT\n'
")"
)

3
tests/test_fts.py
View file

@ -107,7 +107,8 @@ def test_search_limit_offset(fresh_db):
assert len(list(table.search("are", limit=1))) == 1
assert next(iter(table.search("are", limit=1, order_by="rowid")))["rowid"] == 1
assert (
next(iter(table.search("are", limit=1, offset=1, order_by="rowid")))["rowid"] == 2
next(iter(table.search("are", limit=1, offset=1, order_by="rowid")))["rowid"]
== 2
)

3
tests/test_plugins.py
View file

@ -1,4 +1,5 @@
import importlib
import sqlite3
import sys
import click
@ -13,7 +14,7 @@ def _supports_pragma_function_list():
try:
db.execute("select * from pragma_function_list()")
return True
except Exception:
except sqlite3.DatabaseError:
return False
finally:
db.close()

54
tests/test_tracer.py
View file

@ -53,16 +53,18 @@ def test_with_tracer():
assert len(collected) == 4
assert collected == [
(
("SELECT name FROM sqlite_master\n"
" WHERE rootpage = 0\n"
" AND (\n"
" sql LIKE :like\n"
" OR sql LIKE :like2\n"
" OR (\n"
" tbl_name = :table\n"
" AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
" )\n"
" )"),
(
"SELECT name FROM sqlite_master\n"
" WHERE rootpage = 0\n"
" AND (\n"
" sql LIKE :like\n"
" OR sql LIKE :like2\n"
" OR (\n"
" tbl_name = :table\n"
" AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
" )\n"
" )"
),
{
"like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%",
"like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
@ -72,21 +74,23 @@ def test_with_tracer():
("select name from sqlite_master where type = 'view'", None),
("select sql from sqlite_master where name = ?", ("dogs_fts",)),
(
('with "original" as (\n'
" select\n"
" rowid,\n"
" *\n"
' from "dogs"\n'
")\n"
"select\n"
' "original".*\n'
"from\n"
' "original"\n'
' join "dogs_fts" on "original".rowid = "dogs_fts".rowid\n'
"where\n"
' "dogs_fts" match :query\n'
"order by\n"
' "dogs_fts".rank'),
(
'with "original" as (\n'
" select\n"
" rowid,\n"
" *\n"
' from "dogs"\n'
")\n"
"select\n"
' "original".*\n'
"from\n"
' "original"\n'
' join "dogs_fts" on "original".rowid = "dogs_fts".rowid\n'
"where\n"
' "dogs_fts" match :query\n'
"order by\n"
' "dogs_fts".rank'
),
{"query": "Cleopaws"},
),
]

2
tests/test_update.py
View file

@ -43,7 +43,7 @@ def test_update_compound_pk_table(fresh_db):
)
def test_update_invalid_pk(fresh_db, pk, update_pk):
table = fresh_db["table"]
table.insert({"id1": 5, "id2": 3, "v": 1}, pk=pk).last_pk
table.insert({"id1": 5, "id2": 3, "v": 1}, pk=pk)
with pytest.raises(NotFoundError):
table.update(update_pk, {"v": 2})