remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-07-23 01:14:31 +02:00
Code Issues Projects Releases Packages Wiki Activity

Implemented sqlite-utils transform command, closes #164

Browse source
This commit is contained in:
Simon Willison 2020-09-22 00:46:32 -07:00
commit 752d261229
5 changed files with 309 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

65
docs/cli.rst
View file

@ -545,7 +545,70 @@ Dropping tables
You can drop a table using the ``drop-table`` command::
$ sqlite-utils drop-table mytable
$ sqlite-utils drop-table mydb.db mytable
.. _cli_transform_table:
Transforming tables
===================
The ``transform`` command allows you to apply complex transformations to a table that cannot be implemented using a regular SQLite ``ALTER TABLE`` command. See :ref:`python_api_transform` for details of how this works.
::
$ sqlite-utils transform mydb.db mytable \
--drop column1 \
--rename column2 column_renamed
Every option for this table (with the exception of ``--pk-none``) can be specified multiple times. The options are as follows:
``--type column-name new-type``
Change the type of the specified column. Valid types are ``integer``, ``text``, ``float``, ``blob``.
``--drop column-name``
Drop the specified column.
``--rename column-name new-name``
Rename this column to a new name.
``--not-null column-name``
Set this column as ``NOT NULL``.
``--not-null-false column-name``
For a column that is currently set as ``NOT NULL``, remove the ``NOT NULL``.
``--pk column-name``
Change the primary key column for this table. Pass ``--pk`` multiple times if you want to create a compound primary key.
``--pk-none``
Remove the primary key from this table, turning it into a ``rowid`` table.
``--default column-name value``
Set the default value of this column.
``--default-none column``
Remove the default value for this column.
``--drop-foreign-key col other_table other_column``
Drop the specified foreign key.
If you want to see the SQL that will be executed to make the change without actually executing it, add the ``--sql`` flag. For example::
% sqlite-utils transform fixtures.db roadside_attractions \
--rename pk id \
--default name Untitled \
--drop address \
--sql
CREATE TABLE [roadside_attractions_new_4033a60276b9] (
[id] INTEGER PRIMARY KEY,
[name] TEXT DEFAULT 'Untitled',
[latitude] FLOAT,
[longitude] FLOAT
);
INSERT INTO [roadside_attractions_new_4033a60276b9] ([id], [name], [latitude], [longitude])
SELECT [pk], [name], [latitude], [longitude] FROM [roadside_attractions];
DROP TABLE [roadside_attractions];
ALTER TABLE [roadside_attractions_new_4033a60276b9] RENAME TO [roadside_attractions];
.. _cli_create_view:

100
sqlite_utils/cli.py
View file

@ -887,6 +887,106 @@ def rows(ctx, path, dbtable, nl, arrays, csv, no_headers, table, fmt, json_cols)
)
@cli.command()
@click.argument(
"path",
type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
required=True,
)
@click.argument("table")
@click.option(
"--type",
type=(str, str),
multiple=True,
help="Change column type to X",
)
@click.option("--drop", type=str, multiple=True, help="Drop this column")
@click.option(
"--rename", type=(str, str), multiple=True, help="Rename this column to X"
)
@click.option("--not-null", type=str, multiple=True, help="Set this column to NOT NULL")
@click.option(
"--not-null-false", type=str, multiple=True, help="Remove NOT NULL from this column"
)
@click.option("--pk", type=str, multiple=True, help="Make this column the primary key")
@click.option(
"--pk-none", is_flag=True, help="Remove primary key (convert to rowid table)"
)
@click.option(
"--default",
type=(str, str),
multiple=True,
help="Set default value for this column",
)
@click.option(
"--default-none", type=str, multiple=True, help="Remove default from this column"
)
@click.option(
"--drop-foreign-key",
type=(str, str, str),
multiple=True,
help="Drop this foreign key constraint",
)
@click.option("--sql", is_flag=True, help="Output SQL without executing it")
def transform(
path,
table,
type,
drop,
rename,
not_null,
not_null_false,
pk,
pk_none,
default,
default_none,
drop_foreign_key,
sql,
):
db = sqlite_utils.Database(path)
types = {}
kwargs = {}
for column, ctype in type:
if ctype.upper() not in VALID_COLUMN_TYPES:
raise click.ClickException(
"column types must be one of {}".format(VALID_COLUMN_TYPES)
)
types[column] = ctype.upper()
not_null_dict = {}
for column in not_null:
not_null_dict[column] = True
for column in not_null_false:
not_null_dict[column] = False
default_dict = {}
for column, value in default:
default_dict[column] = value
for column in default_none:
default_dict[column] = None
kwargs["types"] = types
kwargs["drop"] = set(drop)
kwargs["rename"] = dict(rename)
kwargs["not_null"] = not_null_dict
if pk:
if len(pk) == 1:
kwargs["pk"] = pk[0]
else:
kwargs["pk"] = pk
elif pk_none:
kwargs["pk"] = None
kwargs["defaults"] = default_dict
if drop_foreign_key:
kwargs["drop_foreign_keys"] = drop_foreign_key
if sql:
for line in db[table].transform_sql(**kwargs):
click.echo(line)
else:
db[table].transform(**kwargs)
@cli.command(name="insert-files")
@click.argument(
"path",

31
sqlite_utils/db.py
View file

@ -731,7 +731,7 @@ class Table(Queryable):
sqls = self.transform_sql(
types=types,
rename=rename,
drop=None,
drop=drop,
pk=pk,
not_null=not_null,
defaults=defaults,
@ -790,7 +790,7 @@ class Table(Queryable):
sqls = []
if should_flip_foreign_keys_pragma:
sqls.append("PRAGMA foreign_keys=OFF")
sqls.append("PRAGMA foreign_keys=OFF;")
if pk is DEFAULT:
pks_renamed = tuple(rename.get(p) or p for p in self.pks)
@ -800,7 +800,12 @@ class Table(Queryable):
pk = pks_renamed
# not_null may be a set or dict, need to convert to a set
create_table_not_null = {c.name for c in self.columns if c.notnull}
create_table_not_null = {
rename.get(c.name) or c.name
for c in self.columns
if c.notnull
if c.name not in drop
}
if isinstance(not_null, dict):
# Remove any columns with a value of False
for key, value in not_null.items():
@ -811,13 +816,16 @@ class Table(Queryable):
else:
create_table_not_null.add(key)
elif isinstance(not_null, set):
create_table_not_null.update(rename.get(k) or k for k in not_null)
create_table_not_null.update((rename.get(k) or k) for k in not_null)
elif not_null is None:
pass
else:
assert False, "not_null must be a dict or a set or None"
# defaults=
create_table_defaults = {
(rename.get(c.name) or c.name): c.default_value
for c in self.columns
if c.default_value is not None
if c.default_value is not None and c.name not in drop
}
if defaults is not None:
create_table_defaults.update(
@ -844,13 +852,14 @@ class Table(Queryable):
foreign_keys=create_table_foreign_keys,
).strip()
)
# Copy across data, respecting any renamed columns
new_cols = []
old_cols = []
for from_, to_ in copy_from_to.items():
old_cols.append(from_)
new_cols.append(to_)
copy_sql = "INSERT INTO [{new_table}] ({new_cols}) SELECT {old_cols} FROM [{old_table}]".format(
copy_sql = "INSERT INTO [{new_table}] ({new_cols})\n SELECT {old_cols} FROM [{old_table}];".format(
new_table=new_table_name,
old_table=self.name,
old_cols=", ".join("[{}]".format(col) for col in old_cols),
@ -858,13 +867,13 @@ class Table(Queryable):
)
sqls.append(copy_sql)
# Drop the old table
sqls.append("DROP TABLE [{}]".format(self.name))
sqls.append("DROP TABLE [{}];".format(self.name))
# Rename the new one
sqls.append("ALTER TABLE [{}] RENAME TO [{}]".format(new_table_name, self.name))
sqls.append("ALTER TABLE [{}] RENAME TO [{}];".format(new_table_name, self.name))
if should_flip_foreign_keys_pragma:
sqls.append("PRAGMA foreign_key_check")
sqls.append("PRAGMA foreign_keys=ON")
sqls.append("PRAGMA foreign_key_check;")
sqls.append("PRAGMA foreign_keys=ON;")
return sqls

98
tests/test_cli.py
View file

@ -1451,3 +1451,101 @@ def test_add_foreign_keys(db_path):
table="books", column="author_id", other_table="authors", other_column="id"
)
]
@pytest.mark.parametrize(
"args,expected_schema",
[
(
[],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT\n)",
),
(
["--type", "age", "text"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] TEXT NOT NULL DEFAULT '1',\n [name] TEXT\n)",
),
(
["--drop", "age"],
'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n)',
),
(
["--rename", "age", "age2", "--rename", "id", "pk"],
"CREATE TABLE \"dogs\" (\n [pk] INTEGER PRIMARY KEY,\n [age2] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT\n)",
),
(
["--not-null", "name"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT NOT NULL\n)",
),
(
["--not-null-false", "age"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER DEFAULT '1',\n [name] TEXT\n)",
),
(
["--pk", "name"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT PRIMARY KEY\n)",
),
(
["--pk-none"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT\n)",
),
(
["--default", "name", "Turnip"],
"CREATE TABLE \"dogs\" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER NOT NULL DEFAULT '1',\n [name] TEXT DEFAULT 'Turnip'\n)",
),
(
["--default-none", "age"],
'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [age] INTEGER NOT NULL,\n [name] TEXT\n)',
),
],
)
def test_transform(db_path, args, expected_schema):
db = Database(db_path)
with db.conn:
db["dogs"].insert(
{"id": 1, "age": 4, "name": "Cleo"},
not_null={"age"},
defaults={"age": 1},
pk="id",
)
result = CliRunner().invoke(cli.cli, ["transform", db_path, "dogs"] + args)
print(result.output)
assert result.exit_code == 0
schema = db["dogs"].schema
assert schema == expected_schema
def test_transform_drop_foreign_key(db_path):
db = Database(db_path)
with db.conn:
# Create table with three foreign keys so we can drop two of them
db["country"].insert({"id": 1, "name": "France"}, pk="id")
db["city"].insert({"id": 24, "name": "Paris"}, pk="id")
db["places"].insert(
{
"id": 32,
"name": "Caveau de la Huchette",
"country": 1,
"city": 24,
},
foreign_keys=("country", "city"),
pk="id",
)
result = CliRunner().invoke(
cli.cli,
[
"transform",
db_path,
"places",
"--drop-foreign-key",
"country",
"country",
"id",
],
)
print(result.output)
assert result.exit_code == 0
schema = db["places"].schema
assert (
schema
== 'CREATE TABLE "places" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [country] INTEGER,\n [city] INTEGER REFERENCES [city]([id])\n)'
)

54
tests/test_transform.py
View file

@ -11,9 +11,9 @@ import pytest
{},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] TEXT\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age]) SELECT [id], [name], [age] FROM [dogs]",
"DROP TABLE [dogs]",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs]",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Change column type
@ -21,9 +21,9 @@ import pytest
{"types": {"age": int}},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] INTEGER\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age]) SELECT [id], [name], [age] FROM [dogs]",
"DROP TABLE [dogs]",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs]",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Rename a column
@ -31,9 +31,9 @@ import pytest
{"rename": {"age": "dog_age"}},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [dog_age] TEXT\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [dog_age]) SELECT [id], [name], [age] FROM [dogs]",
"DROP TABLE [dogs]",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs]",
"INSERT INTO [dogs_new_suffix] ([id], [name], [dog_age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Drop a column
@ -41,9 +41,9 @@ import pytest
{"drop": ["age"]},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name]) SELECT [id], [name] FROM [dogs]",
"DROP TABLE [dogs]",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs]",
"INSERT INTO [dogs_new_suffix] ([id], [name])\n SELECT [id], [name] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Convert type AND rename column
@ -51,9 +51,9 @@ import pytest
{"types": {"age": int}, "rename": {"age": "dog_age"}},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [dog_age] INTEGER\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [dog_age]) SELECT [id], [name], [age] FROM [dogs]",
"DROP TABLE [dogs]",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs]",
"INSERT INTO [dogs_new_suffix] ([id], [name], [dog_age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Change primary key
@ -61,9 +61,9 @@ import pytest
{"pk": "age"},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT PRIMARY KEY\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age]) SELECT [id], [name], [age] FROM [dogs]",
"DROP TABLE [dogs]",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs]",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Change primary key to a compound pk
@ -71,9 +71,9 @@ import pytest
{"pk": ("age", "name")},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT,\n PRIMARY KEY ([age], [name])\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age]) SELECT [id], [name], [age] FROM [dogs]",
"DROP TABLE [dogs]",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs]",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
# Remove primary key, creating a rowid table
@ -81,9 +81,9 @@ import pytest
{"pk": None},
[
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT\n);",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age]) SELECT [id], [name], [age] FROM [dogs]",
"DROP TABLE [dogs]",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs]",
"INSERT INTO [dogs_new_suffix] ([id], [name], [age])\n SELECT [id], [name], [age] FROM [dogs];",
"DROP TABLE [dogs];",
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];",
],
),
],
@ -93,9 +93,9 @@ def test_transform_sql(fresh_db, params, expected_sql, use_pragma_foreign_keys):
dogs = fresh_db["dogs"]
if use_pragma_foreign_keys:
fresh_db.conn.execute("PRAGMA foreign_keys=ON")
expected_sql.insert(0, "PRAGMA foreign_keys=OFF")
expected_sql.append("PRAGMA foreign_key_check")
expected_sql.append("PRAGMA foreign_keys=ON")
expected_sql.insert(0, "PRAGMA foreign_keys=OFF;")
expected_sql.append("PRAGMA foreign_key_check;")
expected_sql.append("PRAGMA foreign_keys=ON;")
dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
sql = dogs.transform_sql(**{**params, **{"tmp_suffix": "suffix"}})
assert sql == expected_sql