mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-07-26 10:54:32 +02:00
Initial Claude implementation of list mode
From https://github.com/simonw/research/tree/main/sqlite-utils-iterator-support I made very minor changes to the test (removing a sys.path insert) Refs #672 Refs https://github.com/simonw/research/pull/31
This commit is contained in:
parent
c479ca0f44
commit
bf354b776c
2 changed files with 276 additions and 37 deletions
175
tests/test_list_mode.py
Normal file
175
tests/test_list_mode.py
Normal file
|
|
@ -0,0 +1,175 @@
|
|||
"""
|
||||
Tests for list-based iteration in insert_all and upsert_all
|
||||
"""
|
||||
import pytest
|
||||
from sqlite_utils import Database
|
||||
|
||||
|
||||
def test_insert_all_list_mode_basic():
|
||||
"""Test basic insert_all with list-based iteration"""
|
||||
db = Database(memory=True)
|
||||
|
||||
def data_generator():
|
||||
# First yield column names
|
||||
yield ["id", "name", "age"]
|
||||
# Then yield data rows
|
||||
yield [1, "Alice", 30]
|
||||
yield [2, "Bob", 25]
|
||||
yield [3, "Charlie", 35]
|
||||
|
||||
db["people"].insert_all(data_generator())
|
||||
|
||||
rows = list(db["people"].rows)
|
||||
assert len(rows) == 3
|
||||
assert rows[0] == {"id": 1, "name": "Alice", "age": 30}
|
||||
assert rows[1] == {"id": 2, "name": "Bob", "age": 25}
|
||||
assert rows[2] == {"id": 3, "name": "Charlie", "age": 35}
|
||||
|
||||
|
||||
def test_insert_all_list_mode_with_pk():
|
||||
"""Test insert_all with list mode and primary key"""
|
||||
db = Database(memory=True)
|
||||
|
||||
def data_generator():
|
||||
yield ["id", "name", "score"]
|
||||
yield [1, "Alice", 95]
|
||||
yield [2, "Bob", 87]
|
||||
|
||||
db["scores"].insert_all(data_generator(), pk="id")
|
||||
|
||||
assert db["scores"].pks == ["id"]
|
||||
rows = list(db["scores"].rows)
|
||||
assert len(rows) == 2
|
||||
|
||||
|
||||
def test_upsert_all_list_mode():
|
||||
"""Test upsert_all with list-based iteration"""
|
||||
db = Database(memory=True)
|
||||
|
||||
# Initial insert
|
||||
def initial_data():
|
||||
yield ["id", "name", "value"]
|
||||
yield [1, "Alice", 100]
|
||||
yield [2, "Bob", 200]
|
||||
|
||||
db["data"].insert_all(initial_data(), pk="id")
|
||||
|
||||
# Upsert with some updates and new records
|
||||
def upsert_data():
|
||||
yield ["id", "name", "value"]
|
||||
yield [1, "Alice", 150] # Update existing
|
||||
yield [3, "Charlie", 300] # Insert new
|
||||
|
||||
db["data"].upsert_all(upsert_data(), pk="id")
|
||||
|
||||
rows = list(db["data"].rows_where(order_by="id"))
|
||||
assert len(rows) == 3
|
||||
assert rows[0] == {"id": 1, "name": "Alice", "value": 150}
|
||||
assert rows[1] == {"id": 2, "name": "Bob", "value": 200}
|
||||
assert rows[2] == {"id": 3, "name": "Charlie", "value": 300}
|
||||
|
||||
|
||||
def test_list_mode_with_various_types():
|
||||
"""Test list mode with different data types"""
|
||||
db = Database(memory=True)
|
||||
|
||||
def data_generator():
|
||||
yield ["id", "name", "score", "active"]
|
||||
yield [1, "Alice", 95.5, True]
|
||||
yield [2, "Bob", 87.3, False]
|
||||
yield [3, "Charlie", None, True]
|
||||
|
||||
db["mixed"].insert_all(data_generator())
|
||||
|
||||
rows = list(db["mixed"].rows)
|
||||
assert len(rows) == 3
|
||||
assert rows[0]["score"] == 95.5
|
||||
assert rows[1]["active"] == 0 # SQLite stores boolean as int
|
||||
assert rows[2]["score"] is None
|
||||
|
||||
|
||||
def test_list_mode_error_non_string_columns():
|
||||
"""Test that non-string column names raise an error"""
|
||||
db = Database(memory=True)
|
||||
|
||||
def bad_data():
|
||||
yield [1, 2, 3] # Non-string column names
|
||||
yield ["a", "b", "c"]
|
||||
|
||||
with pytest.raises(ValueError, match="must be a list of column name strings"):
|
||||
db["bad"].insert_all(bad_data())
|
||||
|
||||
|
||||
def test_list_mode_error_mixed_types():
|
||||
"""Test that mixing list and dict raises an error"""
|
||||
db = Database(memory=True)
|
||||
|
||||
def bad_data():
|
||||
yield ["id", "name"]
|
||||
yield {"id": 1, "name": "Alice"} # Should be a list, not dict
|
||||
|
||||
with pytest.raises(ValueError, match="must also be lists"):
|
||||
db["bad"].insert_all(bad_data())
|
||||
|
||||
|
||||
def test_list_mode_empty_after_headers():
|
||||
"""Test that only headers without data works gracefully"""
|
||||
db = Database(memory=True)
|
||||
|
||||
def data_generator():
|
||||
yield ["id", "name", "age"]
|
||||
# No data rows
|
||||
|
||||
result = db["people"].insert_all(data_generator())
|
||||
assert result is not None
|
||||
assert not db["people"].exists()
|
||||
|
||||
|
||||
def test_list_mode_batch_processing():
|
||||
"""Test list mode with large dataset requiring batching"""
|
||||
db = Database(memory=True)
|
||||
|
||||
def large_data():
|
||||
yield ["id", "value"]
|
||||
for i in range(1000):
|
||||
yield [i, f"value_{i}"]
|
||||
|
||||
db["large"].insert_all(large_data(), batch_size=100)
|
||||
|
||||
count = db.execute("SELECT COUNT(*) as c FROM large").fetchone()[0]
|
||||
assert count == 1000
|
||||
|
||||
|
||||
def test_list_mode_shorter_rows():
|
||||
"""Test that rows shorter than column list get NULL values"""
|
||||
db = Database(memory=True)
|
||||
|
||||
def data_generator():
|
||||
yield ["id", "name", "age", "city"]
|
||||
yield [1, "Alice", 30, "NYC"]
|
||||
yield [2, "Bob"] # Missing age and city
|
||||
yield [3, "Charlie", 35] # Missing city
|
||||
|
||||
db["people"].insert_all(data_generator())
|
||||
|
||||
rows = list(db["people"].rows_where(order_by="id"))
|
||||
assert rows[0] == {"id": 1, "name": "Alice", "age": 30, "city": "NYC"}
|
||||
assert rows[1] == {"id": 2, "name": "Bob", "age": None, "city": None}
|
||||
assert rows[2] == {"id": 3, "name": "Charlie", "age": 35, "city": None}
|
||||
|
||||
|
||||
def test_backwards_compatibility_dict_mode():
|
||||
"""Ensure dict mode still works (backward compatibility)"""
|
||||
db = Database(memory=True)
|
||||
|
||||
# Traditional dict-based insert
|
||||
data = [
|
||||
{"id": 1, "name": "Alice", "age": 30},
|
||||
{"id": 2, "name": "Bob", "age": 25},
|
||||
]
|
||||
|
||||
db["people"].insert_all(data)
|
||||
|
||||
rows = list(db["people"].rows)
|
||||
assert len(rows) == 2
|
||||
assert rows[0] == {"id": 1, "name": "Alice", "age": 30}
|
||||
Loading…
Add table
Add a link
Reference in a new issue