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

Tests for reusing lookup table in extract()

Browse source
This commit is contained in:
Simon Willison 2020-09-24 08:40:55 -07:00
commit 227095ee90
2 changed files with 51 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

20
sqlite_utils/db.py
View file

@ -901,15 +901,15 @@ class Table(Queryable):
for col, typ in self.columns_dict.items()
if col in columns
}
if lookup_table.exists() and not set(lookup_columns_definition.keys()).issubset(
lookup_table.columns_dict.keys()
):
# TODO: Write test for this
raise InvalidColumns(
"Lookup table {} already exists but does not have columns {}".format(
table, lookup_columns_definition.keys()
if lookup_table.exists():
if not set(lookup_columns_definition.items()).issubset(
lookup_table.columns_dict.items()
):
raise InvalidColumns(
"Lookup table {} already exists but does not have columns {}".format(
table, lookup_columns_definition
)
)
)
else:
lookup_table.create(
{
@ -921,9 +921,9 @@ class Table(Queryable):
pk="id",
)
lookup_columns = [(rename.get(col) or col) for col in columns]
lookup_table.create_index(lookup_columns, unique=True)
lookup_table.create_index(lookup_columns, unique=True, if_not_exists=True)
self.db.execute(
"INSERT INTO [{lookup_table}] ({lookup_columns}) SELECT DISTINCT {table_cols} FROM [{table}]".format(
"INSERT OR IGNORE INTO [{lookup_table}] ({lookup_columns}) SELECT DISTINCT {table_cols} FROM [{table}]".format(
lookup_table=table,
lookup_columns=", ".join("[{}]".format(c) for c in lookup_columns),
table_cols=", ".join("[{}]".format(c) for c in columns),

41
tests/test_extract.py
View file

@ -130,3 +130,44 @@ def test_extract_rowid_table(fresh_db):
" FOREIGN KEY(common_name_latin_name_id) REFERENCES common_name_latin_name(id)\n"
")"
)
def test_reuse_lookup_table(fresh_db):
fresh_db["species"].insert({"id": 1, "name": "Wolf"}, pk="id")
fresh_db["sightings"].insert({"id": 10, "species": "Wolf"}, pk="id")
fresh_db["individuals"].insert(
{"id": 10, "name": "Terriana", "species": "Fox"}, pk="id"
)
fresh_db["sightings"].extract("species", rename={"species": "name"})
fresh_db["individuals"].extract("species", rename={"species": "name"})
assert fresh_db["sightings"].schema == (
'CREATE TABLE "sightings" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [species_id] INTEGER,\n"
" FOREIGN KEY(species_id) REFERENCES species(id)\n"
")"
)
assert fresh_db["individuals"].schema == (
'CREATE TABLE "individuals" (\n'
" [id] INTEGER PRIMARY KEY,\n"
" [name] TEXT,\n"
" [species_id] INTEGER,\n"
" FOREIGN KEY(species_id) REFERENCES species(id)\n"
")"
)
assert list(fresh_db["species"].rows) == [
{"id": 1, "name": "Wolf"},
{"id": 2, "name": "Fox"},
]
def test_extract_error_on_incompatible_existing_lookup_table(fresh_db):
fresh_db["species"].insert({"id": 1})
fresh_db["tree"].insert({"name": "Tree 1", "common_name": "Palm"})
with pytest.raises(InvalidColumns):
fresh_db["tree"].extract("common_name", table="species")
# Try again with incompatible existing column type
fresh_db["species2"].insert({"id": 1, "common_name": 3.5})
with pytest.raises(InvalidColumns):
fresh_db["tree"].extract("common_name", table="species2")