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

insert --replace and insert(..., replace=True)

Browse source 
Refs #66
This commit is contained in:
Simon Willison 2019-12-27 09:15:31 +00:00
commit 607a2a9ff6
5 changed files with 45 additions and 67 deletions
Download patch file Download diff file Expand all files Collapse all files

18
tests/test_cli.py
View file

@ -581,23 +581,23 @@ def test_only_allow_one_of_nl_tsv_csv(options, db_path, tmpdir):
assert "Error: Use just one of --nl, --csv or --tsv" == result.output.strip()
def test_upsert(db_path, tmpdir):
def test_insert_replace(db_path, tmpdir):
test_insert_multiple_with_primary_key(db_path, tmpdir)
json_path = str(tmpdir / "upsert.json")
json_path = str(tmpdir / "insert-replace.json")
db = Database(db_path)
assert 20 == db["dogs"].count
upsert_dogs = [
{"id": 1, "name": "Upserted 1", "age": 4},
{"id": 2, "name": "Upserted 2", "age": 4},
insert_replace_dogs = [
{"id": 1, "name": "Insert replaced 1", "age": 4},
{"id": 2, "name": "Insert replaced 2", "age": 4},
{"id": 21, "name": "Fresh insert 21", "age": 6},
]
open(json_path, "w").write(json.dumps(upsert_dogs))
open(json_path, "w").write(json.dumps(insert_replace_dogs))
result = CliRunner().invoke(
cli.cli, ["upsert", db_path, "dogs", json_path, "--pk", "id"]
cli.cli, ["insert", db_path, "dogs", json_path, "--pk", "id", "--replace"]
)
assert 0 == result.exit_code
assert 0 == result.exit_code, result.output
assert 21 == db["dogs"].count
assert upsert_dogs == db.execute_returning_dicts(
assert insert_replace_dogs == db.execute_returning_dicts(
"select * from dogs where id in (1, 2, 21) order by id"
)

16
tests/test_create.py
View file

@ -445,7 +445,7 @@ def test_insert_row_alter_table(
@pytest.mark.parametrize("use_table_factory", [True, False])
def test_upsert_rows_alter_table(fresh_db, use_table_factory):
def test_insert_replace_rows_alter_table(fresh_db, use_table_factory):
first_row = {"id": 1, "title": "Hedgehogs of the world", "author_id": 1}
next_rows = [
{"id": 1, "title": "Hedgehogs of the World", "species": "hedgehogs"},
@ -459,11 +459,11 @@ def test_upsert_rows_alter_table(fresh_db, use_table_factory):
if use_table_factory:
table = fresh_db.table("books", pk="id", alter=True)
table.insert(first_row)
table.upsert_all(next_rows)
table.insert_all(next_rows, replace=True)
else:
table = fresh_db["books"]
table.insert(first_row, pk="id")
table.upsert_all(next_rows, alter=True)
table.insert_all(next_rows, alter=True, replace=True)
assert {
"author_id": int,
"id": int,
@ -664,11 +664,11 @@ def test_insert_ignore(fresh_db):
def test_insert_hash_id(fresh_db):
dogs = fresh_db["dogs"]
id = dogs.upsert({"name": "Cleo", "twitter": "cleopaws"}, hash_id="id").last_pk
id = dogs.insert({"name": "Cleo", "twitter": "cleopaws"}, hash_id="id").last_pk
assert "f501265970505d9825d8d9f590bfab3519fb20b1" == id
assert 1 == dogs.count
# Upserting a second time should not create a new row
id2 = dogs.upsert({"name": "Cleo", "twitter": "cleopaws"}, hash_id="id").last_pk
# Insert replacing a second time should not create a new row
id2 = dogs.insert({"name": "Cleo", "twitter": "cleopaws"}, hash_id="id", replace=True).last_pk
assert "f501265970505d9825d8d9f590bfab3519fb20b1" == id2
assert 1 == dogs.count
@ -791,10 +791,10 @@ def test_drop_view(fresh_db):
assert [] == fresh_db.view_names()
def test_insert_upsert_all_empty_list(fresh_db):
def test_insert_all_empty_list(fresh_db):
fresh_db["t"].insert({"foo": 1})
assert 1 == fresh_db["t"].count
fresh_db["t"].insert_all([])
assert 1 == fresh_db["t"].count
fresh_db["t"].upsert_all([])
fresh_db["t"].insert_all([], replace=True)
assert 1 == fresh_db["t"].count