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

Upgrade syntax with pyupgrade --py39-plus

Browse source
This commit is contained in:
Hugo van Kemenade 2024-11-24 11:48:16 +02:00
commit 15f15933b1
22 changed files with 255 additions and 259 deletions
Download patch file Download diff file Expand all files Collapse all files

26
tests/test_cli.py
View file

@ -182,9 +182,9 @@ def test_output_table(db_path, options, expected):
db["rows"].insert_all(
[
{
"c1": "verb{}".format(i),
"c2": "noun{}".format(i),
"c3": "adjective{}".format(i),
"c1": f"verb{i}",
"c2": f"noun{i}",
"c3": f"adjective{i}",
}
for i in range(4)
]
@ -614,9 +614,9 @@ def test_optimize(db_path, tables):
db[table].insert_all(
[
{
"c1": "verb{}".format(i),
"c2": "noun{}".format(i),
"c3": "adjective{}".format(i),
"c1": f"verb{i}",
"c2": f"noun{i}",
"c3": f"adjective{i}",
}
for i in range(10000)
]
@ -640,9 +640,9 @@ def test_rebuild_fts_fixes_docsize_error(db_path):
db = Database(db_path, recursive_triggers=False)
records = [
{
"c1": "verb{}".format(i),
"c2": "noun{}".format(i),
"c3": "adjective{}".format(i),
"c1": f"verb{i}",
"c2": f"noun{i}",
"c3": f"adjective{i}",
}
for i in range(10000)
]
@ -845,7 +845,6 @@ def test_query_json_binary(db_path):
"data": {
"$base64": True,
"encoded": (
(
"eJzt0c1xAyEMBeC7q1ABHleR3HxNAQrIjmb4M0gelx+RTY7p4N2WBYT0vmufUknH"
"8kq5lz5pqRFXsTOl3pYkE/NJnHXoStruJEVjc0mOCyTqq/ZMJnXEZW1Js2ZvRm5U+"
"DPKk9hRWqjyvTFx0YfzhT6MpGmN2lR1fzxjyfVMD9dFrS+bnkleMpMam/ZGXgrX1I"
@ -854,7 +853,6 @@ def test_query_json_binary(db_path):
"iCnG7Jql7RR3UvFo8jJ4z039dtOkTFmWzL1be9lt8A5II471m6vXy+l0BR/4wAc+8"
"IEPfOADH/jABz7wgQ984AMf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984A"
"Mf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984PuP7xubBoN9"
)
),
},
}
@ -2093,7 +2091,7 @@ def test_long_csv_column_value(tmpdir):
with open(csv_path, "w") as csv_file:
long_string = "a" * 131073
csv_file.write("id,text\n")
csv_file.write("1,{}\n".format(long_string))
csv_file.write(f"1,{long_string}\n")
result = CliRunner().invoke(
cli.cli,
["insert", db_path, "bigtable", csv_path, "--csv"],
@ -2396,14 +2394,14 @@ def test_load_extension(entrypoint, should_pass, should_fail):
for func in should_pass:
result = CliRunner().invoke(
cli.cli,
["memory", "select {}()".format(func), "--load-extension", ext],
["memory", f"select {func}()", "--load-extension", ext],
catch_exceptions=False,
)
assert result.exit_code == 0
for func in should_fail:
result = CliRunner().invoke(
cli.cli,
["memory", "select {}()".format(func), "--load-extension", ext],
["memory", f"select {func}()", "--load-extension", ext],
catch_exceptions=False,
)
assert result.exit_code == 1

4
tests/test_cli_convert.py
View file

@ -425,7 +425,7 @@ def test_recipe_jsonsplit(tmpdir, delimiter):
)
code = "r.jsonsplit(value)"
if delimiter:
code = 'recipes.jsonsplit(value, delimiter="{}")'.format(delimiter)
code = f'recipes.jsonsplit(value, delimiter="{delimiter}")'
args = ["convert", db_path, "example", "tags", code]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 0, result.output
@ -453,7 +453,7 @@ def test_recipe_jsonsplit_type(fresh_db_and_path, type, expected_array):
)
code = "r.jsonsplit(value)"
if type:
code = "recipes.jsonsplit(value, type={})".format(type)
code = f"recipes.jsonsplit(value, type={type})"
args = ["convert", db_path, "example", "records", code]
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 0, result.output

8
tests/test_cli_insert.py
View file

@ -99,7 +99,7 @@ def test_insert_with_primary_keys(db_path, tmpdir, args, expected_pks):
def test_insert_multiple_with_primary_key(db_path, tmpdir):
json_path = str(tmpdir / "dogs.json")
dogs = [{"id": i, "name": "Cleo {}".format(i), "age": i + 3} for i in range(1, 21)]
dogs = [{"id": i, "name": f"Cleo {i}", "age": i + 3} for i in range(1, 21)]
with open(json_path, "w") as fp:
fp.write(json.dumps(dogs))
result = CliRunner().invoke(
@ -114,7 +114,7 @@ def test_insert_multiple_with_primary_key(db_path, tmpdir):
def test_insert_multiple_with_compound_primary_key(db_path, tmpdir):
json_path = str(tmpdir / "dogs.json")
dogs = [
{"breed": "mixed", "id": i, "name": "Cleo {}".format(i), "age": i + 3}
{"breed": "mixed", "id": i, "name": f"Cleo {i}", "age": i + 3}
for i in range(1, 21)
]
with open(json_path, "w") as fp:
@ -140,7 +140,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": "Cleo {}".format(i), "age": i + 3, "score": 10}
{"id": i, "name": f"Cleo {i}", "age": i + 3, "score": 10}
for i in range(1, 21)
]
with open(json_path, "w") as fp:
@ -587,7 +587,7 @@ def test_insert_streaming_batch_size_1(db_path):
return
tries += 1
if tries > 10:
assert False, "Expected {}, got {}".format(expected, rows)
assert False, f"Expected {expected}, got {rows}"
time.sleep(tries * 0.1)
try_until([{"name": "Azi"}])

10
tests/test_cli_memory.py
View file

@ -28,7 +28,7 @@ def test_memory_csv(tmpdir, sql_from, use_stdin):
fp.write(content)
result = CliRunner().invoke(
cli.cli,
["memory", csv_path, "select * from {}".format(sql_from), "--nl"],
["memory", csv_path, f"select * from {sql_from}", "--nl"],
input=input,
)
assert result.exit_code == 0
@ -53,7 +53,7 @@ def test_memory_tsv(tmpdir, use_stdin):
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
["memory", path, f"select * from {sql_from}"],
input=input,
)
assert result.exit_code == 0, result.output
@ -79,7 +79,7 @@ def test_memory_json(tmpdir, use_stdin):
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
["memory", path, f"select * from {sql_from}"],
input=input,
)
assert result.exit_code == 0, result.output
@ -105,7 +105,7 @@ def test_memory_json_nl(tmpdir, use_stdin):
sql_from = "chickens"
result = CliRunner().invoke(
cli.cli,
["memory", path, "select * from {}".format(sql_from)],
["memory", path, f"select * from {sql_from}"],
input=input,
)
assert result.exit_code == 0, result.output
@ -135,7 +135,7 @@ def test_memory_csv_encoding(tmpdir, use_stdin):
CliRunner()
.invoke(
cli.cli,
["memory", csv_path, "select * from {}".format(sql_from), "--nl"],
["memory", csv_path, f"select * from {sql_from}", "--nl"],
input=input,
)
.exit_code

2
tests/test_column_affinity.py
View file

@ -41,5 +41,5 @@ def test_column_affinity(column_def, expected_type):
@pytest.mark.parametrize("column_def,expected_type", EXAMPLES)
def test_columns_dict(fresh_db, column_def, expected_type):
fresh_db.execute("create table foo (col {})".format(column_def))
fresh_db.execute(f"create table foo (col {column_def})")
assert {"col": expected_type} == fresh_db["foo"].columns_dict

10
tests/test_create.py
View file

@ -691,7 +691,7 @@ def test_bulk_insert_more_than_999_values(fresh_db):
"num_columns,should_error", ((900, False), (999, False), (1000, True))
)
def test_error_if_more_than_999_columns(fresh_db, num_columns, should_error):
record = dict([("c{}".format(i), i) for i in range(num_columns)])
record = {f"c{i}": i for i in range(num_columns)}
if should_error:
with pytest.raises(AssertionError):
fresh_db["big"].insert(record)
@ -711,7 +711,7 @@ def test_columns_not_in_first_record_should_not_cause_batch_to_be_too_large(fres
{"c0": "first record"}, # one column in first record -> batch size = 999
# fill out the batch with 99 records with enough columns to exceed THRESHOLD
*[
dict([("c{}".format(i), j) for i in range(extra_columns)])
{f"c{i}": j for i in range(extra_columns)}
for j in range(batch_size - 1)
],
]
@ -890,7 +890,7 @@ def test_insert_memoryview(fresh_db):
def test_insert_thousands_using_generator(fresh_db):
fresh_db["test"].insert_all(
{"i": i, "word": "word_{}".format(i)} for i in range(10000)
{"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
@ -902,7 +902,7 @@ def test_insert_thousands_raises_exception_with_extra_columns_after_first_100(fr
# https://github.com/simonw/sqlite-utils/issues/139
with pytest.raises(Exception, match="table test has no column named extra"):
fresh_db["test"].insert_all(
[{"i": i, "word": "word_{}".format(i)} for i in range(100)]
[{"i": i, "word": f"word_{i}"} for i in range(100)]
+ [{"i": 101, "extra": "This extra column should cause an exception"}],
)
@ -910,7 +910,7 @@ def test_insert_thousands_raises_exception_with_extra_columns_after_first_100(fr
def test_insert_thousands_adds_extra_columns_after_first_100_with_alter(fresh_db):
# https://github.com/simonw/sqlite-utils/issues/139
fresh_db["test"].insert_all(
[{"i": i, "word": "word_{}".format(i)} for i in range(100)]
[{"i": i, "word": f"word_{i}"} for i in range(100)]
+ [{"i": 101, "extra": "Should trigger ALTER"}],
alter=True,
)

2
tests/test_default_value.py
View file

@ -27,7 +27,7 @@ EXAMPLES = [
@pytest.mark.parametrize("column_def,initial_value,expected_value", EXAMPLES)
def test_quote_default_value(fresh_db, column_def, initial_value, expected_value):
fresh_db.execute("create table foo (col {})".format(column_def))
fresh_db.execute(f"create table foo (col {column_def})")
assert initial_value == fresh_db["foo"].columns[0].default_value
assert expected_value == fresh_db.quote_default_value(
fresh_db["foo"].columns[0].default_value

2
tests/test_docs.py
View file

@ -34,7 +34,7 @@ def test_commands_are_documented(documented_commands, command):
@pytest.mark.parametrize("command", cli.cli.commands.values())
def test_commands_have_help(command):
assert command.help, "{} is missing its help".format(command)
assert command.help, f"{command} is missing its help"
def test_convert_help():

4
tests/test_enable_counts.py
View file

@ -8,7 +8,7 @@ def test_enable_counts_specific_table(fresh_db):
foo = fresh_db["foo"]
assert fresh_db.table_names() == []
for i in range(10):
foo.insert({"name": "item {}".format(i)})
foo.insert({"name": f"item {i}"})
assert fresh_db.table_names() == ["foo"]
assert foo.count == 10
# Now enable counts
@ -44,7 +44,7 @@ def test_enable_counts_specific_table(fresh_db):
assert list(fresh_db["_counts"].rows) == [{"count": 10, "table": "foo"}]
# Add some items to test the triggers
for i in range(5):
foo.insert({"name": "item {}".format(10 + i)})
foo.insert({"name": f"item {10 + i}"})
assert foo.count == 15
assert list(fresh_db["_counts"].rows) == [{"count": 15, "table": "foo"}]
# Delete some items

8
tests/test_extract.py
View file

@ -7,13 +7,13 @@ import pytest
@pytest.mark.parametrize("fk_column", [None, "species"])
def test_extract_single_column(fresh_db, table, fk_column):
expected_table = table or "species"
expected_fk = fk_column or "{}_id".format(expected_table)
expected_fk = fk_column or f"{expected_table}_id"
iter_species = itertools.cycle(["Palm", "Spruce", "Mangrove", "Oak"])
fresh_db["tree"].insert_all(
(
{
"id": i,
"name": "Tree {}".format(i),
"name": f"Tree {i}",
"species": next(iter_species),
"end": 1,
}
@ -31,7 +31,7 @@ def test_extract_single_column(fresh_db, table, fk_column):
+ ")"
)
assert fresh_db[expected_table].schema == (
"CREATE TABLE [{}] (\n".format(expected_table)
f"CREATE TABLE [{expected_table}] (\n"
+ " [id] INTEGER PRIMARY KEY,\n"
" [species] TEXT\n"
")"
@ -57,7 +57,7 @@ def test_extract_multiple_columns_with_rename(fresh_db):
(
{
"id": i,
"name": "Tree {}".format(i),
"name": f"Tree {i}",
"common_name": next(iter_common),
"latin_name": next(iter_latin),
}

2
tests/test_extracts.py
View file

@ -51,7 +51,7 @@ def test_extracts(fresh_db, kwargs, expected_table, use_table_factory):
assert [
Index(
seq=0,
name="idx_{}_value".format(expected_table),
name=f"idx_{expected_table}_value",
unique=1,
origin="c",
partial=0,

16
tests/test_fts.py
View file

@ -209,20 +209,20 @@ def test_populate_fts_escape_table_names(fresh_db):
@pytest.mark.parametrize("fts_version", ("4", "5"))
def test_fts_tokenize(fresh_db, fts_version):
table_name = "searchable_{}".format(fts_version)
table_name = f"searchable_{fts_version}"
table = fresh_db[table_name]
table.insert_all(search_records)
# Test without porter stemming
table.enable_fts(
["text", "country"],
fts_version="FTS{}".format(fts_version),
fts_version=f"FTS{fts_version}",
)
assert [] == list(table.search("bite"))
# Test WITH stemming
table.disable_fts()
table.enable_fts(
["text", "country"],
fts_version="FTS{}".format(fts_version),
fts_version=f"FTS{fts_version}",
tokenize="porter",
)
rows = list(table.search("bite", order_by="rowid"))
@ -237,10 +237,10 @@ def test_fts_tokenize(fresh_db, fts_version):
def test_optimize_fts(fresh_db):
for fts_version in ("4", "5"):
table_name = "searchable_{}".format(fts_version)
table_name = f"searchable_{fts_version}"
table = fresh_db[table_name]
table.insert_all(search_records)
table.enable_fts(["text", "country"], fts_version="FTS{}".format(fts_version))
table.enable_fts(["text", "country"], fts_version=f"FTS{fts_version}")
# You can call optimize successfully against the tables OR their _fts equivalents:
for table_name in (
"searchable_4",
@ -296,12 +296,12 @@ def test_disable_fts(fresh_db, create_triggers):
expected_triggers = {"searchable_ai", "searchable_ad", "searchable_au"}
else:
expected_triggers = set()
assert expected_triggers == set(
assert expected_triggers == {
r[0]
for r in fresh_db.execute(
"select name from sqlite_master where type = 'trigger'"
).fetchall()
)
}
# Now run .disable_fts() and confirm it worked
table.disable_fts()
assert (
@ -392,7 +392,7 @@ def test_enable_fts_replace(kwargs):
db["books"].enable_fts(**kwargs, replace=True)
# Check that the new configuration is correct
if should_have_changed_columns:
assert db["books_fts"].columns_dict.keys() == set(["title"])
assert db["books_fts"].columns_dict.keys() == {"title"}
if "create_triggers" in kwargs:
assert db["books"].triggers
if "fts_version" in kwargs:

2
tests/test_gis.py
View file

@ -113,7 +113,7 @@ def test_query_load_extension(use_spatialite_shortcut):
[
":memory:",
"select spatialite_version()",
"--load-extension={}".format(load_extension),
f"--load-extension={load_extension}",
],
)
assert result.exit_code == 0, result.stdout

4
tests/test_insert_files.py
View file

@ -44,7 +44,7 @@ def test_insert_files(silent, pk_args, expected_pks):
)
cols = []
for coltype in coltypes:
cols += ["-c", "{}:{}".format(coltype, coltype)]
cols += ["-c", f"{coltype}:{coltype}"]
result = runner.invoke(
cli.cli,
["insert-files", db_path, "files", str(tmpdir)]
@ -167,5 +167,5 @@ def test_insert_files_bad_text_encoding_error():
)
assert result.exit_code == 1, result.output
assert result.output.strip().startswith(
"Error: Could not read file '{}' as text".format(str(latin.resolve()))
f"Error: Could not read file '{str(latin.resolve())}' as text"
)

4
tests/test_introspect.py
View file

@ -49,8 +49,8 @@ def test_detect_fts_similar_tables(fresh_db, reverse_order):
fresh_db[table2].insert({"title": "Hello"}).enable_fts(
["title"], fts_version="FTS4"
)
assert fresh_db[table1].detect_fts() == "{}_fts".format(table1)
assert fresh_db[table2].detect_fts() == "{}_fts".format(table2)
assert fresh_db[table1].detect_fts() == f"{table1}_fts"
assert fresh_db[table2].detect_fts() == f"{table2}_fts"
def test_tables(existing_db):

3
tests/test_m2m.py
View file

@ -65,8 +65,7 @@ def test_insert_m2m_iterable(fresh_db):
iterable_records = ({"id": 1, "name": "Phineas"}, {"id": 2, "name": "Ferb"})
def iterable():
for record in iterable_records:
yield record
yield from iterable_records
platypuses = fresh_db["platypuses"]
platypuses.insert({"id": 1, "name": "Perry"}, pk="id").m2m(

2
tests/test_utils.py
View file

@ -57,7 +57,7 @@ def test_maximize_csv_field_size_limit():
# Reset to default in case other tests have changed it
csv.field_size_limit(utils.ORIGINAL_CSV_FIELD_SIZE_LIMIT)
long_value = "a" * 131073
long_csv = "id,text\n1,{}".format(long_value)
long_csv = f"id,text\n1,{long_value}"
fp = io.BytesIO(long_csv.encode("utf-8"))
# Using rows_from_file should error
with pytest.raises(csv.Error):