remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-10-23 17:57:03 +02:00
Code Issues Projects Releases Packages Wiki Activity

Merge 85dd980ccf into 9c66bafa52

Browse source
This commit is contained in:
WU Leizhi 2026-10-11 19:21:13 -07:00 • committed by GitHub
commit 8f08fac085
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 639 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

5
docs/changelog.rst
View file

@ -4,6 +4,11 @@
Changelog
===========
Unreleased
----------
- The Python table creation, insert, upsert and lookup APIs now accept ``autoincrement=`` for single integer primary keys. The ``create-table``, ``insert``, ``upsert`` and ``transform`` commands accept ``--autoincrement`` and ``--no-autoincrement``. Existing table mode changes can be applied or previewed with ``transform()`` and ``transform_sql()``. (:issue:`664`)
.. _v4_2_1:
4.2.1 (2026-08-13)

171
docs/cli-reference.rst
View file

@ -287,40 +287,47 @@ See :ref:`cli_inserting_data`, :ref:`cli_insert_csv_tsv`, :ref:`cli_insert_unstr
' --pk id
Options:
--pk TEXT Columns to use as the primary key, e.g. id
--code TEXT Python code defining a rows() function or iterable
of rows to insert
--flatten Flatten nested JSON objects, so {"a": {"b": 1}}
becomes {"a_b": 1}
--nl Expect newline-delimited JSON
-c, --csv Expect CSV input
--tsv Expect TSV input
--empty-null Treat empty strings as NULL
--lines Treat each line as a single value called 'line'
--text Treat input as a single value called 'text'
--convert TEXT Python code to convert each item
--import TEXT Python modules to import
--delimiter TEXT Delimiter to use for CSV files
--quotechar TEXT Quote character to use for CSV/TSV
--sniff Detect delimiter and quote character
--no-headers CSV file has no header row
--encoding TEXT Character encoding for input, defaults to utf-8
--batch-size INTEGER Commit every X records
--stop-after INTEGER Stop after X records
--alter Alter existing table to add any missing columns
--not-null TEXT Columns that should be created as NOT NULL
--default <TEXT TEXT>... Default value that should be set for a column
--type <TEXT CHOICE>... Column types to use when creating the table
--no-detect-types Treat all CSV/TSV columns as TEXT
--analyze Run ANALYZE at the end of this operation
--load-extension TEXT Path to SQLite extension, with optional :entrypoint
--silent Do not show progress bar
--strict Apply STRICT mode to created table
--ignore Ignore records if pk already exists
--replace Replace records if pk already exists
--truncate Truncate table before inserting records, if table
already exists
-h, --help Show this message and exit.
--pk TEXT Columns to use as the primary key, e.g. id
--code TEXT Python code defining a rows() function or
iterable of rows to insert
--flatten Flatten nested JSON objects, so {"a": {"b":
1}} becomes {"a_b": 1}
--nl Expect newline-delimited JSON
-c, --csv Expect CSV input
--tsv Expect TSV input
--empty-null Treat empty strings as NULL
--lines Treat each line as a single value called
'line'
--text Treat input as a single value called 'text'
--convert TEXT Python code to convert each item
--import TEXT Python modules to import
--delimiter TEXT Delimiter to use for CSV files
--quotechar TEXT Quote character to use for CSV/TSV
--sniff Detect delimiter and quote character
--no-headers CSV file has no header row
--encoding TEXT Character encoding for input, defaults to
utf-8
--batch-size INTEGER Commit every X records
--stop-after INTEGER Stop after X records
--alter Alter existing table to add any missing
columns
--not-null TEXT Columns that should be created as NOT NULL
--default <TEXT TEXT>... Default value that should be set for a column
--type <TEXT CHOICE>... Column types to use when creating the table
--no-detect-types Treat all CSV/TSV columns as TEXT
--analyze Run ANALYZE at the end of this operation
--load-extension TEXT Path to SQLite extension, with optional
:entrypoint
--silent Do not show progress bar
--strict Apply STRICT mode to created table
--autoincrement / --no-autoincrement
Use AUTOINCREMENT for the INTEGER PRIMARY KEY
when creating the table
--ignore Ignore records if pk already exists
--replace Replace records if pk already exists
--truncate Truncate table before inserting records, if
table already exists
-h, --help Show this message and exit.
.. _cli_ref_upsert:
@ -351,36 +358,43 @@ See :ref:`cli_upsert`.
]' | sqlite-utils upsert data.db chickens - --pk id
Options:
--pk TEXT Columns to use as the primary key, e.g. id
--code TEXT Python code defining a rows() function or iterable
of rows to insert
--flatten Flatten nested JSON objects, so {"a": {"b": 1}}
becomes {"a_b": 1}
--nl Expect newline-delimited JSON
-c, --csv Expect CSV input
--tsv Expect TSV input
--empty-null Treat empty strings as NULL
--lines Treat each line as a single value called 'line'
--text Treat input as a single value called 'text'
--convert TEXT Python code to convert each item
--import TEXT Python modules to import
--delimiter TEXT Delimiter to use for CSV files
--quotechar TEXT Quote character to use for CSV/TSV
--sniff Detect delimiter and quote character
--no-headers CSV file has no header row
--encoding TEXT Character encoding for input, defaults to utf-8
--batch-size INTEGER Commit every X records
--stop-after INTEGER Stop after X records
--alter Alter existing table to add any missing columns
--not-null TEXT Columns that should be created as NOT NULL
--default <TEXT TEXT>... Default value that should be set for a column
--type <TEXT CHOICE>... Column types to use when creating the table
--no-detect-types Treat all CSV/TSV columns as TEXT
--analyze Run ANALYZE at the end of this operation
--load-extension TEXT Path to SQLite extension, with optional :entrypoint
--silent Do not show progress bar
--strict Apply STRICT mode to created table
-h, --help Show this message and exit.
--pk TEXT Columns to use as the primary key, e.g. id
--code TEXT Python code defining a rows() function or
iterable of rows to insert
--flatten Flatten nested JSON objects, so {"a": {"b":
1}} becomes {"a_b": 1}
--nl Expect newline-delimited JSON
-c, --csv Expect CSV input
--tsv Expect TSV input
--empty-null Treat empty strings as NULL
--lines Treat each line as a single value called
'line'
--text Treat input as a single value called 'text'
--convert TEXT Python code to convert each item
--import TEXT Python modules to import
--delimiter TEXT Delimiter to use for CSV files
--quotechar TEXT Quote character to use for CSV/TSV
--sniff Detect delimiter and quote character
--no-headers CSV file has no header row
--encoding TEXT Character encoding for input, defaults to
utf-8
--batch-size INTEGER Commit every X records
--stop-after INTEGER Stop after X records
--alter Alter existing table to add any missing
columns
--not-null TEXT Columns that should be created as NOT NULL
--default <TEXT TEXT>... Default value that should be set for a column
--type <TEXT CHOICE>... Column types to use when creating the table
--no-detect-types Treat all CSV/TSV columns as TEXT
--analyze Run ANALYZE at the end of this operation
--load-extension TEXT Path to SQLite extension, with optional
:entrypoint
--silent Do not show progress bar
--strict Apply STRICT mode to created table
--autoincrement / --no-autoincrement
Use AUTOINCREMENT for the INTEGER PRIMARY KEY
when creating the table
-h, --help Show this message and exit.
.. _cli_ref_bulk:
@ -510,6 +524,9 @@ See :ref:`cli_transform_table`.
--drop-foreign-key TEXT Drop foreign key constraint for this column
--strict / --no-strict Enable or disable STRICT mode (default:
preserve current mode)
--autoincrement / --no-autoincrement
Enable or disable AUTOINCREMENT (default:
preserve current mode)
--sql Output SQL without executing it
--load-extension TEXT Path to SQLite extension, with optional
:entrypoint
@ -966,17 +983,21 @@ See :ref:`cli_create_table`.
Valid column types are text, integer, real, float, blob and any.
Options:
--pk TEXT Column to use as primary key
--not-null TEXT Columns that should be created as NOT NULL
--default <TEXT TEXT>... Default value that should be set for a column
--fk <TEXT TEXT TEXT>... Column, other table, other column to set as a
foreign key
--ignore If table already exists, do nothing
--replace If table already exists, replace it
--transform If table already exists, try to transform the schema
--load-extension TEXT Path to SQLite extension, with optional :entrypoint
--strict Apply STRICT mode to created table
-h, --help Show this message and exit.
--pk TEXT Column to use as primary key
--not-null TEXT Columns that should be created as NOT NULL
--default <TEXT TEXT>... Default value that should be set for a column
--fk <TEXT TEXT TEXT>... Column, other table, other column to set as a
foreign key
--ignore If table already exists, do nothing
--replace If table already exists, replace it
--transform If table already exists, try to transform the
schema
--load-extension TEXT Path to SQLite extension, with optional
:entrypoint
--strict Apply STRICT mode to created table
--autoincrement / --no-autoincrement
Use AUTOINCREMENT for the INTEGER PRIMARY KEY
-h, --help Show this message and exit.
.. _cli_ref_create_index:

19
docs/cli.rst
View file

@ -2079,6 +2079,22 @@ You can pass as many column-name column-type pairs as you like. Valid types are
Pass ``--pk`` more than once for a compound primary key that covers multiple columns.
For a single integer primary key, add ``--autoincrement`` to prevent reuse of deleted IDs:
.. code-block:: bash
sqlite-utils create-table events.db events id integer name text --pk id --autoincrement
The option also works when ``insert`` or ``upsert`` creates a new table:
.. code-block:: bash
echo '{"name": "first event"}' | sqlite-utils insert events.db events - --pk id --autoincrement
An omitted primary key column is created as an integer column. ``upsert`` still requires primary key values in the input. Compound and non-integer primary keys cannot use this option. Ordinary integer primary keys already generate missing IDs; ``AUTOINCREMENT`` prevents their reuse and does not guarantee consecutive values. Use ``--no-autoincrement`` to explicitly create a table without it.
Insert options do not change the mode of an existing table. A conflicting explicit mode is rejected; use ``transform --autoincrement`` or ``transform --no-autoincrement`` to change it. ``create-table --transform`` also applies an explicit mode change.
You can specify columns that should be NOT NULL using ``--not-null colname``. You can specify default values for columns using ``--default colname defaultvalue``.
.. code-block:: bash
@ -2271,6 +2287,9 @@ Every option for this table (with the exception of ``--pk-none``) can be specifi
``--add-foreign-key column other_table other_column``
Add a foreign key constraint to ``column`` pointing to ``other_table.other_column``.
``--autoincrement`` / ``--no-autoincrement``
Enable or disable ``AUTOINCREMENT`` for a single integer primary key. If neither option is supplied the current mode is preserved. Enabling it tracks the current IDs, but cannot recover IDs deleted before it was enabled. ``--sql`` previews the same change without applying it.
``--strict``
Convert the table to a `SQLite STRICT table <https://www.sqlite.org/stricttables.html>`__. The command fails if the available SQLite version does not support strict tables. If existing rows contain values that are incompatible with their declared column types the transformation fails and the original table is left unchanged.

27
docs/python-api.rst
View file

@ -844,6 +844,33 @@ An ``ANY`` column can store integers, floating point values, text, binary data o
.. note::
In the CLI: :ref:`sqlite-utils create-table <cli_create_table>`
.. _python_api_autoincrement:
Preventing reuse of deleted IDs
-------------------------------
An ``INTEGER PRIMARY KEY`` already generates an ID when an inserted record omits that column. Pass ``autoincrement=True`` to prevent SQLite from reusing IDs of deleted rows:
.. code-block:: python
table = db["events"].create({"id": int, "name": str}, pk="id", autoincrement=True)
table.insert({"name": "first event"})
This option also works with ``db.create_table()``, ``db.create_table_sql()``, ``insert()``, ``insert_all()``, ``upsert()``, ``upsert_all()`` and ``lookup()``. To create the table on the first insert:
.. code-block:: python
db["events"].insert({"name": "first event"}, pk="id", autoincrement=True)
You can configure it as a table default using ``db.table("events", pk="id", autoincrement=True)``. The primary key must be a single ``INTEGER`` column; compound, text and hash primary keys cannot use this option. A primary key omitted from the column definitions is created as an integer column. ``upsert()`` still requires primary key values in every record.
The default ``None`` keeps the usual ID allocation when creating a new table. An explicit ``False`` creates a table without ``AUTOINCREMENT``. An explicit option that conflicts with an existing table is rejected by inserts and lookups; use ``table.transform(autoincrement=True)`` or ``table.transform(autoincrement=False)`` to change the mode. ``create(..., transform=True)`` also applies an explicit mode change. Ordinary transformations preserve the existing mode and its sequence.
Enabling the mode on an existing table starts tracking its current IDs. It cannot recover the history of IDs deleted before it was enabled. IDs need not be consecutive. SQLite documents the additional cost of this mode in `its AUTOINCREMENT documentation <https://www.sqlite.org/autoinc.html>`__.
.. note::
In the CLI: :ref:`sqlite-utils create-table <cli_create_table>` and :ref:`sqlite-utils transform <cli_transform_table>`.
.. _python_api_compound_primary_keys:
Compound primary keys

26
sqlite_utils/cli.py
View file

@ -1050,6 +1050,11 @@ def insert_upsert_options(*, require_pk=False):
default=False,
help="Apply STRICT mode to created table",
),
click.option(
"--autoincrement/--no-autoincrement",
default=None,
help="Use AUTOINCREMENT for the INTEGER PRIMARY KEY when creating the table",
),
)
):
fn = decorator(fn)
@ -1095,6 +1100,7 @@ def insert_upsert_implementation(
functions=None,
strict=False,
code=None,
autoincrement=None,
):
db = sqlite_utils.Database(path)
_register_db_for_cleanup(db)
@ -1109,6 +1115,7 @@ def insert_upsert_implementation(
"truncate": truncate,
"analyze": analyze,
"strict": strict,
"autoincrement": autoincrement,
}
if not_null:
extra_kwargs["not_null"] = set(not_null)
@ -1386,6 +1393,7 @@ def insert(
default,
types,
strict,
autoincrement,
):
"""
Insert records from FILE into a table, creating the table if it
@ -1480,6 +1488,7 @@ def insert(
default=default,
types=types,
strict=strict,
autoincrement=autoincrement,
code=code,
)
except UnicodeDecodeError as ex:
@ -1519,6 +1528,7 @@ def upsert(
load_extension,
silent,
strict,
autoincrement,
):
"""
Upsert records based on their primary key. Works like 'insert' but if
@ -1570,6 +1580,7 @@ def upsert(
load_extension=load_extension,
silent=silent,
strict=strict,
autoincrement=autoincrement,
code=code,
)
except UnicodeDecodeError as ex:
@ -1743,6 +1754,11 @@ def create_database(path, enable_wal, init_spatialite, load_extension):
is_flag=True,
help="Apply STRICT mode to created table",
)
@click.option(
"--autoincrement/--no-autoincrement",
default=None,
help="Use AUTOINCREMENT for the INTEGER PRIMARY KEY",
)
def create_table(
path,
table,
@ -1756,6 +1772,7 @@ def create_table(
transform,
load_extension,
strict,
autoincrement,
):
"""
Add a table with the specified columns. Columns should be specified using
@ -1802,6 +1819,7 @@ def create_table(
replace=replace,
transform=transform,
strict=strict,
autoincrement=autoincrement,
)
@ -2724,6 +2742,11 @@ def schema(
default=None,
help="Enable or disable STRICT mode (default: preserve current mode)",
)
@click.option(
"--autoincrement/--no-autoincrement",
default=None,
help="Enable or disable AUTOINCREMENT (default: preserve current mode)",
)
@click.option("--sql", is_flag=True, help="Output SQL without executing it")
@load_extension_option
def transform(
@ -2744,6 +2767,7 @@ def transform(
strict,
sql,
load_extension,
autoincrement,
):
"""Transform a table beyond the capabilities of ALTER TABLE
@ -2804,6 +2828,7 @@ def transform(
drop_foreign_keys=drop_foreign_keys_value,
add_foreign_keys=add_foreign_keys_value,
strict=strict,
autoincrement=autoincrement,
):
click.echo(line)
else:
@ -2818,6 +2843,7 @@ def transform(
drop_foreign_keys=drop_foreign_keys_value,
add_foreign_keys=add_foreign_keys_value,
strict=strict,
autoincrement=autoincrement,
)

115
sqlite_utils/db.py
View file

@ -1444,6 +1444,7 @@ class Database:
_column_comments: Mapping[str, ColumnComments] | None = None,
_autoincrement: str | None = None,
_uniques: Iterable[Unique] | None = None,
autoincrement: bool | None = None,
) -> str:
"""
Returns the SQL ``CREATE TABLE`` statement for creating the specified table.
@ -1460,7 +1461,12 @@ class Database:
:param extracts: List or dictionary of columns to be extracted during inserts, see :ref:`python_api_extracts`
:param if_not_exists: Use ``CREATE TABLE IF NOT EXISTS``
:param strict: Apply STRICT mode to table
:param autoincrement: Use AUTOINCREMENT for the single INTEGER PRIMARY KEY.
"""
if autoincrement:
_validate_autoincrement_columns(
pk, columns, hash_id, hash_id_columns, extracts
)
if hash_id_columns and (hash_id is None):
hash_id = "id"
resolved_fks: list[ForeignKey] = [
@ -1601,6 +1607,10 @@ class Database:
column_items.insert(0, (pk, int))
elif pk:
pk = [resolve_casing(p, [c[0] for c in column_items]) for p in pk]
if autoincrement is not None:
if autoincrement and single_pk is None:
raise ValueError("AUTOINCREMENT requires a single-column primary key")
_autoincrement = single_pk if autoincrement else None
if _autoincrement is not None:
_autoincrement = resolve_casing(
_autoincrement, [c[0] for c in column_items]
@ -1714,6 +1724,7 @@ class Database:
ignore: bool = False,
transform: bool = False,
strict: bool = False,
autoincrement: bool | None = None,
) -> "Table":
"""
Create a table with the specified name and the specified ``{column_name: type}`` columns.
@ -1735,16 +1746,25 @@ class Database:
:param ignore: Silently do nothing if table already exists
:param transform: If table already exists transform it to fit the specified schema
:param strict: Apply STRICT mode to table
:param autoincrement: Use AUTOINCREMENT for the single INTEGER PRIMARY KEY.
"""
# Transform table to match the new definition if table already exists:
if self[name].exists():
if ignore:
return self.table(name)
elif replace:
self[name].drop()
if autoincrement:
_validate_autoincrement_columns(
pk, columns, hash_id, hash_id_columns, extracts
)
if replace and self[name].exists():
self[name].drop()
if transform and self[name].exists():
table = self.table(name)
should_transform = False
if autoincrement is not None:
should_transform = (
bool(parse_autoincrement(table.schema)) != autoincrement
)
# First add missing columns and figure out columns to drop
existing_columns = table.columns_dict
# Match existing columns case-insensitively, the way SQLite does
@ -1807,6 +1827,7 @@ class Database:
not_null=not_null,
defaults=defaults,
pk=pk,
autoincrement=autoincrement,
)
return table
sql = self.create_table_sql(
@ -1822,6 +1843,7 @@ class Database:
extracts=extracts,
if_not_exists=if_not_exists,
strict=strict,
autoincrement=autoincrement,
)
self.execute(sql)
return self.table(
@ -1833,6 +1855,7 @@ class Database:
defaults=defaults,
hash_id=hash_id,
hash_id_columns=hash_id_columns,
autoincrement=autoincrement,
)
def rename_table(self, name: str, new_name: str) -> None:
@ -2301,6 +2324,7 @@ class Table(Queryable):
conversions: dict | None = None,
columns: dict[str, Any] | None = None,
strict: bool = False,
autoincrement: bool | None = None,
):
super().__init__(db, name)
self._defaults = {
@ -2319,6 +2343,7 @@ class Table(Queryable):
"conversions": conversions or {},
"columns": columns,
"strict": strict,
"autoincrement": autoincrement,
}
def __repr__(self) -> str:
@ -2553,6 +2578,7 @@ class Table(Queryable):
ignore: bool = False,
transform: bool = False,
strict: bool | Default | None = DEFAULT,
autoincrement: bool | Default | None = DEFAULT,
) -> "Table":
"""
Create a table with the specified columns.
@ -2573,6 +2599,7 @@ class Table(Queryable):
:param ignore: Silently do nothing if table already exists
:param transform: If table already exists transform it to fit the specified schema
:param strict: Apply STRICT mode to table
:param autoincrement: Use AUTOINCREMENT for the single INTEGER PRIMARY KEY.
"""
# Resolve defaults from _defaults (issue #655)
pk = self.value_or_default("pk", pk)
@ -2584,6 +2611,7 @@ class Table(Queryable):
hash_id_columns = self.value_or_default("hash_id_columns", hash_id_columns)
extracts = self.value_or_default("extracts", extracts)
strict = self.value_or_default("strict", strict)
autoincrement = self.value_or_default("autoincrement", autoincrement)
# Store configuration in _defaults for subsequent operations (issue #655)
# Don't store pk if hash_id is set, since pk is derived from hash_id in that case
@ -2605,6 +2633,8 @@ class Table(Queryable):
self._defaults["extracts"] = extracts
if strict:
self._defaults["strict"] = strict
if autoincrement is not None and not (ignore and self.exists()):
self._defaults["autoincrement"] = autoincrement
columns = {name: value for (name, value) in columns.items()}
with self.db.atomic():
@ -2624,6 +2654,7 @@ class Table(Queryable):
ignore=ignore,
transform=transform,
strict=cast(bool, strict),
autoincrement=autoincrement,
)
return self
@ -2655,6 +2686,7 @@ class Table(Queryable):
column_order: list[str] | None = None,
keep_table: str | None = None,
strict: bool | None = None,
autoincrement: bool | None = None,
) -> "Table":
"""
Apply an advanced alter table, including operations that are not supported by
@ -2684,6 +2716,8 @@ class Table(Queryable):
dropped
:param strict: Set to ``True`` to make the table strict or ``False`` to make it
non-strict. Defaults to ``None``, which preserves the existing strict mode.
:param autoincrement: Use AUTOINCREMENT for the single INTEGER PRIMARY KEY.
Set to False to disable it; None preserves the existing mode.
"""
if not self.exists():
raise ValueError("Cannot transform a table that doesn't exist yet")
@ -2700,6 +2734,7 @@ class Table(Queryable):
column_order=column_order,
keep_table=keep_table,
strict=strict,
autoincrement=autoincrement,
)
pragma_foreign_keys_was_on = bool(
self.db.execute("PRAGMA foreign_keys").fetchone()[0]
@ -2768,6 +2803,8 @@ class Table(Queryable):
self.db.execute("PRAGMA foreign_keys=1;")
if strict is not None:
self._defaults["strict"] = strict
if autoincrement is not None:
self._defaults["autoincrement"] = autoincrement
return self
def transform_sql(
@ -2786,6 +2823,7 @@ class Table(Queryable):
tmp_suffix: str | None = None,
keep_table: str | None = None,
strict: bool | None = None,
autoincrement: bool | None = None,
) -> list[str]:
"""
Return a list of SQL statements that should be executed in order to apply this transformation.
@ -2808,6 +2846,8 @@ class Table(Queryable):
dropped
:param strict: Set to ``True`` to make the table strict or ``False`` to make it
non-strict. Defaults to ``None``, which preserves the existing strict mode.
:param autoincrement: Use AUTOINCREMENT for the single INTEGER PRIMARY KEY.
Set to False to disable it; None preserves the existing mode.
"""
if strict is True and not self.db.supports_strict:
raise TransformError("SQLite does not support STRICT tables")
@ -3027,8 +3067,18 @@ class Table(Queryable):
):
create_table_autoincrement = renamed_autoincrement
if autoincrement is not None:
single_pk = pk[0] if isinstance(pk, (list, tuple)) and len(pk) == 1 else pk
create_table_autoincrement = (
resolve_casing(single_pk, dict(new_column_pairs))
if autoincrement and isinstance(single_pk, str)
else None
)
autoincrement_sequence = None
if create_table_autoincrement:
if existing_autoincrement and create_table_autoincrement == (
rename.get(existing_autoincrement) or existing_autoincrement
):
sequence_row = self.db.execute(
"SELECT seq FROM sqlite_sequence WHERE name = ?", [self.name]
).fetchone()
@ -3088,6 +3138,7 @@ class Table(Queryable):
_column_comments=create_table_column_comments,
_autoincrement=create_table_autoincrement,
_uniques=create_table_uniques,
autoincrement=autoincrement,
).strip()
)
@ -4601,6 +4652,27 @@ class Table(Queryable):
raise
return result
def _validate_autoincrement(
self, autoincrement: bool | None, pk: PrimaryKey | None
) -> None:
if autoincrement is None or not self.exists():
return
existing_autoincrement = parse_autoincrement(self.schema)
if bool(existing_autoincrement) != autoincrement:
raise ValueError(
"AUTOINCREMENT does not match the existing table; use transform() to change it"
)
if autoincrement and pk:
single_pk = pk[0] if isinstance(pk, (list, tuple)) and len(pk) == 1 else pk
if (
not isinstance(single_pk, str)
or resolve_casing(single_pk, self.columns_dict)
!= existing_autoincrement
):
raise ValueError(
"AUTOINCREMENT requires the table's INTEGER PRIMARY KEY"
)
def insert(
self,
record: dict[str, Any],
@ -4618,6 +4690,7 @@ class Table(Queryable):
conversions: dict[str, str] | Default | None = DEFAULT,
columns: dict[str, Any] | Default | None = DEFAULT,
strict: bool | Default | None = DEFAULT,
autoincrement: bool | Default | None = DEFAULT,
) -> "Table":
"""
Insert a single record into the table. The table will be created with a schema that matches
@ -4651,6 +4724,7 @@ class Table(Queryable):
:param columns: Dictionary over-riding the detected types used for the columns, for example
``{"age": int, "weight": float}``.
:param strict: Boolean, apply STRICT mode if creating the table.
:param autoincrement: Use AUTOINCREMENT for the single INTEGER PRIMARY KEY.
"""
return self.insert_all(
[record],
@ -4668,6 +4742,7 @@ class Table(Queryable):
conversions=conversions,
columns=columns,
strict=strict,
autoincrement=autoincrement,
)
def insert_all(
@ -4691,6 +4766,7 @@ class Table(Queryable):
upsert: bool = False,
analyze: bool = False,
strict: bool | Default | None = DEFAULT,
autoincrement: bool | Default | None = DEFAULT,
) -> "Table":
"""
Like ``.insert()`` but takes a list of records and ensures that the table
@ -4713,6 +4789,7 @@ class Table(Queryable):
conversions = self.value_or_default("conversions", conversions) or {}
columns = self.value_or_default("columns", columns)
strict = self.value_or_default("strict", strict)
autoincrement = self.value_or_default("autoincrement", autoincrement)
if hash_id_columns and hash_id is None:
hash_id = "id"
@ -4818,6 +4895,7 @@ class Table(Queryable):
if num_columns == 0
else max(1, min(batch_size, SQLITE_MAX_VARS // num_columns))
)
self._validate_autoincrement(autoincrement, pk)
self.last_rowid = None
self.last_pk = None
if truncate and self.exists():
@ -4855,6 +4933,7 @@ class Table(Queryable):
hash_id_columns=hash_id_columns,
extracts=extracts,
strict=strict,
autoincrement=autoincrement,
)
if list_mode:
# In list mode, columns are already known
@ -5043,6 +5122,7 @@ class Table(Queryable):
conversions: dict[str, str] | Default | None = DEFAULT,
columns: dict[str, Any] | Default | None = DEFAULT,
strict: bool | Default | None = DEFAULT,
autoincrement: bool | Default | None = DEFAULT,
) -> "Table":
"""
Like ``.insert()`` but performs an ``UPSERT``, where records are inserted if they do
@ -5064,6 +5144,7 @@ class Table(Queryable):
conversions=conversions,
columns=columns,
strict=strict,
autoincrement=autoincrement,
)
def upsert_all(
@ -5083,6 +5164,7 @@ class Table(Queryable):
columns: dict[str, Any] | Default | None = DEFAULT,
analyze: bool = False,
strict: bool | Default | None = DEFAULT,
autoincrement: bool | Default | None = DEFAULT,
) -> "Table":
"""
Like ``.upsert()`` but can be applied to a list of records.
@ -5104,6 +5186,7 @@ class Table(Queryable):
upsert=True,
analyze=analyze,
strict=strict,
autoincrement=autoincrement,
)
def add_missing_columns(self, records: Iterable[dict[str, Any]]) -> "Table":
@ -5127,6 +5210,7 @@ class Table(Queryable):
conversions: dict[str, str] | None = None,
columns: dict[str, Any] | None = None,
strict: bool | None = False,
autoincrement: bool | Default | None = DEFAULT,
):
"""
Create or populate a lookup table with the specified values.
@ -5150,6 +5234,7 @@ class Table(Queryable):
:param lookup_values: Dictionary specifying column names and values to use for the lookup
: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.
:param autoincrement: Use AUTOINCREMENT for the single INTEGER PRIMARY KEY.
"""
if not isinstance(lookup_values, dict):
raise ValueError("lookup_values must be a dictionary") # noqa: TRY004
@ -5157,6 +5242,8 @@ class Table(Queryable):
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")
autoincrement = self.value_or_default("autoincrement", autoincrement)
self._validate_autoincrement(autoincrement, pk)
combined_values = dict(lookup_values)
if extra_values is not None:
combined_values.update(extra_values)
@ -5190,6 +5277,7 @@ class Table(Queryable):
conversions=conversions,
columns=columns,
strict=strict,
autoincrement=autoincrement,
).last_pk
else:
pk = self.insert(
@ -5203,6 +5291,7 @@ class Table(Queryable):
conversions=conversions,
columns=columns,
strict=strict,
autoincrement=autoincrement,
).last_pk
self.create_index(lookup_values.keys(), unique=True)
return pk
@ -5498,6 +5587,26 @@ class View(Queryable):
raise
def _validate_autoincrement_columns(
pk: Any | None,
columns: dict[str, Any],
hash_id: str | None,
hash_id_columns: Iterable[str] | None,
extracts: dict[str, str] | list[str] | None,
) -> None:
single_pk = pk[0] if isinstance(pk, (list, tuple)) and len(pk) == 1 else pk
if not isinstance(single_pk, str):
raise ValueError("AUTOINCREMENT requires a single-column primary key")
if hash_id or hash_id_columns:
raise ValueError("AUTOINCREMENT requires an INTEGER PRIMARY KEY column")
effective_columns = dict(columns)
for column in resolve_extracts(extracts):
effective_columns[column] = int
single_pk = resolve_casing(single_pk, effective_columns)
if COLUMN_TYPE_MAPPING.get(effective_columns.get(single_pk, int)) != "INTEGER":
raise ValueError("AUTOINCREMENT requires an INTEGER PRIMARY KEY column")
def jsonify_if_needed(value: object) -> object:
if isinstance(value, decimal.Decimal):
return float(value)

91
tests/test_cli.py
View file

@ -2981,3 +2981,94 @@ def test_extract_view_clean_error(db_path):
assert result.exit_code == 1
assert result.exception is None or isinstance(result.exception, SystemExit)
assert result.output.startswith("Error:")
@pytest.mark.parametrize("command", ("create-table", "insert", "upsert"))
def test_cli_autoincrement_creation(tmp_path, command):
path = str(tmp_path / "autoincrement.db")
args = [command, path, "entries"]
if command == "create-table":
args += ["id", "integer", "value", "text"]
else:
args += ["-"]
records = [{"value": "one"}, {"value": "two"}]
if command == "upsert":
records = [{"id": i, **record} for i, record in enumerate(records, 1)]
result = CliRunner().invoke(
cli.cli, args + ["--pk", "id", "--autoincrement"], input=json.dumps(records)
)
assert result.exit_code == 0, result.output
db = Database(path)
table = db.table("entries")
if command == "create-table":
table.insert_all([{"value": "one"}, {"value": "two"}])
assert "INTEGER PRIMARY KEY AUTOINCREMENT" in table.schema
table.delete(2)
db.close()
result = CliRunner().invoke(
cli.cli, ["transform", path, "entries", "--rename", "value", "label"]
)
assert result.exit_code == 0, result.output
result = CliRunner().invoke(
cli.cli, ["insert", path, "entries", "-"], input='{"label":"three"}'
)
assert result.exit_code == 0, result.output
assert list(Database(path).table("entries").rows) == [
{"id": 1, "label": "one"},
{"id": 3, "label": "three"},
]
def test_cli_autoincrement_transform_preview_and_toggle(tmp_path):
path = str(tmp_path / "toggle.db")
db = Database(path)
db.table("entries").insert({"id": 5, "value": "keep"}, pk="id")
schema = db.table("entries").schema
db.close()
args = ["transform", path, "entries", "--autoincrement"]
result = CliRunner().invoke(cli.cli, args + ["--sql"])
assert result.exit_code == 0, result.output
assert "INTEGER PRIMARY KEY AUTOINCREMENT" in result.output
assert Database(path).table("entries").schema == schema
result = CliRunner().invoke(cli.cli, args)
assert result.exit_code == 0, result.output
result = CliRunner().invoke(
cli.cli, ["insert", path, "entries", "-"], input='{"value":"new"}'
)
assert result.exit_code == 0, result.output
assert list(Database(path).table("entries").rows)[-1]["id"] == 6
result = CliRunner().invoke(
cli.cli, ["transform", path, "entries", "--no-autoincrement"]
)
assert result.exit_code == 0, result.output
assert "AUTOINCREMENT" not in Database(path).table("entries").schema
@pytest.mark.parametrize("mode", ("--replace", "--transform"))
def test_cli_autoincrement_invalid_create_preserves_table(tmp_path, mode):
path = str(tmp_path / "keep.db")
db = Database(path)
table = db.table("entries").insert({"id": 1, "value": "keep"}, pk="id")
schema = table.schema
db.close()
result = CliRunner().invoke(
cli.cli,
[
"create-table",
path,
"entries",
"id",
"text",
"added",
"integer",
"--pk",
"id",
"--autoincrement",
mode,
],
)
assert result.exit_code != 0
assert "AUTOINCREMENT" in str(result.exception)
table = Database(path).table("entries")
assert table.schema == schema
assert list(table.rows) == [{"id": 1, "value": "keep"}]

219
tests/test_create.py
View file

@ -19,6 +19,7 @@ from sqlite_utils.db import (
NoTable,
NoView,
OperationalError,
PrimaryKeyRequired,
Table,
View,
)
@ -1739,3 +1740,221 @@ def test_chained_create_sets_pks(fresh_db):
{"id": int, "name": str, "color": str}
)
assert table.pks == ["id"]
@pytest.mark.parametrize(
"api",
(
"sql",
"database",
"table",
"configured",
"insert",
"insert_all",
"upsert",
"upsert_all",
"lookup",
),
)
def test_autoincrement_public_creation(fresh_db, api):
columns = {"id": int, "value": str}
records = [{"value": "one"}, {"value": "two"}]
if api in ("upsert", "upsert_all"):
records = [{"id": i, **record} for i, record in enumerate(records, 1)]
if api == "sql":
fresh_db.execute(
fresh_db.create_table_sql("entries", columns, pk="id", autoincrement=True)
)
table = fresh_db["entries"].insert_all(records)
elif api == "database":
table = fresh_db.create_table(
"entries", columns, pk="id", autoincrement=True
).insert_all(records)
elif api == "table":
table = (
fresh_db["entries"]
.create(columns, pk="id", autoincrement=True)
.insert_all(records)
)
elif api == "configured":
table = fresh_db.table("entries", pk="id", autoincrement=True).insert_all(
records
)
elif api == "lookup":
table = fresh_db["entries"]
assert table.lookup(records[0], autoincrement=True) == 1
assert table.lookup(records[1], autoincrement=True) == 2
elif api in ("insert", "upsert"):
table = fresh_db["entries"]
getattr(table, api)(records[0], pk="id", autoincrement=True)
getattr(table, api)(records[1], pk="id", autoincrement=True)
else:
table = getattr(fresh_db["entries"], api)(records, pk="id", autoincrement=True)
assert "INTEGER PRIMARY KEY AUTOINCREMENT" in table.schema
assert list(table.rows) == [{"id": 1, "value": "one"}, {"id": 2, "value": "two"}]
table.delete(2)
table.transform(rename={"value": "label"})
table.insert({"label": "three"})
assert list(table.rows) == [{"id": 1, "label": "one"}, {"id": 3, "label": "three"}]
@pytest.mark.parametrize("enabled", (None, False, True))
def test_autoincrement_default_and_explicit_modes(fresh_db, enabled):
kwargs = {} if enabled is None else {"autoincrement": enabled}
table = fresh_db["entries"].create({"id": int, "value": str}, pk="id", **kwargs)
table.insert_all([{"value": "one"}, {"value": "two"}])
table.delete(2)
table.insert({"value": "three"})
assert table.last_pk == (3 if enabled else 2)
assert ("AUTOINCREMENT" in table.schema) is bool(enabled)
@pytest.mark.parametrize("pk", ("ID", ["id"], ("id",)))
@pytest.mark.parametrize("strict", (False, True))
def test_autoincrement_single_pk_variants(fresh_db, pk, strict):
table = fresh_db["entries"].create(
{"id": "integer", "value": str}, pk=pk, strict=strict, autoincrement=True
)
table.insert({"value": "one"})
assert list(table.rows) == [{"id": 1, "value": "one"}]
assert "INTEGER PRIMARY KEY AUTOINCREMENT" in table.schema
def test_autoincrement_inferred_quoted_pk(fresh_db):
table = fresh_db["entries"].insert(
{"value": "one"}, pk='key"name', autoincrement=True
)
assert list(table.rows) == [{'key"name': 1, "value": "one"}]
table.delete(1)
table.insert({"value": "two"})
assert list(table.rows) == [{'key"name': 2, "value": "two"}]
@pytest.mark.parametrize("api", ("database", "table"))
@pytest.mark.parametrize("mode", ("replace", "transform"))
@pytest.mark.parametrize(
"columns, options",
[
({"id": int, "value": str, "added": str}, {"pk": None}),
({"id": str, "value": str, "added": str}, {"pk": "id"}),
({"id": int, "value": str, "added": str}, {"pk": ("id", "value")}),
({"id": int, "value": str, "added": str}, {"hash_id": "hash"}),
],
)
def test_autoincrement_invalid_create_preserves_table(
fresh_db, api, mode, columns, options
):
table = fresh_db["entries"].insert({"id": 1, "value": "keep"}, pk="id")
schema = table.schema
with pytest.raises(ValueError, match="AUTOINCREMENT"):
if api == "database":
fresh_db.create_table(
"entries", columns, autoincrement=True, **{mode: True}, **options
)
else:
fresh_db["entries"].create(
columns, autoincrement=True, **{mode: True}, **options
)
assert table.schema == schema
assert list(table.rows) == [{"id": 1, "value": "keep"}]
@pytest.mark.parametrize("records", ([], [[]], [["value"]]))
def test_autoincrement_empty_input_no_table(fresh_db, records):
fresh_db["entries"].insert_all(records, pk="id", autoincrement=True)
assert not fresh_db["entries"].exists()
@pytest.mark.parametrize("operation", ("insert", "lookup"))
def test_autoincrement_existing_mismatch_preserves_data(fresh_db, operation):
table = fresh_db["entries"].insert({"id": 1, "value": "keep"}, pk="id")
schema = table.schema
with pytest.raises(ValueError, match="AUTOINCREMENT does not match"):
if operation == "insert":
table.insert_all(
[{"value": "new", "added": 1}],
pk="id",
autoincrement=True,
truncate=True,
alter=True,
)
else:
table.lookup(
{"value": "new"}, extra_values={"added": 1}, autoincrement=True
)
assert table.schema == schema
assert list(table.rows) == [{"id": 1, "value": "keep"}]
def test_autoincrement_create_ignore_and_if_not_exists(fresh_db):
table = fresh_db["entries"].create(
{"id": int, "value": str}, pk="id", autoincrement=True
)
table.insert({"value": "keep"})
schema = table.schema
fresh_db.create_table(
"entries", {"id": int}, pk="id", autoincrement=True, if_not_exists=True
)
fresh_db["entries"].create({"id": str}, autoincrement=True, ignore=True)
assert table.schema == schema
assert list(table.rows) == [{"id": 1, "value": "keep"}]
def test_autoincrement_does_not_change_upsert_pk_requirement(fresh_db):
with pytest.raises(
PrimaryKeyRequired, match="requires a value for the primary key"
):
fresh_db["entries"].upsert({"value": "one"}, pk="id", autoincrement=True)
assert fresh_db["entries"].count == 0
@pytest.mark.parametrize("api", ("sql", "database", "table"))
@pytest.mark.parametrize(
"options",
({"pk": None}, {"pk": ("name", "other")}, {"pk": "name"}, {"hash_id": "hash"}),
)
def test_autoincrement_invalid_extracts_create_nothing(fresh_db, api, options):
columns = {"name": str, "other": str}
with pytest.raises(ValueError, match="AUTOINCREMENT"):
if api == "sql":
fresh_db.create_table_sql(
"entries",
columns,
extracts={"other": "lookup"},
autoincrement=True,
**options,
)
elif api == "database":
fresh_db.create_table(
"entries",
columns,
extracts={"other": "lookup"},
autoincrement=True,
**options,
)
else:
fresh_db.table("entries").create(
columns, extracts={"other": "lookup"}, autoincrement=True, **options
)
assert fresh_db.table_names() == []
def test_autoincrement_extracted_integer_primary_key(fresh_db):
sql = fresh_db.create_table_sql(
"entries",
{"name": str},
pk="name",
extracts={"name": "lookup"},
autoincrement=True,
)
fresh_db.execute(sql)
assert "INTEGER PRIMARY KEY AUTOINCREMENT" in fresh_db.table("entries").schema
assert set(fresh_db.table_names()) == {"entries", "lookup", "sqlite_sequence"}
def test_autoincrement_ignored_create_preserves_defaults(fresh_db):
table = fresh_db.table("entries", pk="id").create({"id": int, "value": str})
table.create({"id": str}, ignore=True, autoincrement=True)
table.insert({"value": "one"})
assert "AUTOINCREMENT" not in table.schema
assert list(table.rows) == [{"id": 1, "value": "one"}]

44
tests/test_transform.py
View file

@ -1472,3 +1472,47 @@ def test_transform_refuses_to_drop_column_used_by_remaining_check(fresh_db):
):
ranges.transform(drop={"maximum"})
assert ranges.schema == schema_before
@pytest.mark.parametrize("api", ("transform", "create"))
def test_autoincrement_toggle_refreshes_defaults(fresh_db, api):
table = fresh_db.table("entries", pk="id", autoincrement=False).create(
{"id": int, "value": str}
)
table.insert({"value": "one"})
if api == "transform":
preview = table.transform_sql(autoincrement=True)
assert "AUTOINCREMENT" in preview[0]
assert "AUTOINCREMENT" not in table.schema
table.transform(autoincrement=True)
else:
table.create(
{"id": int, "value": str}, pk="id", transform=True, autoincrement=True
)
table.insert({"value": "two"})
table.delete(2)
table.transform(rename={"value": "label"}, autoincrement=True)
table.insert({"label": "three"})
assert table.last_pk == 3
if api == "transform":
table.transform(autoincrement=False)
else:
table.create(
{"id": int, "label": str}, pk="id", transform=True, autoincrement=False
)
assert "AUTOINCREMENT" not in table.schema
table.delete(3)
table.insert({"label": "four"})
assert table.last_pk == 2
@pytest.mark.parametrize(
"options", ({"pk": None}, {"pk": ("id", "value")}, {"types": {"id": str}})
)
def test_autoincrement_invalid_transform_preserves_table(fresh_db, options):
table = fresh_db["entries"].insert({"id": 1, "value": "keep"}, pk="id")
schema = table.schema
with pytest.raises(ValueError, match="AUTOINCREMENT"):
table.transform(autoincrement=True, **options)
assert table.schema == schema
assert list(table.rows) == [{"id": 1, "value": "keep"}]