remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-07-26 19:04:32 +02:00
Code Issues Projects Releases Packages Wiki Activity

Ran latest Black

Browse source
This commit is contained in:
Simon Willison 2026-05-17 16:27:07 -07:00
commit cb143c0bcf
12 changed files with 46 additions and 110 deletions
Download patch file Download diff file Expand all files Collapse all files

19
sqlite_utils/cli.py
View file

@ -42,7 +42,6 @@ from .utils import (
TypeTracker,
)
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
@ -2911,8 +2910,7 @@ def _analyze(db, tables, columns, save, common_limit=10, no_most=False, no_least
)
details = (
(
textwrap.dedent(
"""
textwrap.dedent("""
{table}.{column}: ({i}/{total})
Total rows: {total_rows}
@ -2920,8 +2918,7 @@ def _analyze(db, tables, columns, save, common_limit=10, no_most=False, no_least
Blank rows: {num_blank}
Distinct values: {num_distinct}{most_common_rendered}{least_common_rendered}
"""
)
""")
.strip()
.format(
i=i + 1,
@ -2968,8 +2965,7 @@ def uninstall(packages, yes):
def _generate_convert_help():
help = textwrap.dedent(
"""
help = textwrap.dedent("""
Convert columns using Python code you supply. For example:
\b
@ -2982,8 +2978,7 @@ def _generate_convert_help():
Use "-" for CODE to read Python code from standard input.
The following common operations are available as recipe functions:
"""
).strip()
""").strip()
recipe_names = [
n
for n in dir(recipes)
@ -2997,15 +2992,13 @@ def _generate_convert_help():
name, str(inspect.signature(fn)), textwrap.dedent(fn.__doc__.rstrip())
)
help += "\n\n"
help += textwrap.dedent(
"""
help += textwrap.dedent("""
You can use these recipes like so:
\b
sqlite-utils convert my.db mytable mycolumn \\
'r.jsonsplit(value, delimiter=":")'
"""
).strip()
""").strip()
return help

60
sqlite_utils/db.py
View file

@ -2275,12 +2275,10 @@ class Table(Queryable):
"{}_{}".format(index_name, suffix) if suffix else index_name
)
sql = (
textwrap.dedent(
"""
textwrap.dedent("""
CREATE {unique}INDEX {if_not_exists}{index_name}
ON {table_name} ({columns});
"""
)
""")
.strip()
.format(
index_name=quote_identifier(created_index_name),
@ -2475,8 +2473,7 @@ class Table(Queryable):
See :ref:`python_api_cached_table_counts` for details.
"""
sql = (
textwrap.dedent(
"""
textwrap.dedent("""
{create_counts_table}
CREATE TRIGGER IF NOT EXISTS {trigger_insert} AFTER INSERT ON {table}
BEGIN
@ -2501,8 +2498,7 @@ class Table(Queryable):
);
END;
INSERT OR REPLACE INTO _counts VALUES ({table_quoted}, (select count(*) from {table}));
"""
)
""")
.strip()
.format(
create_counts_table=_COUNTS_TABLE_CREATE_SQL.format(
@ -2554,14 +2550,12 @@ class Table(Queryable):
:param replace: Should any existing FTS index for this table be replaced by the new one?
"""
create_fts_sql = (
textwrap.dedent(
"""
textwrap.dedent("""
CREATE VIRTUAL TABLE {table_fts} USING {fts_version} (
{columns},{tokenize}
content={table}
)
"""
)
""")
.strip()
.format(
table=quote_identifier(self.name),
@ -2599,8 +2593,7 @@ class Table(Queryable):
table = quote_identifier(self.name)
table_fts = quote_identifier(self.name + "_fts")
triggers = (
textwrap.dedent(
"""
textwrap.dedent("""
CREATE TRIGGER {table_ai} AFTER INSERT ON {table} BEGIN
INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
END;
@ -2611,8 +2604,7 @@ class Table(Queryable):
INSERT INTO {table_fts} ({table_fts}, rowid, {columns}) VALUES('delete', old.rowid, {old_cols});
INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
END;
"""
)
""")
.strip()
.format(
table=table,
@ -2637,12 +2629,10 @@ class Table(Queryable):
"""
columns_quoted = ", ".join(quote_identifier(c) for c in columns)
sql = (
textwrap.dedent(
"""
textwrap.dedent("""
INSERT INTO {table_fts} (rowid, {columns})
SELECT rowid, {columns} FROM {table};
"""
)
""")
.strip()
.format(
table=quote_identifier(self.name),
@ -2659,17 +2649,11 @@ class Table(Queryable):
if fts_table:
self.db[fts_table].drop()
# Now delete the triggers that related to that table
sql = (
textwrap.dedent(
"""
sql = textwrap.dedent("""
SELECT name FROM sqlite_master
WHERE type = 'trigger'
AND (sql LIKE '% INSERT INTO [{}]%' OR sql LIKE '% INSERT INTO "{}"%')
"""
)
.strip()
.format(fts_table, fts_table)
)
""").strip().format(fts_table, fts_table)
trigger_names = []
for row in self.db.execute(sql).fetchall():
trigger_names.append(row[0])
@ -2695,8 +2679,7 @@ class Table(Queryable):
def detect_fts(self) -> Optional[str]:
"Detect if table has a corresponding FTS virtual table and return it"
sql = textwrap.dedent(
"""
sql = textwrap.dedent("""
SELECT name FROM sqlite_master
WHERE rootpage = 0
AND (
@ -2707,8 +2690,7 @@ class Table(Queryable):
AND sql LIKE '%VIRTUAL TABLE%USING FTS%'
)
)
"""
).strip()
""").strip()
args = {
"like": '%VIRTUAL TABLE%USING FTS%content="{}"%'.format(self.name),
"like2": '%VIRTUAL TABLE%USING FTS%content="{}"%'.format(self.name),
@ -2724,13 +2706,9 @@ class Table(Queryable):
"Run the ``optimize`` operation against the associated full-text search index table."
fts_table = self.detect_fts()
if fts_table is not None:
self.db.execute(
"""
self.db.execute("""
INSERT INTO {table} ({table}) VALUES ("optimize");
""".strip().format(
table=quote_identifier(fts_table)
)
)
""".strip().format(table=quote_identifier(fts_table)))
return self
def search_sql(
@ -2768,8 +2746,7 @@ class Table(Queryable):
)
fts_table_quoted = quote_identifier(fts_table)
virtual_table_using = self.db.table(fts_table).virtual_table_using
sql = textwrap.dedent(
"""
sql = textwrap.dedent("""
with {original} as (
select
rowid,
@ -2786,8 +2763,7 @@ class Table(Queryable):
order by
{order_by}
{limit_offset}
"""
).strip()
""").strip()
if virtual_table_using == "FTS5":
rank_implementation = "{}.rank".format(fts_table_quoted)
else:

6
tests/conftest.py
View file

@ -42,14 +42,12 @@ def fresh_db():
@pytest.fixture
def existing_db():
database = Database(memory=True)
database.executescript(
"""
database.executescript("""
CREATE TABLE foo (text TEXT);
INSERT INTO foo (text) values ("one");
INSERT INTO foo (text) values ("two");
INSERT INTO foo (text) values ("three");
"""
)
""")
return database

9
tests/test_analyze_tables.py
View file

@ -143,10 +143,7 @@ def db_to_analyze_path(db_to_analyze, tmpdir):
def test_analyze_table(db_to_analyze_path):
result = CliRunner().invoke(cli.cli, ["analyze-tables", db_to_analyze_path])
assert (
result.output.strip()
== (
"""
assert result.output.strip() == ("""
stuff.id: (1/3)
Total rows: 8
@ -179,9 +176,7 @@ stuff.size: (3/3)
Most common:
5: 5
3: 4"""
).strip()
)
3: 4""").strip()
def test_analyze_table_save(db_to_analyze_path):

21
tests/test_cli.py
View file

@ -967,12 +967,9 @@ def test_query_json_with_json_cols(db_path):
result = CliRunner().invoke(
cli.cli, [db_path, "select id, name, friends from dogs"]
)
assert (
r"""
assert r"""
[{"id": 1, "name": "Cleo", "friends": "[{\"name\": \"Pancakes\"}, {\"name\": \"Bailey\"}]"}]
""".strip()
== result.output.strip()
)
""".strip() == result.output.strip()
# With --json-cols:
result = CliRunner().invoke(
cli.cli, [db_path, "select id, name, friends from dogs", "--json-cols"]
@ -1998,12 +1995,10 @@ def test_search_quote(tmpdir):
def test_indexes(tmpdir):
db_path = str(tmpdir / "test.db")
db = Database(db_path)
db.conn.executescript(
"""
db.conn.executescript("""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_idx on Gosh(c2, c3 desc);
"""
)
""")
result = CliRunner().invoke(
cli.cli,
["indexes", str(db_path)],
@ -2094,16 +2089,12 @@ def test_triggers(tmpdir, extra_args, expected):
pk="id",
)
db["counter"].insert({"count": 1})
db.conn.execute(
textwrap.dedent(
"""
db.conn.execute(textwrap.dedent("""
CREATE TRIGGER blah AFTER INSERT ON articles
BEGIN
UPDATE counter SET count = count + 1;
END
"""
)
)
"""))
args = ["triggers", db_path]
if extra_args:
args.extend(extra_args)

6
tests/test_cli_convert.py
View file

@ -371,16 +371,14 @@ def test_convert_multi_complex_column_types(fresh_db_and_path):
],
pk="id",
)
code = textwrap.dedent(
"""
code = textwrap.dedent("""
if value == 1:
return {"is_str": "", "is_float": 1.2, "is_int": None}
elif value == 2:
return {"is_float": 1, "is_int": 12}
elif value == 3:
return {"is_bytes": b"blah"}
"""
)
""")
result = CliRunner().invoke(
cli.cli,
[

1
tests/test_create.py
View file

@ -20,7 +20,6 @@ import pathlib
import pytest
import uuid
try:
import pandas as pd # type: ignore
except ImportError:

1
tests/test_default_value.py
View file

@ -1,6 +1,5 @@
import pytest
EXAMPLES = [
("TEXT DEFAULT 'foo'", "'foo'", "'foo'"),
("TEXT DEFAULT 'foo)'", "'foo)'", "'foo)'"),

6
tests/test_duplicate.py
View file

@ -5,14 +5,12 @@ import pytest
def test_duplicate(fresh_db):
# Create table using native Sqlite statement:
fresh_db.execute(
"""CREATE TABLE "table1" (
fresh_db.execute("""CREATE TABLE "table1" (
"text_col" TEXT,
"real_col" REAL,
"int_col" INTEGER,
"bool_col" INTEGER,
"datetime_col" TEXT)"""
)
"datetime_col" TEXT)""")
# Insert one row of mock data:
dt = datetime.datetime.now()
data = {

9
tests/test_extract.py
View file

@ -126,9 +126,7 @@ def test_extract_rowid_table(fresh_db):
' "common_name_latin_name_id" INTEGER REFERENCES "common_name_latin_name"("id")\n'
")"
)
assert (
fresh_db.execute(
"""
assert fresh_db.execute("""
select
tree.name,
common_name_latin_name.common_name,
@ -136,10 +134,7 @@ def test_extract_rowid_table(fresh_db):
from tree
join common_name_latin_name
on tree.common_name_latin_name_id = common_name_latin_name.id
"""
).fetchall()
== [("Tree 1", "Palm", "Arecaceae")]
)
""").fetchall() == [("Tree 1", "Palm", "Arecaceae")]
def test_reuse_lookup_table(fresh_db):

12
tests/test_introspect.py
View file

@ -109,13 +109,11 @@ def test_table_repr(fresh_db):
def test_indexes(fresh_db):
fresh_db.executescript(
"""
fresh_db.executescript("""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_c1 on Gosh(c1);
create index Gosh_c2c3 on Gosh(c2, c3);
"""
)
""")
assert [
Index(
seq=0,
@ -130,13 +128,11 @@ def test_indexes(fresh_db):
def test_xindexes(fresh_db):
fresh_db.executescript(
"""
fresh_db.executescript("""
create table Gosh (c1 text, c2 text, c3 text);
create index Gosh_c1 on Gosh(c1);
create index Gosh_c2c3 on Gosh(c2, c3 desc);
"""
)
""")
assert fresh_db["Gosh"].xindexes == [
XIndex(
name="Gosh_c2c3",

6
tests/test_transform.py
View file

@ -638,15 +638,13 @@ def test_transform_with_indexes_errors(fresh_db, transform_params):
def test_transform_with_unique_constraint_implicit_index(fresh_db):
dogs = fresh_db["dogs"]
# Create a table with a UNIQUE constraint on 'name', which creates an implicit index
fresh_db.execute(
"""
fresh_db.execute("""
CREATE TABLE dogs (
id INTEGER PRIMARY KEY,
name TEXT UNIQUE,
age INTEGER
);
"""
)
""")
dogs.insert({"id": 1, "name": "Cleo", "age": 5})
# Attempt to transform the table without modifying 'name'