From a2e6ea2eca5f2a49c116c9af1a37aba4b1a7eaac Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 19:19:24 +0000 Subject: [PATCH] Raise ValueError instead of AssertionError for invalid arguments User-facing argument validation in db.py previously used bare assert statements, which vanish entirely under python -O and raise AssertionError - an exception type callers should not have to catch for input validation. Fifteen validation sites now raise ValueError with the same messages. Internal invariants (an unreachable branch and a post-update rowcount check) remain asserts. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01UnLnhsH25Nnv7LHhekUfPd --- docs/changelog.rst | 1 + sqlite_utils/db.py | 123 ++++++++++++++++++++++---------------- tests/test_convert.py | 2 +- tests/test_create.py | 12 ++-- tests/test_create_view.py | 2 +- tests/test_m2m.py | 4 +- tests/test_recreate.py | 2 +- 7 files changed, 83 insertions(+), 63 deletions(-) diff --git a/docs/changelog.rst b/docs/changelog.rst index ce1397f..0b41875 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -9,6 +9,7 @@ Unreleased ---------- +- **Breaking change**: Python API validation errors now raise ``ValueError`` instead of ``AssertionError``. Previously invalid arguments - such as ``create_table()`` with no columns, ``transform()`` on a table that does not exist, or passing both ``ignore=True`` and ``replace=True`` - were rejected using bare ``assert`` statements, which are silently skipped when Python runs with the ``-O`` flag. Code that caught ``AssertionError`` for these cases should catch ``ValueError`` instead. - **Breaking change**: The no-op ``-d/--detect-types`` flag has been removed from the ``insert`` and ``upsert`` commands. Type detection has been the default for CSV/TSV data since 4.0a1, so the flag did nothing - invocations using it should simply drop it. ``--no-detect-types`` remains available to disable detection. - **Breaking change**: ``db.query()`` now executes its SQL as soon as it is called, rather than waiting until the returned generator is first iterated. Rows are still fetched lazily during iteration. SQL errors are now raised at the call site, statements such as ``INSERT ... RETURNING`` take effect without needing to iterate over their results, and passing a statement that returns no rows - previously a silent no-op - now raises a ``ValueError`` recommending ``db.execute()`` instead. - Fixed a bug where ``table.delete_where()``, ``table.optimize()`` and ``table.rebuild_fts()`` did not commit their changes, leaving the connection inside an open transaction. Their work - and any subsequent writes - could then be silently rolled back when the connection was closed. All three now use ``db.atomic()``, consistent with the other write methods. diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py index b53df51..a0ca6ed 100644 --- a/sqlite_utils/db.py +++ b/sqlite_utils/db.py @@ -367,9 +367,11 @@ class Database: self.memory_name = None self.memory = False self.use_old_upsert = use_old_upsert - assert (filename_or_conn is not None and (not memory and not memory_name)) or ( - filename_or_conn is None and (memory or memory_name) - ), "Either specify a filename_or_conn or pass memory=True" + if not ( + (filename_or_conn is not None and (not memory and not memory_name)) + or (filename_or_conn is None and (memory or memory_name)) + ): + raise ValueError("Either specify a filename_or_conn or pass memory=True") if memory_name: uri = "file:{}?mode=memory&cache=shared".format(memory_name) self.conn = sqlite3.connect( @@ -393,7 +395,8 @@ class Database: raise self.conn = sqlite3.connect(str(filename_or_conn)) else: - assert not recreate, "recreate cannot be used with connections, only paths" + if recreate: + raise ValueError("recreate cannot be used with connections, only paths") self.conn = cast(sqlite3.Connection, filename_or_conn) self._tracer: Optional[Tracer] = tracer if recursive_triggers: @@ -968,20 +971,21 @@ class Database: other_column = table.guess_foreign_column(other_table) fks.append(ForeignKey(name, column, other_table, other_column)) return fks - assert all( - isinstance(fk, (tuple, list)) for fk in foreign_keys - ), "foreign_keys= should be a list of tuples" + if not all(isinstance(fk, (tuple, list)) for fk in foreign_keys): + raise ValueError("foreign_keys= should be a list of tuples") fks = [] for tuple_or_list in foreign_keys: if len(tuple_or_list) == 4: - assert ( - tuple_or_list[0] == name - ), "First item in {} should have been {}".format(tuple_or_list, name) - assert len(tuple_or_list) in ( - 2, - 3, - 4, - ), "foreign_keys= should be a list of tuple pairs or triples" + if tuple_or_list[0] != name: + raise ValueError( + "First item in {} should have been {}".format( + tuple_or_list, name + ) + ) + if len(tuple_or_list) not in (2, 3, 4): + raise ValueError( + "foreign_keys= should be a list of tuple pairs or triples" + ) if len(tuple_or_list) in (3, 4): if len(tuple_or_list) == 4: tuple_or_list = cast(Tuple[str, str, str], tuple_or_list[1:]) @@ -1054,17 +1058,20 @@ class Database: # Soundness check not_null, and defaults if provided not_null = not_null or set() defaults = defaults or {} - assert columns, "Tables must have at least one column" - assert all( - n in columns for n in not_null - ), "not_null set {} includes items not in columns {}".format( - repr(not_null), repr(set(columns.keys())) - ) - assert all( - n in columns for n in defaults - ), "defaults set {} includes items not in columns {}".format( - repr(set(defaults)), repr(set(columns.keys())) - ) + if not columns: + raise ValueError("Tables must have at least one column") + if not all(n in columns for n in not_null): + raise ValueError( + "not_null set {} includes items not in columns {}".format( + repr(not_null), repr(set(columns.keys())) + ) + ) + if not all(n in columns for n in defaults): + raise ValueError( + "defaults set {} includes items not in columns {}".format( + repr(set(defaults)), repr(set(columns.keys())) + ) + ) column_items = list(columns.items()) if column_order is not None: @@ -1295,9 +1302,8 @@ class Database: :param ignore: Set to ``True`` to do nothing if a view with this name already exists :param replace: Set to ``True`` to replace the view if one with this name already exists """ - assert not ( - ignore and replace - ), "Use one or the other of ignore/replace, not both" + if ignore and replace: + raise ValueError("Use one or the other of ignore/replace, not both") create_sql = "CREATE VIEW {name} AS {sql}".format( name=quote_identifier(name), sql=sql ) @@ -1342,9 +1348,13 @@ class Database: tuples """ # foreign_keys is a list of explicit 4-tuples - assert all( + if not all( len(fk) == 4 and isinstance(fk, (list, tuple)) for fk in foreign_keys - ), "foreign_keys must be a list of 4-tuples, (table, column, other_table, other_column)" + ): + raise ValueError( + "foreign_keys must be a list of 4-tuples, " + "(table, column, other_table, other_column)" + ) foreign_keys_to_create = [] @@ -1989,7 +1999,8 @@ class Table(Queryable): :param keep_table: If specified, the existing table will be renamed to this and will not be dropped """ - assert self.exists(), "Cannot transform a table that doesn't exist yet" + if not self.exists(): + raise ValueError("Cannot transform a table that doesn't exist yet") sqls = self.transform_sql( types=types, rename=rename, @@ -2161,8 +2172,10 @@ class Table(Queryable): elif not not_null: pass else: - assert False, "not_null must be a dict or a set or None, it was {}".format( - repr(not_null) + raise ValueError( + "not_null must be a dict or a set or None, it was {}".format( + repr(not_null) + ) ) # defaults= create_table_defaults = { @@ -2870,9 +2883,10 @@ class Table(Queryable): "{}.{}".format(original_quoted, quote_identifier(c)) for c in columns ) fts_table = self.detect_fts() - assert fts_table, "Full-text search is not configured for table '{}'".format( - self.name - ) + if not fts_table: + raise ValueError( + "Full-text search is not configured for table '{}'".format(self.name) + ) fts_table_quoted = quote_identifier(fts_table) virtual_table_using = self.db.table(fts_table).virtual_table_using sql = textwrap.dedent(""" @@ -3119,7 +3133,8 @@ class Table(Queryable): ) if output is not None: - assert len(columns) == 1, "output= can only be used with a single column" + if len(columns) != 1: + raise ValueError("output= can only be used with a single column") if output not in self.columns_dict: self.add_column(output, output_type or "text") @@ -3632,15 +3647,15 @@ class Table(Queryable): if upsert and (not pk and not hash_id): raise PrimaryKeyRequired("upsert() requires a pk") - assert not (hash_id and pk), "Use either pk= or hash_id=" + if hash_id and pk: + raise ValueError("Use either pk= or hash_id=") if hash_id_columns and (hash_id is None): hash_id = "id" if hash_id: pk = hash_id - assert not ( - ignore and replace - ), "Use either ignore=True or replace=True, not both" + if ignore and replace: + raise ValueError("Use either ignore=True or replace=True, not both") all_columns = [] first = True num_records_processed = 0 @@ -3689,9 +3704,10 @@ class Table(Queryable): first_record = cast(Dict[str, Any], first_record) num_columns = len(first_record.keys()) - assert ( - num_columns <= SQLITE_MAX_VARS - ), "Rows can have a maximum of {} columns".format(SQLITE_MAX_VARS) + if num_columns > SQLITE_MAX_VARS: + raise ValueError( + "Rows can have a maximum of {} columns".format(SQLITE_MAX_VARS) + ) batch_size = ( 1 if num_columns == 0 @@ -3946,10 +3962,12 @@ class Table(Queryable): :param extra_values: Additional column values to be used only if creating a new record :param strict: Boolean, apply STRICT mode if creating the table. """ - assert isinstance(lookup_values, dict) - assert pk is not None - if extra_values is not None: - assert isinstance(extra_values, dict) + if not isinstance(lookup_values, dict): + raise ValueError("lookup_values must be a dictionary") + if pk is None: + raise ValueError("pk cannot be None") + if extra_values is not None and not isinstance(extra_values, dict): + raise ValueError("extra_values must be a dictionary") combined_values = dict(lookup_values) if extra_values is not None: combined_values.update(extra_values) @@ -4034,9 +4052,10 @@ class Table(Queryable): other_table = self.db.table(other_table, pk=pk) our_id = self.last_pk if lookup is not None: - assert record_or_iterable is None, "Provide lookup= or record, not both" - else: - assert record_or_iterable is not None, "Provide lookup= or record, not both" + if record_or_iterable is not None: + raise ValueError("Provide lookup= or record, not both") + elif record_or_iterable is None: + raise ValueError("Provide lookup= or record, not both") tables = list(sorted([self.name, other_table.name])) columns = ["{}_id".format(t) for t in tables] if m2m_table is not None: diff --git a/tests/test_convert.py b/tests/test_convert.py index 848b39a..ea3fd96 100644 --- a/tests/test_convert.py +++ b/tests/test_convert.py @@ -77,7 +77,7 @@ def test_convert_output(fresh_db, drop, expected): def test_convert_output_multiple_column_error(fresh_db): table = fresh_db["table"] - with pytest.raises(AssertionError) as excinfo: + with pytest.raises(ValueError) as excinfo: table.convert(["title", "other"], lambda v: v, output="out") assert "output= can only be used with a single column" in str(excinfo.value) diff --git a/tests/test_create.py b/tests/test_create.py index afa7a65..decefcf 100644 --- a/tests/test_create.py +++ b/tests/test_create.py @@ -108,7 +108,7 @@ def test_create_table_with_defaults(fresh_db): def test_create_table_with_bad_not_null(fresh_db): - with pytest.raises(AssertionError): + with pytest.raises(ValueError): fresh_db.create_table( "players", {"name": str, "score": int}, not_null={"mouse"} ) @@ -243,11 +243,11 @@ def test_create_table_column_order(fresh_db, use_table_factory): # If you specify a column that doesn't point to a table, you get an error: (("one_id", "two_id", "three_id"), NoObviousTable), # Tuples of the wrong length get an error: - ((("one_id", "one", "id", "five"), ("two_id", "two", "id")), AssertionError), + ((("one_id", "one", "id", "five"), ("two_id", "two", "id")), ValueError), # Likewise a bad column: ((("one_id", "one", "id2"),), AlterError), # Or a list of dicts - (({"one_id": "one"},), AssertionError), + (({"one_id": "one"},), ValueError), ), ) @pytest.mark.parametrize("use_table_factory", [True, False]) @@ -700,7 +700,7 @@ def test_bulk_insert_more_than_999_values(fresh_db): def test_error_if_more_than_999_columns(fresh_db, num_columns, should_error): record = dict([("c{}".format(i), i) for i in range(num_columns)]) if should_error: - with pytest.raises(AssertionError): + with pytest.raises(ValueError): fresh_db["big"].insert(record) else: fresh_db["big"].insert(record) @@ -1061,7 +1061,7 @@ def test_create_table_numpy(fresh_db): def test_cannot_provide_both_filename_and_memory(): with pytest.raises( - AssertionError, match="Either specify a filename_or_conn or pass memory=True" + ValueError, match="Either specify a filename_or_conn or pass memory=True" ): Database("/tmp/foo.db", memory=True) @@ -1214,7 +1214,7 @@ def test_create_if_not_exists(fresh_db): def test_create_if_no_columns(fresh_db): - with pytest.raises(AssertionError) as error: + with pytest.raises(ValueError) as error: fresh_db["t"].create({}) assert error.value.args[0] == "Tables must have at least one column" diff --git a/tests/test_create_view.py b/tests/test_create_view.py index f0e2855..056e246 100644 --- a/tests/test_create_view.py +++ b/tests/test_create_view.py @@ -15,7 +15,7 @@ def test_create_view_error(fresh_db): def test_create_view_only_arrow_one_param(fresh_db): - with pytest.raises(AssertionError): + with pytest.raises(ValueError): fresh_db.create_view("bar", "select 1 + 2", ignore=True, replace=True) diff --git a/tests/test_m2m.py b/tests/test_m2m.py index 2cb2ca3..d613bb9 100644 --- a/tests/test_m2m.py +++ b/tests/test_m2m.py @@ -139,9 +139,9 @@ def test_m2m_lookup(fresh_db): def test_m2m_requires_either_records_or_lookup(fresh_db): people = fresh_db.table("people", pk="id").insert({"name": "Wahyu"}) - with pytest.raises(AssertionError): + with pytest.raises(ValueError): people.m2m("tags") - with pytest.raises(AssertionError): + with pytest.raises(ValueError): people.m2m("tags", {"tag": "hello"}, lookup={"foo": "bar"}) diff --git a/tests/test_recreate.py b/tests/test_recreate.py index 224d182..bce53d5 100644 --- a/tests/test_recreate.py +++ b/tests/test_recreate.py @@ -15,7 +15,7 @@ def test_recreate_ignored_for_in_memory(): def test_recreate_not_allowed_for_connection(): conn = sqlite3.connect(":memory:") try: - with pytest.raises(AssertionError): + with pytest.raises(ValueError): Database(conn, recreate=True) finally: conn.close()