Compare commits

...

16 commits

Author SHA1 Message Date
Simon Willison
85b1be10c8 Same fix for test-coverage.yml
Refs https://github.com/simonw/sqlite-utils/pull/852/changes/ef50b31a21104351445c4856ff3a80b81b52e847
2026-09-01 21:44:52 -07:00
Simon Willison
f7e3174401 Fix for SpatialLite installation failure
Fable 5.1 explains:

> apt on the runner image has stale package lists. It tried to download libminizip1t64_1.3.dfsg-3.1ubuntu2.1 from security.ubuntu.com and got a 404, because Ubuntu has since published a newer build of that package and pulled the old .deb from the mirror.
2026-09-01 21:44:52 -07:00
Simon Willison
b97295271c Test against 3.15 RCs
See https://simonwillison.net/2026/Sep/1/python-315-rc-2/
2026-09-01 21:44:52 -07:00
Simon Willison
56dd09702f Run no-default-groups smoke test from Justfile
Refs #842

I had to add --isolated because otherwise this test would pass if a .venv
folder already existed with the dev dependencies installed in it.
2026-08-13 17:01:47 -07:00
Simon Willison
28dc6278cc Release 4.2.1
Refs #842, #843
2026-08-13 16:52:30 -07:00
Simon Willison
f6d73112c8
Fix for sqlite-utils 4.2 crashing bug (#843)
- Remove from typing_extensions import Self
- Smoke test: uv run --no-default-groups sqlite-utils --help

Closes #842
2026-08-13 16:52:03 -07:00
Simon Willison
1d98613f28 Release 4.2
Refs #488, #602, #762, #790, #805, #808, #811, #816, #821, #822, #824, #825, #828, #829, #831, #833, #834, #836, #837
2026-08-13 13:09:42 -07:00
ikatyal2110
e4935e0644
transform: coerce empty strings to NULL when converting TEXT columns to numeric types (#805)
* transform: coerce empty strings to NULL when converting TEXT columns to numeric types

When a TEXT column is transformed to INTEGER, FLOAT, or REAL and a row
contains an empty string, the empty string is now converted to NULL during
the INSERT...SELECT copy, matching the expected behavior described in #488.

Fixes #488
2026-08-13 12:56:51 -07:00
Simon Willison
75ba588462 Preserve composite UNIQUE constraints in transforms 2026-08-12 18:46:18 -07:00
Simon Willison
2b52b5ed6f Preserve AUTOINCREMENT through transforms 2026-08-12 18:39:00 -07:00
Simon Willison
fcfccea813 Support ANY column types for strict tables
Closes #790, #820
2026-08-12 16:44:01 -07:00
Simon Willison
57192ef4e3 table.transform(rename=...) now preserves indexes, closes #822 2026-08-12 14:42:24 -07:00
Simon Willison
e4784ec120 Changelog updates
Refs #808, #811, #816, #821, #824, #825, #828, #829, #833, #836, #837
2026-08-12 14:38:30 -07:00
Simon Willison
88b48fa167 Fixed introspection of default values TRUE / FALSE / NULL
Closes #836
2026-08-12 14:19:38 -07:00
nyxst4ck
e6be6267a4
Use quote_identifier() in indexes/xindexes PRAGMA statements (#825)
Closes #824
2026-08-12 14:15:17 -07:00
Simon Willison
c5063f67b1 Use quoted SQL identifiers in convert --dry-run, closes #829 2026-08-12 14:14:27 -07:00
21 changed files with 1199 additions and 90 deletions

View file

@ -20,7 +20,7 @@ jobs:
cache: pip cache: pip
cache-dependency-path: pyproject.toml cache-dependency-path: pyproject.toml
- name: Install SpatiaLite - name: Install SpatiaLite
run: sudo apt-get install libsqlite3-mod-spatialite run: sudo apt-get update && sudo apt-get install -y libsqlite3-mod-spatialite
- name: Install Python dependencies - name: Install Python dependencies
run: | run: |
python -m pip install --upgrade pip python -m pip install --upgrade pip

View file

@ -10,18 +10,19 @@ jobs:
runs-on: ${{ matrix.os }} runs-on: ${{ matrix.os }}
strategy: strategy:
matrix: matrix:
python-version: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.15-dev"] python-version: ["3.10", "3.11", "3.12", "3.13", "3.14", "3.15"]
numpy: [0, 1] numpy: [0, 1]
os: [ubuntu-latest, macos-latest, windows-latest, macos-14] os: [ubuntu-latest, macos-latest, windows-latest, macos-14]
steps: steps:
- uses: actions/checkout@v7 - uses: actions/checkout@v7
- name: Set up Python ${{ matrix.python-version }} - name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v6 uses: actions/setup-python@v7
with: with:
python-version: ${{ matrix.python-version }} python-version: ${{ matrix.python-version }}
allow-prereleases: true allow-prereleases: true
cache: pip cache: pip
cache-dependency-path: pyproject.toml cache-dependency-path: pyproject.toml
check-latest: true
- name: Install dependencies - name: Install dependencies
run: | run: |
pip install . --group dev pip install . --group dev
@ -30,7 +31,7 @@ jobs:
run: pip install numpy run: pip install numpy
- name: Install SpatiaLite - name: Install SpatiaLite
if: matrix.os == 'ubuntu-latest' if: matrix.os == 'ubuntu-latest'
run: sudo apt-get install libsqlite3-mod-spatialite run: sudo apt-get update && sudo apt-get install -y libsqlite3-mod-spatialite
- name: Build extension for --load-extension test - name: Build extension for --load-extension test
if: matrix.os == 'ubuntu-latest' if: matrix.os == 'ubuntu-latest'
run: |- run: |-
@ -53,6 +54,11 @@ jobs:
run: | run: |
pip install uv pip install uv
uv run ty check sqlite_utils uv run ty check sqlite_utils
- name: Check no accidental dev= dependencies needed
if: matrix.os == 'ubuntu-latest'
run: |
pip install uv
uv run --no-default-groups sqlite-utils --help
- name: Check formatting - name: Check formatting
run: black . --check run: black . --check
- name: Check if cog needs to be run - name: Check if cog needs to be run

View file

@ -2,9 +2,12 @@
@default: test lint @default: test lint
# Run pytest with supplied options # Run pytest with supplied options
@test *options: @test *options: test-no-dev-dependencies
uv run pytest {{options}} uv run pytest {{options}}
@test-no-dev-dependencies:
uv run --isolated --no-default-groups sqlite-utils --help > /dev/null
@run *options: @run *options:
uv run -- {{options}} uv run -- {{options}}

View file

@ -4,16 +4,40 @@
Changelog Changelog
=========== ===========
.. _unreleased: .. _v4_2_1:
Unreleased 4.2.1 (2026-08-13)
---------- ------------------
- Fix for ``No module named 'typing_extensions'`` crashing bug accidentally shipped in version 4.2. (:issue:`842`)
.. _v4_2:
4.2 (2026-08-13)
----------------
- New ``table.checks``, ``table.column_checks`` and ``table.table_checks`` introspection properties expose column-level and table-level ``CHECK`` constraints. (:issue:`834`) - New ``table.checks``, ``table.column_checks`` and ``table.table_checks`` introspection properties expose column-level and table-level ``CHECK`` constraints. (:issue:`834`)
- New ``sqlite_utils.ANY`` marker type for creating and introspecting SQLite ``ANY`` columns. The Python API and CLI can create, add and transform these columns, and ``table.transform()`` and ``table.extract()`` now preserve ``ANY`` columns and their values in ``STRICT`` tables. (:issue:`790`)
- ``table.default_values`` now unescapes doubled single quotes in string defaults, so a default such as ``'O''Brien'`` is returned as ``"O'Brien"``. Thanks, `ikatyal2110 <https://github.com/ikatyal2110>`__. (`#811 <https://github.com/simonw/sqlite-utils/pull/811>`__)
- ``table.default_values`` now decodes unquoted ``TRUE``, ``FALSE`` and ``NULL`` default literals as ``True``, ``False`` and ``None`` respectively. (:issue:`836`)
- ``table.enable_fts(..., tokenize=...)`` and ``sqlite-utils enable-fts --tokenize`` now safely quote the tokenizer argument, preventing a crafted value from injecting additional SQL. Thanks, `Bunlong Heng <https://github.com/bunlongheng>`__. (`#828 <https://github.com/simonw/sqlite-utils/pull/828>`__)
- ``rows_where()``, ``pks_and_rows_where()``, ``search()`` and ``search_sql()`` now support ``offset=`` without requiring ``limit=``. The ``sqlite-utils rows --offset`` option now works without ``--limit`` too. Thanks, `ethanhawkes-gif <https://github.com/ethanhawkes-gif>`__. (:issue:`816`, `#821 <https://github.com/simonw/sqlite-utils/pull/821>`__)
- Empty or whitespace-only input passed to ``rows_from_file()`` is now handled as an empty CSV file instead of raising ``csv.Error``. Thanks, `Rami Abdelrazzaq <https://github.com/RamiNoodle733>`__. (:issue:`808`, `#837 <https://github.com/simonw/sqlite-utils/pull/837>`__)
- ``sqlite-utils convert --dry-run`` now works for table and column names containing closing square brackets. (:issue:`829`)
- ``table.indexes`` and ``table.xindexes`` now work for table, index and column names containing double quotes. This also fixes ``table.transform()`` for tables with those identifiers. Thanks, `nyxst4ck <https://github.com/nyxst4ck>`__. (:issue:`824`, `#825 <https://github.com/simonw/sqlite-utils/pull/825>`__)
- Improved type annotations throughout the package and added Pyright regression checks to CI. (:issue:`833`)
- Changing a ``TEXT`` column to ``INTEGER``, ``FLOAT`` or ``REAL`` using ``table.transform()`` or ``sqlite-utils transform`` now converts exact empty strings to ``NULL``. Previously they remained empty strings in the numeric column. Thanks, `ikatyal2110 <https://github.com/ikatyal2110>`__. (:issue:`488`, `#805 <https://github.com/simonw/sqlite-utils/pull/805>`__)
``table.transform()`` can handle many more edge-cases:
- ``table.transform()`` now preserves column-level and composite ``UNIQUE`` constraints, including constraint names, collations, sort order and ``ON CONFLICT`` behavior. Renaming columns updates those constraints, while dropping any constituent column removes the entire constraint. (:issue:`762`)
- ``table.transform()`` now preserves ``AUTOINCREMENT`` primary keys and their sequence high-water marks. Previously a transform removed ``AUTOINCREMENT`` and could reuse deleted row IDs. (:issue:`602`)
- ``table.transform()`` now preserves ``CHECK`` constraints, including comments within their expressions. Renaming a column rewrites identifier references in checks without changing string literals or function names. Dropping a column drops a check owned by that column, and raises ``TransformError`` if a remaining check depends on it. (:issue:`762`) - ``table.transform()`` now preserves ``CHECK`` constraints, including comments within their expressions. Renaming a column rewrites identifier references in checks without changing string literals or function names. Dropping a column drops a check owned by that column, and raises ``TransformError`` if a remaining check depends on it. (:issue:`762`)
- ``table.transform()`` now preserves comments immediately before or after column definitions. These comments move with the column if it is renamed or reordered, and are removed if the column is dropped. (:issue:`762`) - ``table.transform()`` now preserves comments immediately before or after column definitions. These comments move with the column if it is renamed or reordered, and are removed if the column is dropped. (:issue:`762`)
- ``table.transform(rename=...)`` now preserves explicit indexes on renamed columns by dropping and recreating those indexes against the new column names. Previously this raised a ``TransformError``. (:issue:`822`)
- ``table.transform()`` now works for tables that are referenced by views. Previously the ``ALTER TABLE ... RENAME TO`` step raised ``no such table`` if a view referenced the table being transformed. View definitions are left unchanged - see :ref:`python_api_transform_views`. This also fixes a bug where ``transform(keep_table=...)`` silently rewrote dependent views to point at the frozen backup table instead of the live one. (:issue:`831`) - ``table.transform()`` now works for tables that are referenced by views. Previously the ``ALTER TABLE ... RENAME TO`` step raised ``no such table`` if a view referenced the table being transformed. View definitions are left unchanged - see :ref:`python_api_transform_views`. This also fixes a bug where ``transform(keep_table=...)`` silently rewrote dependent views to point at the frozen backup table instead of the live one. (:issue:`831`)
.. _v3_39_1: .. _v3_39_1:
3.39.1 (2026-07-25) 3.39.1 (2026-07-25)

View file

@ -494,7 +494,7 @@ See :ref:`cli_transform_table`.
Options: Options:
--type <TEXT CHOICE>... Change column type to INTEGER, TEXT, FLOAT, --type <TEXT CHOICE>... Change column type to INTEGER, TEXT, FLOAT,
REAL or BLOB REAL, BLOB or ANY
--drop TEXT Drop this column --drop TEXT Drop this column
--rename <TEXT TEXT>... Rename this column to X --rename <TEXT TEXT>... Rename this column to X
-o, --column-order TEXT Reorder columns -o, --column-order TEXT Reorder columns
@ -963,7 +963,7 @@ See :ref:`cli_create_table`.
height real \ height real \
photo blob --pk id photo blob --pk id
Valid column types are text, integer, real, float and blob. Valid column types are text, integer, real, float, blob and any.
Options: Options:
--pk TEXT Column to use as primary key --pk TEXT Column to use as primary key
@ -1257,7 +1257,7 @@ See :ref:`cli_add_column`.
:: ::
Usage: sqlite-utils add-column [OPTIONS] PATH TABLE COL_NAME Usage: sqlite-utils add-column [OPTIONS] PATH TABLE COL_NAME
[integer|int|float|real|text|str|blob|bytes] [integer|int|float|real|text|str|blob|bytes|any]
Add a column to the specified table Add a column to the specified table

View file

@ -1390,7 +1390,14 @@ Use ``--type column-name type`` to override the type automatically chosen when t
This is useful for values such as ZIP codes, which may look like integers but should be stored as ``TEXT`` to preserve leading zeros. This is useful for values such as ZIP codes, which may look like integers but should be stored as ``TEXT`` to preserve leading zeros.
The column type should be one of ``TEXT``, ``INTEGER``, ``FLOAT``, ``REAL`` or ``BLOB``. Column types are matched case-insensitively. The column type should be one of ``TEXT``, ``INTEGER``, ``FLOAT``, ``REAL``, ``BLOB`` or ``ANY``. Column types are matched case-insensitively.
``ANY`` is especially useful with ``--strict``. An ``ANY`` column in a strict table preserves values without coercion, so text such as ``000123`` remains text instead of being converted to an integer:
.. code-block:: bash
sqlite-utils insert events.db events events.csv --csv --strict \
--type payload any
As with detected column types, ``--type`` only affects tables created by the command. If the table already exists, its existing column types are left unchanged. As with detected column types, ``--type`` only affects tables created by the command. If the table already exists, its existing column types are left unchanged.
@ -2141,6 +2148,12 @@ You can create a table in `SQLite STRICT mode <https://www.sqlite.org/stricttabl
sqlite-utils create-table mydb.db mytable id integer name text --strict sqlite-utils create-table mydb.db mytable id integer name text --strict
Use the ``any`` type for a strict column that should accept integers, floating point values, text, binary data or null without coercion:
.. code-block:: bash
sqlite-utils create-table events.db events id integer payload any --strict
.. code-block:: bash .. code-block:: bash
sqlite-utils tables mydb.db --schema -t sqlite-utils tables mydb.db --schema -t
@ -2223,7 +2236,7 @@ The ``transform`` command allows you to apply complex transformations to a table
Every option for this table (with the exception of ``--pk-none``) can be specified multiple times. The options are as follows: Every option for this table (with the exception of ``--pk-none``) can be specified multiple times. The options are as follows:
``--type column-name new-type`` ``--type column-name new-type``
Change the type of the specified column. Valid types are ``integer``, ``text``, ``float``, ``blob``. Change the type of the specified column. Valid types are ``integer``, ``text``, ``float``, ``real``, ``blob`` and ``any``. Changing a ``TEXT`` column to ``INTEGER``, ``FLOAT`` or ``REAL`` converts exact empty-string values to ``NULL``.
``--drop column-name`` ``--drop column-name``
Drop the specified column. Drop the specified column.

View file

@ -828,6 +828,19 @@ You can pass ``strict=True`` to create a table in ``STRICT`` mode:
"name": str, "name": str,
}, strict=True) }, strict=True)
SQLite ``STRICT`` tables can use the ``ANY`` column type for values that should retain their exact SQLite storage class without coercion. Use the ``sqlite_utils.ANY`` marker type:
.. code-block:: python
import sqlite_utils
db.table("events").create({
"id": int,
"payload": sqlite_utils.ANY,
}, pk="id", strict=True)
An ``ANY`` column can store integers, floating point values, text, binary data or ``None``. In a ``STRICT`` table a text value such as ``"000123"`` remains text with its leading zeroes intact. SQLite also accepts ``ANY`` columns in ordinary non-``STRICT`` tables, but those columns apply numeric affinity and would store that same value as the integer ``123``.
.. note:: .. note::
In the CLI: :ref:`sqlite-utils create-table <cli_create_table>` In the CLI: :ref:`sqlite-utils create-table <cli_create_table>`
@ -1569,7 +1582,7 @@ You can specify the ``col_type`` argument either using a SQLite type as a string
The ``col_type`` is optional - if you omit it the type of ``TEXT`` will be used. The ``col_type`` is optional - if you omit it the type of ``TEXT`` will be used.
SQLite types you can specify are ``"TEXT"``, ``"INTEGER"``, ``"FLOAT"``, ``"REAL"`` or ``"BLOB"``. SQLite types you can specify are ``"TEXT"``, ``"INTEGER"``, ``"FLOAT"``, ``"REAL"``, ``"BLOB"`` or ``"ANY"``. You can use the ``sqlite_utils.ANY`` marker instead of the ``"ANY"`` string.
If you pass a Python type, it will be mapped to SQLite types as shown here:: If you pass a Python type, it will be mapped to SQLite types as shown here::
@ -1582,6 +1595,7 @@ If you pass a Python type, it will be mapped to SQLite types as shown here::
datetime.date: "TEXT" datetime.date: "TEXT"
datetime.time: "TEXT" datetime.time: "TEXT"
datetime.timedelta: "TEXT" datetime.timedelta: "TEXT"
sqlite_utils.ANY: "ANY"
# If numpy is installed # If numpy is installed
np.int8: "INTEGER" np.int8: "INTEGER"
@ -1812,6 +1826,8 @@ To alter the type of a column, use the ``types=`` argument:
# Convert the 'age' column to an integer, and 'weight' to a float # Convert the 'age' column to an integer, and 'weight' to a float
table.transform(types={"age": int, "weight": float}) table.transform(types={"age": int, "weight": float})
When a ``TEXT`` column is changed to ``INTEGER``, ``FLOAT`` or ``REAL``, exact empty-string values are stored as ``NULL``. Other values, including whitespace-only strings, are copied normally.
See :ref:`python_api_add_column` for a list of available types. See :ref:`python_api_add_column` for a list of available types.
.. _python_api_transform_strict: .. _python_api_transform_strict:
@ -1831,6 +1847,8 @@ Pass ``strict=False`` to convert a strict table back to a regular non-strict tab
table.transform(strict=False) table.transform(strict=False)
If the table has ``ANY`` columns, converting it to non-strict mode can coerce text values that look numeric. For example, SQLite converts ``"000123"`` to the integer ``123`` when copying it into an ordinary ``ANY`` column. This is SQLite's documented distinction between `STRICT and ordinary ANY columns <https://www.sqlite.org/stricttables.html#the_any_datatype>`__.
The default is ``strict=None``, which preserves the table's existing strict mode. The default is ``strict=None``, which preserves the table's existing strict mode.
Passing ``strict=True`` raises ``sqlite_utils.db.TransformError`` if the available SQLite version does not support strict tables. Passing ``strict=True`` raises ``sqlite_utils.db.TransformError`` if the available SQLite version does not support strict tables.
@ -2458,6 +2476,11 @@ The ``.columns_dict`` property returns a dictionary version of the columns with
>>> db.table("PlantType").columns_dict >>> db.table("PlantType").columns_dict
{'id': <class 'int'>, 'value': <class 'str'>} {'id': <class 'int'>, 'value': <class 'str'>}
SQLite ``ANY`` columns are represented by the ``sqlite_utils.ANY`` marker type::
>>> db.table("events").columns_dict
{'id': <class 'int'>, 'payload': <class 'sqlite_utils.utils.ANY'>}
.. _python_api_introspection_default_values: .. _python_api_introspection_default_values:
.default_values .default_values

View file

@ -1,6 +1,6 @@
[project] [project]
name = "sqlite-utils" name = "sqlite-utils"
version = "4.1.1" version = "4.2.1"
description = "CLI tool and Python library for manipulating SQLite databases" description = "CLI tool and Python library for manipulating SQLite databases"
readme = { file = "README.md", content-type = "text/markdown" } readme = { file = "README.md", content-type = "text/markdown" }
authors = [ authors = [

View file

@ -1,6 +1,13 @@
from .db import Database from .db import Database
from .hookspecs import hookimpl, hookspec from .hookspecs import hookimpl, hookspec
from .migrations import Migrations from .migrations import Migrations
from .utils import suggest_column_types from .utils import ANY, suggest_column_types
__all__ = ["Database", "Migrations", "hookimpl", "hookspec", "suggest_column_types"] __all__ = [
"ANY",
"Database",
"Migrations",
"hookimpl",
"hookspec",
"suggest_column_types",
]

View file

@ -76,7 +76,7 @@ def _close_databases(ctx):
pass pass
VALID_COLUMN_TYPES = ("INTEGER", "TEXT", "FLOAT", "REAL", "BLOB") VALID_COLUMN_TYPES = ("INTEGER", "TEXT", "FLOAT", "REAL", "BLOB", "ANY")
UNICODE_ERROR = """ UNICODE_ERROR = """
{} {}
@ -489,7 +489,17 @@ def dump(path, load_extension):
@click.argument( @click.argument(
"col_type", "col_type",
type=click.Choice( type=click.Choice(
["integer", "int", "float", "real", "text", "str", "blob", "bytes"], [
"integer",
"int",
"float",
"real",
"text",
"str",
"blob",
"bytes",
"any",
],
case_sensitive=False, case_sensitive=False,
), ),
required=False, required=False,
@ -1758,7 +1768,7 @@ def create_table(
height real \\ height real \\
photo blob --pk id photo blob --pk id
Valid column types are text, integer, real, float and blob. Valid column types are text, integer, real, float, blob and any.
""" """
db = sqlite_utils.Database(path) db = sqlite_utils.Database(path)
_register_db_for_cleanup(db) _register_db_for_cleanup(db)
@ -2668,12 +2678,10 @@ def schema(
"--type", "--type",
type=( type=(
str, str,
click.Choice( click.Choice(list(VALID_COLUMN_TYPES), case_sensitive=False),
["INTEGER", "TEXT", "FLOAT", "REAL", "BLOB"], case_sensitive=False
),
), ),
multiple=True, multiple=True,
help="Change column type to INTEGER, TEXT, FLOAT, REAL or BLOB", help="Change column type to INTEGER, TEXT, FLOAT, REAL, BLOB or ANY",
) )
@click.option("--drop", type=str, multiple=True, help="Drop this column") @click.option("--drop", type=str, multiple=True, help="Drop this column")
@click.option( @click.option(
@ -3283,12 +3291,12 @@ def convert(
db.conn.create_function("preview_transform", 1, preview) db.conn.create_function("preview_transform", 1, preview)
sql = """ sql = """
select select
[{column}] as value, {column} as value,
preview_transform([{column}]) as preview preview_transform({column}) as preview
from [{table}]{where} limit 10 from {table}{where} limit 10
""".format( """.format(
column=columns[0], column=quote_identifier(columns[0]),
table=table, table=quote_identifier(table),
where=f" where {where}" if where is not None else "", where=f" where {where}" if where is not None else "",
) )
for row in db.conn.execute(sql, where_args).fetchall(): for row in db.conn.execute(sql, where_args).fetchall():

View file

@ -1,4 +1,4 @@
"""Helpers for parsing CHECK constraints from SQLite CREATE TABLE SQL. """Helpers for parsing constraints from SQLite CREATE TABLE SQL.
SQLite does not expose CHECK constraints through a pragma, so preserving them SQLite does not expose CHECK constraints through a pragma, so preserving them
across a table rebuild requires reading ``sqlite_schema.sql``. This module is across a table rebuild requires reading ``sqlite_schema.sql``. This module is
@ -32,6 +32,24 @@ class ColumnComments:
after: str = "" after: str = ""
@dataclass(frozen=True)
class UniqueColumn:
name: str
collation: str = ""
order: str = ""
@dataclass
class Unique:
columns: tuple[UniqueColumn, ...]
name: str = ""
column: str = ""
conflict: str = ""
sql: str = field(default="", compare=False, repr=False)
start: int = field(default=-1, compare=False, repr=False)
end: int = field(default=-1, compare=False, repr=False)
class ParseError(ValueError): class ParseError(ValueError):
pass pass
@ -564,6 +582,209 @@ def parse_checks(create_sql: str) -> list[Check]:
return checks return checks
def parse_autoincrement(create_sql: str) -> str | None:
"""Return the AUTOINCREMENT column from a valid CREATE TABLE statement."""
body_info = _table_body(create_sql)
if body_info is None:
return None
body, _ = body_info
for item, _, _ in _split_spans(body, _lex(body)):
item_tokens = _meaningful(_lex(item))
if not item_tokens:
continue
head = item_tokens[0]
if (
head.kind == "word" and head.text.upper() in _TABLE_CONSTRAINT_KEYWORDS
) or head.is_keyword("CONSTRAINT"):
continue
column = _unquote(head.text)
index = 1
while index < len(item_tokens):
token = item_tokens[index]
if token.text == "(":
index = _matching_paren(item_tokens, index) + 1
continue
if token.is_keyword("AUTOINCREMENT"):
return column
index += 1
return None
_CONFLICT_ACTIONS = frozenset(("ROLLBACK", "ABORT", "FAIL", "IGNORE", "REPLACE"))
def _conflict_after(tokens: list[_Token], index: int) -> tuple[str, int]:
if index >= len(tokens) or not tokens[index].is_keyword("ON"):
return "", index
if index + 2 >= len(tokens) or not tokens[index + 1].is_keyword("CONFLICT"):
raise ParseError("ON after UNIQUE must be followed by CONFLICT and an action")
action = tokens[index + 2].text.upper()
if tokens[index + 2].kind != "word" or action not in _CONFLICT_ACTIONS:
raise ParseError("Invalid UNIQUE ON CONFLICT action")
return action, index + 3
def _unique_columns(
item: str, tokens: list[_Token], open_index: int
) -> tuple[tuple[UniqueColumn, ...], int]:
close = _matching_paren(tokens, open_index)
inner = item[tokens[open_index].end : tokens[close].start]
columns: list[UniqueColumn] = []
for raw_column in _split_ranges(inner, _lex(inner)):
column_tokens = _meaningful(_lex(raw_column))
if not column_tokens or column_tokens[0].kind not in (
"word",
"identifier",
"string",
):
raise ParseError("UNIQUE constraint has an invalid column")
name = _unquote(column_tokens[0].text)
collation = ""
order = ""
index = 1
if index < len(column_tokens) and column_tokens[index].is_keyword("COLLATE"):
if index + 1 >= len(column_tokens):
raise ParseError("COLLATE in UNIQUE constraint is missing its name")
collation = _unquote(column_tokens[index + 1].text)
index += 2
if index < len(column_tokens) and (
column_tokens[index].is_keyword("ASC")
or column_tokens[index].is_keyword("DESC")
):
order = column_tokens[index].text.upper()
index += 1
if index != len(column_tokens):
raise ParseError("UNIQUE constraint has an invalid indexed column")
columns.append(UniqueColumn(name, collation=collation, order=order))
if not columns:
raise ParseError("UNIQUE constraint must include at least one column")
return tuple(columns), close + 1
def _column_uniques(
item: str, tokens: list[_Token], column: str, base_offset: int
) -> list[Unique]:
uniques: list[Unique] = []
collation = ""
collation_index = 1
while collation_index < len(tokens):
token = tokens[collation_index]
if token.text == "(":
collation_index = _matching_paren(tokens, collation_index) + 1
continue
if token.is_keyword("COLLATE"):
if collation_index + 1 >= len(tokens):
raise ParseError("COLLATE is missing its name")
collation = _unquote(tokens[collation_index + 1].text)
collation_index += 2
continue
collation_index += 1
pending_name = ""
pending_start: int | None = None
index = 1
while index < len(tokens):
token = tokens[index]
if token.text == "(":
index = _matching_paren(tokens, index) + 1
continue
if token.is_keyword("CONSTRAINT"):
if index + 1 >= len(tokens):
raise ParseError("CONSTRAINT is missing its name")
pending_name = _unquote(tokens[index + 1].text)
pending_start = index
index += 2
continue
if token.is_keyword("UNIQUE"):
source_start = tokens[
pending_start if pending_start is not None else index
].start
conflict, next_index = _conflict_after(tokens, index + 1)
source_end = tokens[next_index - 1].end
uniques.append(
Unique(
(UniqueColumn(column, collation=collation),),
name=pending_name,
column=column,
conflict=conflict,
sql=item[source_start:source_end],
start=base_offset + source_start,
end=base_offset + source_end,
)
)
pending_name = ""
pending_start = None
index = next_index
continue
if (
token.kind == "word"
and token.text.upper() in _OTHER_COLUMN_CONSTRAINT_KEYWORDS
):
pending_name = ""
pending_start = None
index += 1
return uniques
def parse_uniques(create_sql: str) -> list[Unique]:
"""Return column-level and table-level UNIQUE constraints."""
body_info = _table_body(create_sql)
if body_info is None:
return []
body, body_start = body_info
uniques: list[Unique] = []
for item, item_start, _ in _split_spans(body, _lex(body)):
item_tokens = _meaningful(_lex(item))
if not item_tokens:
continue
item_index = 0
constraint_name = ""
if item_tokens[item_index].is_keyword("CONSTRAINT"):
if len(item_tokens) < 2:
raise ParseError("CONSTRAINT is missing its name")
constraint_name = _unquote(item_tokens[1].text)
item_index = 2
head = item_tokens[item_index] if item_index < len(item_tokens) else None
if head and head.is_keyword("UNIQUE"):
if (
item_index + 1 >= len(item_tokens)
or item_tokens[item_index + 1].text != "("
):
raise ParseError("Table UNIQUE must be followed by a column list")
columns, next_index = _unique_columns(item, item_tokens, item_index + 1)
conflict, next_index = _conflict_after(item_tokens, next_index)
if next_index != len(item_tokens):
raise ParseError("Unexpected SQL after UNIQUE constraint")
source_start = item_tokens[0].start
source_end = item_tokens[next_index - 1].end
uniques.append(
Unique(
columns,
name=constraint_name,
conflict=conflict,
sql=item[source_start:source_end],
start=body_start + item_start + source_start,
end=body_start + item_start + source_end,
)
)
continue
if (
head
and head.kind == "word"
and head.text.upper() in _TABLE_CONSTRAINT_KEYWORDS
):
continue
column = _unquote(item_tokens[0].text)
uniques.extend(
_column_uniques(
item,
item_tokens,
column,
body_start + item_start,
)
)
return uniques
def parse_column_comments(create_sql: str) -> dict[str, ColumnComments]: def parse_column_comments(create_sql: str) -> dict[str, ColumnComments]:
"""Return comments immediately before and after each column definition.""" """Return comments immediately before and after each column definition."""
body_info = _table_body(create_sql) body_info = _table_body(create_sql)

View file

@ -24,21 +24,24 @@ from typing import (
) )
from sqlite_fts4 import rank_bm25 from sqlite_fts4 import rank_bm25
from typing_extensions import Self
from sqlite_utils.plugins import ensure_plugins_loaded, pm from sqlite_utils.plugins import ensure_plugins_loaded, pm
from .create_table_parser import ( from .create_table_parser import (
Check, Check,
ColumnComments, ColumnComments,
ParseError, ParseError,
Unique,
UniqueColumn,
check_references_identifier, check_references_identifier,
parse_autoincrement,
parse_checks, parse_checks,
parse_column_comments, parse_column_comments,
parse_uniques,
rewrite_check_expression, rewrite_check_expression,
sql_ends_in_line_comment, sql_ends_in_line_comment,
) )
from .utils import ( from .utils import (
ANY,
OperationalError, OperationalError,
chunks, chunks,
column_affinity, column_affinity,
@ -102,6 +105,24 @@ def _check_constraint_sql(check: Check) -> str:
return f"{prefix}CHECK ({check.check}{newline})" return f"{prefix}CHECK ({check.check}{newline})"
def _unique_constraint_sql(unique: Unique) -> str:
prefix = f"CONSTRAINT {quote_identifier(unique.name)} " if unique.name else ""
if unique.column:
constraint = "UNIQUE"
else:
columns = []
for column in unique.columns:
column_sql = quote_identifier(column.name)
if column.collation:
column_sql += f" COLLATE {quote_identifier(column.collation)}"
if column.order:
column_sql += f" {column.order}"
columns.append(column_sql)
constraint = "UNIQUE ({})".format(", ".join(columns))
conflict = f" ON CONFLICT {unique.conflict}" if unique.conflict else ""
return f"{prefix}{constraint}{conflict}"
def _column_definition_with_comments( def _column_definition_with_comments(
definition: str, comments: ColumnComments | None definition: str, comments: ColumnComments | None
) -> str: ) -> str:
@ -366,6 +387,7 @@ COLUMN_TYPE_MAPPING: dict[Any, str] = {
decimal.Decimal: "REAL", decimal.Decimal: "REAL",
None.__class__: "TEXT", None.__class__: "TEXT",
uuid.UUID: "TEXT", uuid.UUID: "TEXT",
ANY: "ANY",
# SQLite explicit types # SQLite explicit types
"TEXT": "TEXT", "TEXT": "TEXT",
"INTEGER": "INTEGER", "INTEGER": "INTEGER",
@ -380,6 +402,8 @@ COLUMN_TYPE_MAPPING: dict[Any, str] = {
"real": "REAL", "real": "REAL",
"blob": "BLOB", "blob": "BLOB",
"bytes": "BLOB", "bytes": "BLOB",
"ANY": "ANY",
"any": "ANY",
} }
# If numpy is available, add more types # If numpy is available, add more types
if np: if np:
@ -611,7 +635,7 @@ class Database:
pm.hook.prepare_connection(conn=self.conn) pm.hook.prepare_connection(conn=self.conn)
self.strict = strict self.strict = strict
def __enter__(self) -> Self: def __enter__(self):
return self return self
def __exit__( def __exit__(
@ -1418,6 +1442,8 @@ class Database:
strict: bool = False, strict: bool = False,
_checks: Iterable[Check] | None = None, _checks: Iterable[Check] | None = None,
_column_comments: Mapping[str, ColumnComments] | None = None, _column_comments: Mapping[str, ColumnComments] | None = None,
_autoincrement: str | None = None,
_uniques: Iterable[Unique] | None = None,
) -> str: ) -> str:
""" """
Returns the SQL ``CREATE TABLE`` statement for creating the specified table. Returns the SQL ``CREATE TABLE`` statement for creating the specified table.
@ -1480,6 +1506,60 @@ class Database:
checks_by_column.setdefault(column, []).append(check) checks_by_column.setdefault(column, []).append(check)
else: else:
table_checks.append(check) table_checks.append(check)
uniques_by_column: dict[str, list[Unique]] = {}
table_uniques: list[Unique] = []
for unique in _uniques or ():
resolved_unique = Unique(
tuple(
UniqueColumn(
resolve_casing(column.name, columns),
collation=column.collation,
order=column.order,
)
for column in unique.columns
),
name=unique.name,
column=(
resolve_casing(unique.column, columns) if unique.column else ""
),
conflict=unique.conflict,
)
missing = [
column.name
for column in resolved_unique.columns
if column.name not in columns
]
if missing:
raise AlterError(
"No such column for UNIQUE constraint: {}".format(
", ".join(missing)
)
)
if resolved_unique.column:
if (
len(resolved_unique.columns) != 1
or resolved_unique.columns[0].name != resolved_unique.column
):
raise AlterError("Invalid column-level UNIQUE constraint")
if any(
column.collation or column.order
for column in resolved_unique.columns
):
# Render this as a table constraint so the collation or sort
# order that governs uniqueness can be represented explicitly.
table_uniques.append(
Unique(
resolved_unique.columns,
name=resolved_unique.name,
conflict=resolved_unique.conflict,
)
)
else:
uniques_by_column.setdefault(resolved_unique.column, []).append(
resolved_unique
)
else:
table_uniques.append(resolved_unique)
if not columns: if not columns:
raise ValueError("Tables must have at least one column") raise ValueError("Tables must have at least one column")
if not all(n in columns for n in not_null): if not all(n in columns for n in not_null):
@ -1521,10 +1601,22 @@ class Database:
column_items.insert(0, (pk, int)) column_items.insert(0, (pk, int))
elif pk: elif pk:
pk = [resolve_casing(p, [c[0] for c in column_items]) for p in pk] pk = [resolve_casing(p, [c[0] for c in column_items]) for p in pk]
if _autoincrement is not None:
_autoincrement = resolve_casing(
_autoincrement, [c[0] for c in column_items]
)
if _autoincrement != single_pk:
raise ValueError("AUTOINCREMENT requires a single-column primary key")
for column_name, column_type in column_items: for column_name, column_type in column_items:
column_extras = [] column_extras = []
if column_name == single_pk: if column_name == single_pk:
column_extras.append("PRIMARY KEY") column_extras.append("PRIMARY KEY")
if column_name == _autoincrement:
if COLUMN_TYPE_MAPPING[column_type] != "INTEGER":
raise ValueError(
"AUTOINCREMENT requires an INTEGER PRIMARY KEY column"
)
column_extras.append("AUTOINCREMENT")
if column_name in not_null: if column_name in not_null:
column_extras.append("NOT NULL") column_extras.append("NOT NULL")
if column_name in defaults and defaults[column_name] is not None: if column_name in defaults and defaults[column_name] is not None:
@ -1536,6 +1628,10 @@ class Database:
column_extras.append( column_extras.append(
f"REFERENCES {quote_identifier(fk.other_table)}({quote_identifier(cast(str, fk.other_column))}){_fk_actions_sql(fk)}" f"REFERENCES {quote_identifier(fk.other_table)}({quote_identifier(cast(str, fk.other_column))}){_fk_actions_sql(fk)}"
) )
column_extras.extend(
_unique_constraint_sql(unique)
for unique in uniques_by_column.get(column_name, ())
)
column_extras.extend( column_extras.extend(
_check_constraint_sql(check) _check_constraint_sql(check)
for check in checks_by_column.get(column_name, ()) for check in checks_by_column.get(column_name, ())
@ -1582,6 +1678,9 @@ class Database:
actions=_fk_actions_sql(fk), actions=_fk_actions_sql(fk),
) )
) )
column_defs.extend(
f" {_unique_constraint_sql(unique)}" for unique in table_uniques
)
column_defs.extend( column_defs.extend(
f" {_check_constraint_sql(check)}" for check in table_checks f" {_check_constraint_sql(check)}" for check in table_checks
) )
@ -2375,14 +2474,11 @@ class Table(Queryable):
@property @property
def indexes(self) -> list[Index]: def indexes(self) -> list[Index]:
"List of indexes defined on this table." "List of indexes defined on this table."
sql = f'PRAGMA index_list("{self.name}")' sql = f"PRAGMA index_list({quote_identifier(self.name)})"
indexes = [] indexes = []
for row in self.db.execute_returning_dicts(sql): for row in self.db.execute_returning_dicts(sql):
index_name = row["name"] index_name = row["name"]
index_name_quoted = ( column_sql = f"PRAGMA index_info({quote_identifier(index_name)})"
f'"{index_name}"' if not index_name.startswith('"') else index_name
)
column_sql = f"PRAGMA index_info({index_name_quoted})"
columns = [] columns = []
for seqno, cid, name in self.db.execute(column_sql).fetchall(): for seqno, cid, name in self.db.execute(column_sql).fetchall():
columns.append(name) columns.append(name)
@ -2397,14 +2493,11 @@ class Table(Queryable):
@property @property
def xindexes(self) -> list[XIndex]: def xindexes(self) -> list[XIndex]:
"List of indexes defined on this table using the more detailed ``XIndex`` format." "List of indexes defined on this table using the more detailed ``XIndex`` format."
sql = f'PRAGMA index_list("{self.name}")' sql = f"PRAGMA index_list({quote_identifier(self.name)})"
indexes = [] indexes = []
for row in self.db.execute_returning_dicts(sql): for row in self.db.execute_returning_dicts(sql):
index_name = row["name"] index_name = row["name"]
index_name_quoted = ( column_sql = f"PRAGMA index_xinfo({quote_identifier(index_name)})"
f'"{index_name}"' if not index_name.startswith('"') else index_name
)
column_sql = f"PRAGMA index_xinfo({index_name_quoted})"
index_columns = [] index_columns = []
for info in self.db.execute(column_sql).fetchall(): for info in self.db.execute(column_sql).fetchall():
index_columns.append(XIndexColumn(*info)) index_columns.append(XIndexColumn(*info))
@ -2750,6 +2843,8 @@ class Table(Queryable):
try: try:
existing_checks = self.checks existing_checks = self.checks
existing_column_comments = parse_column_comments(self.schema) existing_column_comments = parse_column_comments(self.schema)
existing_autoincrement = parse_autoincrement(self.schema)
existing_uniques = parse_uniques(self.schema)
except ParseError as ex: except ParseError as ex:
raise TransformError( raise TransformError(
f"Could not parse table schema for table {self.name!r}: {ex}" f"Could not parse table schema for table {self.name!r}: {ex}"
@ -2776,6 +2871,37 @@ class Table(Queryable):
) )
) )
create_table_uniques: list[Unique] = []
for unique in existing_uniques:
columns = tuple(
UniqueColumn(
resolve_casing(column.name, existing_columns),
collation=column.collation,
order=column.order,
)
for column in unique.columns
)
if any(column.name in drop for column in columns):
continue
owner = (
resolve_casing(unique.column, existing_columns) if unique.column else ""
)
create_table_uniques.append(
Unique(
tuple(
UniqueColumn(
rename.get(column.name) or column.name,
collation=column.collation,
order=column.order,
)
for column in columns
),
name=unique.name,
column=rename.get(owner) or owner,
conflict=unique.conflict,
)
)
create_table_column_comments: dict[str, ColumnComments] = {} create_table_column_comments: dict[str, ColumnComments] = {}
for column, comments in existing_column_comments.items(): for column, comments in existing_column_comments.items():
owner = resolve_casing(column, existing_columns) owner = resolve_casing(column, existing_columns)
@ -2872,6 +2998,11 @@ class Table(Queryable):
new_column_pairs.append((new_name, type_)) new_column_pairs.append((new_name, type_))
copy_from_to[name] = new_name copy_from_to[name] = new_name
if existing_autoincrement:
existing_autoincrement = resolve_casing(
existing_autoincrement, existing_columns
)
if pk is DEFAULT: if pk is DEFAULT:
pks_renamed = tuple( pks_renamed = tuple(
rename.get(pk_name) or pk_name rename.get(pk_name) or pk_name
@ -2882,6 +3013,28 @@ class Table(Queryable):
else: else:
pk = pks_renamed pk = pks_renamed
create_table_autoincrement = None
if existing_autoincrement and existing_autoincrement not in drop:
renamed_autoincrement = (
rename.get(existing_autoincrement) or existing_autoincrement
)
single_pk = pk[0] if isinstance(pk, (list, tuple)) and len(pk) == 1 else pk
new_column_types = dict(new_column_pairs)
if (
single_pk == renamed_autoincrement
and COLUMN_TYPE_MAPPING.get(new_column_types.get(renamed_autoincrement))
== "INTEGER"
):
create_table_autoincrement = renamed_autoincrement
autoincrement_sequence = None
if create_table_autoincrement:
sequence_row = self.db.execute(
"SELECT seq FROM sqlite_sequence WHERE name = ?", [self.name]
).fetchone()
if sequence_row is not None:
autoincrement_sequence = sequence_row[0]
# not_null may be a set or dict, need to convert to a set # not_null may be a set or dict, need to convert to a set
create_table_not_null = { create_table_not_null = {
rename.get(c.name) or c.name rename.get(c.name) or c.name
@ -2933,9 +3086,24 @@ class Table(Queryable):
strict=self.strict if strict is None else strict, strict=self.strict if strict is None else strict,
_checks=create_table_checks, _checks=create_table_checks,
_column_comments=create_table_column_comments, _column_comments=create_table_column_comments,
_autoincrement=create_table_autoincrement,
_uniques=create_table_uniques,
).strip() ).strip()
) )
# Columns being changed from TEXT to a numeric type: coerce empty strings to NULL
_numeric_sql_types = {"INTEGER", "REAL", "FLOAT", "NUMERIC"}
text_to_numeric_cols = {
col_name
for col_name, new_type in types.items()
if existing_columns.get(col_name) == str
and COLUMN_TYPE_MAPPING.get(
new_type,
new_type.upper() if isinstance(new_type, str) else "",
)
in _numeric_sql_types
}
# Copy across data, respecting any renamed columns # Copy across data, respecting any renamed columns
new_cols = [] new_cols = []
old_cols = [] old_cols = []
@ -2946,13 +3114,96 @@ class Table(Queryable):
if "rowid" not in new_cols: if "rowid" not in new_cols:
new_cols.insert(0, "rowid") new_cols.insert(0, "rowid")
old_cols.insert(0, "rowid") old_cols.insert(0, "rowid")
def _copy_expr(col):
if col in text_to_numeric_cols:
return "NULLIF({}, '')".format(quote_identifier(col))
return quote_identifier(col)
copy_sql = "INSERT INTO {} ({new_cols})\n SELECT {old_cols} FROM {};".format( copy_sql = "INSERT INTO {} ({new_cols})\n SELECT {old_cols} FROM {};".format(
quote_identifier(new_table_name), quote_identifier(new_table_name),
quote_identifier(self.name), quote_identifier(self.name),
old_cols=", ".join(quote_identifier(col) for col in old_cols), old_cols=", ".join(_copy_expr(col) for col in old_cols),
new_cols=", ".join(quote_identifier(col) for col in new_cols), new_cols=", ".join(quote_identifier(col) for col in new_cols),
) )
sqls.append(copy_sql) sqls.append(copy_sql)
# Capture indexes before the old table is changed. Simple indexes that
# reference renamed columns are recreated from structured PRAGMA
# metadata instead of editing their stored CREATE INDEX SQL.
index_drop_sqls = []
index_create_sqls = []
xindexes_by_name = {index.name: index for index in self.xindexes}
for index in self.indexes:
if index.origin == "pk":
continue
index_sql = self.db.execute(
"""SELECT sql FROM sqlite_master WHERE type = 'index' AND name = :index_name;""",
{"index_name": index.name},
).fetchall()[0][0]
if index_sql is None:
if index.origin == "u":
# UNIQUE constraints are reproduced in CREATE TABLE above.
continue
raise TransformError(
f"Index '{index.name}' on table '{self.name}' does not have a "
"CREATE INDEX statement. You must manually drop this index prior to running this "
"transformation and manually recreate the new index after running this transformation."
)
dropped_index_column = next(
(column for column in index.columns if column in drop), None
)
renamed_index_column = next(
(column for column in index.columns if column in rename), None
)
if dropped_index_column is not None:
raise TransformError(
f"Index '{index.name}' column '{dropped_index_column}' is not in updated table '{self.name}'. "
f"You must manually drop this index prior to running this transformation "
f"and manually recreate the new index after running this transformation. "
f"The original index sql statement is: `{index_sql}`. No changes have been applied to this table."
)
xindex = xindexes_by_name[index.name]
indexed_columns = sorted(
(column for column in xindex.columns if column.key),
key=lambda column: column.seqno,
)
if (rename or drop) and (
index.partial or any(column.name is None for column in indexed_columns)
):
raise TransformError(
f"Index '{index.name}' is a partial or expression index, so it "
f"cannot be safely recreated while columns are renamed or dropped. "
f"You must manually drop this index prior to running this transformation "
f"and manually recreate the new index after running this transformation. "
f"The original index sql statement is: `{index_sql}`. No changes have been applied to this table."
)
if renamed_index_column is not None:
columns_sql = []
for column in indexed_columns:
assert column.name is not None
column_sql = quote_identifier(
rename.get(column.name) or column.name
)
if column.coll and column.coll.upper() != "BINARY":
column_sql += f" COLLATE {quote_identifier(column.coll)}"
if column.desc:
column_sql += " DESC"
columns_sql.append(column_sql)
index_sql = "CREATE {unique}INDEX {index_name} ON {table_name} ({columns})".format(
unique="UNIQUE " if index.unique else "",
index_name=quote_identifier(index.name),
table_name=quote_identifier(self.name),
columns=", ".join(columns_sql),
)
index_drop_sqls.append(
f"DROP INDEX IF EXISTS {quote_identifier(index.name)};"
)
elif keep_table:
index_drop_sqls.append(
f"DROP INDEX IF EXISTS {quote_identifier(index.name)};"
)
index_create_sqls.append(index_sql)
sqls.extend(index_drop_sqls)
# Drop (or keep) the old table, then rename the new one into place. # Drop (or keep) the old table, then rename the new one into place.
# Since SQLite 3.25 ALTER TABLE ... RENAME TO rewrites references to # Since SQLite 3.25 ALTER TABLE ... RENAME TO rewrites references to
# the renamed table in every view definition, which fails if a view # the renamed table in every view definition, which fails if a view
@ -2981,30 +3232,25 @@ class Table(Queryable):
"ON" if legacy_alter_table_was_on else "OFF" "ON" if legacy_alter_table_was_on else "OFF"
) )
) )
if autoincrement_sequence is not None:
table_name_literal = self.db.quote(self.name)
sqls.extend(
(
"UPDATE sqlite_sequence SET seq = MAX(seq, {sequence}) "
"WHERE name = {table_name};".format(
sequence=autoincrement_sequence,
table_name=table_name_literal,
),
"INSERT INTO sqlite_sequence (name, seq) "
"SELECT {table_name}, {sequence} WHERE NOT EXISTS "
"(SELECT 1 FROM sqlite_sequence WHERE name = {table_name});".format(
sequence=autoincrement_sequence,
table_name=table_name_literal,
),
)
)
# Re-add existing indexes # Re-add existing indexes
for index in self.indexes: sqls.extend(index_create_sqls)
if index.origin != "pk":
index_sql = self.db.execute(
"""SELECT sql FROM sqlite_master WHERE type = 'index' AND name = :index_name;""",
{"index_name": index.name},
).fetchall()[0][0]
if index_sql is None:
raise TransformError(
f"Index '{index.name}' on table '{self.name}' does not have a "
"CREATE INDEX statement. You must manually drop this index prior to running this "
"transformation and manually recreate the new index after running this transformation."
)
if keep_table:
sqls.append(f"DROP INDEX IF EXISTS {quote_identifier(index.name)};")
for col in index.columns:
if col in rename or col in drop:
raise TransformError(
f"Index '{index.name}' column '{col}' is not in updated table '{self.name}'. "
f"You must manually drop this index prior to running this transformation "
f"and manually recreate the new index after running this transformation. "
f"The original index sql statement is: `{index_sql}`. No changes have been applied to this table."
)
sqls.append(index_sql)
return sqls return sqls
def extract( def extract(
@ -3046,6 +3292,15 @@ class Table(Queryable):
if col in columns if col in columns
} }
if lookup_table.exists(): if lookup_table.exists():
if (
self.strict
and ANY in lookup_columns_definition.values()
and not lookup_table.strict
):
raise InvalidColumns(
f"Lookup table {table} already exists but is not STRICT, "
"so it cannot preserve ANY column values"
)
if not set(lookup_columns_definition.items()).issubset( if not set(lookup_columns_definition.items()).issubset(
lookup_table.columns_dict.items() lookup_table.columns_dict.items()
): ):
@ -3059,6 +3314,7 @@ class Table(Queryable):
**lookup_columns_definition, **lookup_columns_definition,
}, },
pk="id", pk="id",
strict=self.strict,
) )
lookup_columns = [(rename.get(col) or col) for col in columns] lookup_columns = [(rename.get(col) or col) for col in columns]
lookup_table.create_index(lookup_columns, unique=True, if_not_exists=True) lookup_table.create_index(lookup_columns, unique=True, if_not_exists=True)
@ -5276,6 +5532,13 @@ def _decode_default_value(value: str) -> object:
# It's a binary string, stored as hex # It's a binary string, stored as hex
to_decode = value[2:-1] to_decode = value[2:-1]
return binascii.unhexlify(to_decode) return binascii.unhexlify(to_decode)
upper = value.upper()
if upper == "TRUE":
return True
if upper == "FALSE":
return False
if upper == "NULL":
return None
# If it is a string containing a floating point number: # If it is a string containing a floating point number:
try: try:
return float(value) return float(value)

View file

@ -59,6 +59,10 @@ Row = dict[str, RowValue]
T = TypeVar("T") T = TypeVar("T")
class ANY:
"""Marker type for an SQLite ``ANY`` column."""
class _CloseableIterator(Iterator[Row]): class _CloseableIterator(Iterator[Row]):
"""Iterator wrapper that closes a file when iteration is complete.""" """Iterator wrapper that closes a file when iteration is complete."""
@ -178,6 +182,8 @@ def column_affinity(column_type: str) -> type:
return bytes return bytes
if "REAL" in column_type or "FLOA" in column_type or "DOUB" in column_type: if "REAL" in column_type or "FLOA" in column_type or "DOUB" in column_type:
return float return float
if column_type == "ANY":
return ANY
# Default is 'NUMERIC', which we currently also treat as float # Default is 'NUMERIC', which we currently also treat as float
return float return float

View file

@ -9,7 +9,7 @@ from pathlib import Path
import pytest import pytest
from click.testing import CliRunner from click.testing import CliRunner
from sqlite_utils import Database, cli from sqlite_utils import ANY, Database, cli
from sqlite_utils.db import ForeignKey, Index from sqlite_utils.db import ForeignKey, Index
@ -355,6 +355,7 @@ def test_create_index_desc(db_path):
("blob", "BLOB", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'), ("blob", "BLOB", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'),
("blob", "bytes", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'), ("blob", "bytes", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'),
("blob", "BYTES", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'), ("blob", "BYTES", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'),
("anything", "any", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "anything" ANY)'),
("default", None, 'CREATE TABLE "dogs" (\n "name" TEXT\n, "default" TEXT)'), ("default", None, 'CREATE TABLE "dogs" (\n "name" TEXT\n, "default" TEXT)'),
), ),
) )
@ -2007,6 +2008,25 @@ def test_transform_strict_option_with_invalid_data(db_path):
assert not any(name.startswith("dogs_new_") for name in db.table_names()) assert not any(name.startswith("dogs_new_") for name in db.table_names())
def test_transform_column_to_any(db_path):
db = Database(db_path)
if not db.supports_strict:
pytest.skip("SQLite version does not support strict tables")
db.table("items").create({"data": str}, strict=True)
db.table("items").insert({"data": "000123"})
result = CliRunner().invoke(
cli.cli, ["transform", db_path, "items", "--type", "data", "any"]
)
assert result.exit_code == 0, result.output
assert db.table("items").columns_dict == {"data": ANY}
assert db.execute("select typeof(data), data from items").fetchone() == (
"text",
"000123",
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
"extra_args,expected_schema", "extra_args,expected_schema",
( (
@ -2872,6 +2892,30 @@ def test_create_table_strict(strict):
assert db.table("items").columns_dict == {"id": int, "w": float} assert db.table("items").columns_dict == {"id": int, "w": float}
def test_create_table_strict_any():
runner = CliRunner()
with runner.isolated_filesystem():
db = Database("test.db")
if not db.supports_strict:
pytest.skip("SQLite version does not support strict tables")
result = runner.invoke(
cli.cli,
[
"create-table",
"test.db",
"items",
"id",
"integer",
"data",
"any",
"--strict",
],
)
assert result.exit_code == 0, result.output
assert db.table("items").strict is True
assert db.table("items").columns_dict == {"id": int, "data": ANY}
@pytest.mark.parametrize("method", ("insert", "upsert")) @pytest.mark.parametrize("method", ("insert", "upsert"))
@pytest.mark.parametrize("strict", (False, True)) @pytest.mark.parametrize("strict", (False, True))
def test_insert_upsert_strict(tmpdir, method, strict): def test_insert_upsert_strict(tmpdir, method, strict):
@ -2887,6 +2931,39 @@ def test_insert_upsert_strict(tmpdir, method, strict):
assert db.table("items").strict == strict or not db.supports_strict assert db.table("items").strict == strict or not db.supports_strict
@pytest.mark.parametrize("method", ("insert", "upsert"))
def test_insert_upsert_strict_any(tmpdir, method):
db_path = str(tmpdir / "test.db")
db = Database(db_path)
if not db.supports_strict:
pytest.skip("SQLite version does not support strict tables")
db.close()
result = CliRunner().invoke(
cli.cli,
[
method,
db_path,
"items",
"-",
"--csv",
"--pk",
"id",
"--type",
"data",
"any",
"--strict",
],
input="id,data\n1,000123",
)
assert result.exit_code == 0, result.output
db = Database(db_path)
assert db.table("items").columns_dict == {"id": int, "data": ANY}
assert db.execute("select typeof(data), data from items").fetchone() == (
"text",
"000123",
)
def test_extract_bad_column_clean_error(db_path): def test_extract_bad_column_clean_error(db_path):
db = Database(db_path) db = Database(db_path)
db.table("trees").insert({"id": 1, "species": "Palm"}, pk="id") db.table("trees").insert({"id": 1, "species": "Palm"}, pk="id")

View file

@ -181,6 +181,34 @@ def test_convert_dryrun(test_db_and_path):
assert result.output.strip().split("\n")[-1] == "Would affect 1 row" assert result.output.strip().split("\n")[-1] == "Would affect 1 row"
def test_convert_dryrun_table_and_column_names_containing_closing_bracket(
fresh_db_and_path,
):
db, db_path = fresh_db_and_path
table_name = "table]name"
column_name = "column]name"
db[table_name].insert({column_name: "hello"})
result = CliRunner().invoke(
cli.cli,
[
"convert",
db_path,
table_name,
column_name,
"value.upper()",
"--dry-run",
],
catch_exceptions=False,
)
assert result.exit_code == 0
assert result.output.strip() == (
"hello\n --- becomes:\nHELLO\n\nWould affect 1 row"
)
assert list(db[table_name].rows) == [{column_name: "hello"}]
def test_convert_multi_dryrun(test_db_and_path): def test_convert_multi_dryrun(test_db_and_path):
db_path = test_db_and_path[1] db_path = test_db_and_path[1]
result = CliRunner().invoke( result = CliRunner().invoke(

View file

@ -1,5 +1,6 @@
import pytest import pytest
from sqlite_utils import ANY
from sqlite_utils.utils import column_affinity from sqlite_utils.utils import column_affinity
EXAMPLES = [ EXAMPLES = [
@ -26,6 +27,8 @@ EXAMPLES = [
("DOUBLE", float), ("DOUBLE", float),
("DOUBLE PRECISION", float), ("DOUBLE PRECISION", float),
("FLOAT", float), ("FLOAT", float),
("ANY", ANY),
("any", ANY),
# Numeric, treated as float: # Numeric, treated as float:
("NUMERIC", float), ("NUMERIC", float),
("DECIMAL(10,5)", float), ("DECIMAL(10,5)", float),

View file

@ -7,6 +7,7 @@ import uuid
import pytest import pytest
from sqlite_utils import ANY
from sqlite_utils.db import ( from sqlite_utils.db import (
AlterError, AlterError,
Database, Database,
@ -1366,6 +1367,18 @@ def test_quote(fresh_db, input, expected):
{"col": list}, {"col": list},
'"col" TEXT', '"col" TEXT',
), ),
(
{"col": ANY},
'"col" ANY',
),
(
{"col": "ANY"},
'"col" ANY',
),
(
{"col": "any"},
'"col" ANY',
),
), ),
) )
def test_create_table_sql(fresh_db, columns, expected_sql_middle): def test_create_table_sql(fresh_db, columns, expected_sql_middle):
@ -1509,6 +1522,26 @@ def test_create_transform(fresh_db, cols, kwargs, expected_schema, should_transf
assert fresh_db.table("demo").count == 1 assert fresh_db.table("demo").count == 1
def test_create_transform_keyword_literal_defaults_unchanged(fresh_db):
fresh_db.execute(
"create table demo ("
"id integer primary key, "
"enabled integer default TRUE, "
"disabled integer default FALSE, "
"nullable text default NULL"
")"
)
traces = []
with fresh_db.tracer(lambda sql, parameters: traces.append((sql, parameters))):
fresh_db.table("demo").create(
{"id": int, "enabled": int, "disabled": int, "nullable": str},
pk="id",
defaults={"enabled": True, "disabled": False, "nullable": None},
transform=True,
)
assert not any(sql.startswith("CREATE TABLE") for sql, _ in traces)
def test_rename_table(fresh_db): def test_rename_table(fresh_db):
fresh_db.table("t").insert({"foo": "bar"}) fresh_db.table("t").insert({"foo": "bar"})
assert ["t"] == fresh_db.table_names() assert ["t"] == fresh_db.table_names()
@ -1569,6 +1602,33 @@ def test_create_strict(fresh_db, strict):
assert table.strict == strict or not fresh_db.supports_strict assert table.strict == strict or not fresh_db.supports_strict
def test_create_strict_with_any(fresh_db):
if not fresh_db.supports_strict:
pytest.skip("SQLite version does not support strict tables")
table = fresh_db.table("items").create(
{"id": int, "data": ANY}, pk="id", strict=True
)
table.insert_all(
[
{"id": 1, "data": 42},
{"id": 2, "data": "000123"},
{"id": 3, "data": 3.14},
{"id": 4, "data": b"bytes"},
{"id": 5, "data": None},
]
)
assert table.columns_dict == {"id": int, "data": ANY}
assert fresh_db.execute(
"select typeof(data), data from items order by id"
).fetchall() == [
("integer", 42),
("text", "000123"),
("real", 3.14),
("blob", b"bytes"),
("null", None),
]
def test_bad_table_and_view_exceptions(fresh_db): def test_bad_table_and_view_exceptions(fresh_db):
fresh_db.table("t").insert({"id": 1}, pk="id") fresh_db.table("t").insert({"id": 1}, pk="id")
fresh_db.create_view("v", "select * from t") fresh_db.create_view("v", "select * from t")

View file

@ -8,8 +8,12 @@ from sqlite_utils.create_table_parser import (
Check, Check,
ColumnComments, ColumnComments,
ParseError, ParseError,
Unique,
UniqueColumn,
parse_autoincrement,
parse_checks, parse_checks,
parse_column_comments, parse_column_comments,
parse_uniques,
) )
@ -117,6 +121,81 @@ def test_virtual_table_has_no_checks():
) )
@pytest.mark.parametrize(
"sql,expected",
[
(
"CREATE TABLE t(id INTEGER PRIMARY KEY AUTOINCREMENT, value TEXT)",
"id",
),
(
'CREATE TABLE t("quoted id" INTEGER PRIMARY KEY AUTOINCREMENT)',
"quoted id",
),
(
'CREATE TABLE t("autoincrement" INTEGER PRIMARY KEY, value TEXT)',
None,
),
(
"CREATE TABLE t(id INTEGER PRIMARY KEY /* AUTOINCREMENT */, value TEXT)",
None,
),
(
"CREATE TABLE t(id INTEGER PRIMARY KEY, value TEXT CHECK(value != 'AUTOINCREMENT'))",
None,
),
],
)
def test_parse_autoincrement(sql, expected):
sqlite3.connect(":memory:").execute(sql)
assert parse_autoincrement(sql) == expected
def test_parse_column_and_table_uniques():
sql = """
CREATE TABLE memberships (
email TEXT COLLATE RTRIM CONSTRAINT unique_email UNIQUE ON CONFLICT IGNORE,
account_id INTEGER,
CONSTRAINT unique_membership UNIQUE (
account_id DESC,
email COLLATE NOCASE ASC
) ON CONFLICT REPLACE
)
"""
sqlite3.connect(":memory:").execute(sql)
assert parse_uniques(sql) == [
Unique(
(UniqueColumn("email", collation="RTRIM"),),
name="unique_email",
column="email",
conflict="IGNORE",
),
Unique(
(
UniqueColumn("account_id", order="DESC"),
UniqueColumn("email", collation="NOCASE", order="ASC"),
),
name="unique_membership",
conflict="REPLACE",
),
]
uniques = parse_uniques(sql)
assert uniques[0].sql == "CONSTRAINT unique_email UNIQUE ON CONFLICT IGNORE"
assert sql[uniques[1].start : uniques[1].end] == uniques[1].sql
def test_unique_like_text_in_comments_and_checks_is_ignored():
sql = """
CREATE TABLE t (
value TEXT /* UNIQUE ON CONFLICT REPLACE */
CHECK(value != 'UNIQUE(other)'),
other TEXT
)
"""
sqlite3.connect(":memory:").execute(sql)
assert parse_uniques(sql) == []
comment_or_space = st.sampled_from( comment_or_space = st.sampled_from(
[ [
" ", " ",

View file

@ -2,6 +2,7 @@ import itertools
import pytest import pytest
from sqlite_utils import ANY
from sqlite_utils.db import InvalidColumns from sqlite_utils.db import InvalidColumns
@ -305,3 +306,38 @@ def test_extract_repeated_into_shared_lookup_no_nulls(fresh_db):
fresh_db.table("t1").extract(["species"], table="lk") fresh_db.table("t1").extract(["species"], table="lk")
fresh_db.table("t2").extract(["species"], table="lk") fresh_db.table("t2").extract(["species"], table="lk")
assert fresh_db.table("lk").count == 1 assert fresh_db.table("lk").count == 1
def test_extract_preserves_strict_any(fresh_db):
if not fresh_db.supports_strict:
pytest.skip("SQLite version does not support strict tables")
fresh_db.execute("create table items (id integer primary key, data any) strict")
fresh_db.execute("insert into items values (1, ?)", ("000123",))
fresh_db["items"].extract("data", table="data_values")
lookup = fresh_db["data_values"]
assert lookup.strict is True
assert lookup.columns_dict == {"id": int, "data": ANY}
assert fresh_db.execute(
"select typeof(data), data from data_values"
).fetchone() == ("text", "000123")
def test_extract_strict_any_rejects_non_strict_lookup(fresh_db):
if not fresh_db.supports_strict:
pytest.skip("SQLite version does not support strict tables")
fresh_db.execute("create table items (data any) strict")
fresh_db.execute("insert into items values (?)", ("000123",))
fresh_db.execute("create table data_values (id integer primary key, data any)")
with pytest.raises(
InvalidColumns,
match="is not STRICT, so it cannot preserve ANY column values",
):
fresh_db["items"].extract("data", table="data_values")
assert fresh_db.execute("select typeof(data), data from items").fetchone() == (
"text",
"000123",
)

View file

@ -161,6 +161,31 @@ def test_xindexes(fresh_db):
] ]
def test_indexes_with_double_quotes_in_identifiers(fresh_db):
fresh_db['Go"sh'].insert({"id": 1, 'c"1': 2}, pk="id")
fresh_db['Go"sh'].create_index(['c"1'])
assert [(index.name, index.columns) for index in fresh_db['Go"sh'].indexes] == [
('idx_Go"sh_c"1', ['c"1'])
]
assert fresh_db['Go"sh'].xindexes == [
XIndex(
name='idx_Go"sh_c"1',
columns=[
XIndexColumn(seqno=0, cid=1, name='c"1', desc=0, coll="BINARY", key=1),
XIndexColumn(seqno=1, cid=-1, name=None, desc=0, coll="BINARY", key=0),
],
)
]
def test_transform_table_with_double_quotes_in_identifiers(fresh_db):
fresh_db['Go"sh'].insert({"id": 1, 'c"1': 2, "c2": 3}, pk="id")
fresh_db['Go"sh'].create_index(['c"1'])
fresh_db['Go"sh'].transform(types={"c2": str})
assert fresh_db['Go"sh'].columns_dict["c2"] is str
assert [index.columns for index in fresh_db['Go"sh'].indexes] == [['c"1']]
@pytest.mark.parametrize( @pytest.mark.parametrize(
"column,expected_table_guess", "column,expected_table_guess",
( (
@ -368,6 +393,21 @@ def test_table_default_values_escaped_quotes(fresh_db):
assert fresh_db.table("t").default_values == {"name": "O'Brien"} assert fresh_db.table("t").default_values == {"name": "O'Brien"}
def test_table_default_values_keyword_literals(fresh_db):
fresh_db.execute(
"create table t ("
"enabled integer default TRUE, "
"disabled integer default false, "
"nullable text default NULL"
")"
)
assert fresh_db.table("t").default_values == {
"enabled": True,
"disabled": False,
"nullable": None,
}
def test_pks_use_primary_key_declaration_order(fresh_db): def test_pks_use_primary_key_declaration_order(fresh_db):
# PRIMARY KEY (a, b) declared against columns stored in order (b, a) - # PRIMARY KEY (a, b) declared against columns stored in order (b, a) -
# pks must follow the declaration order, which is what SQLite uses to # pks must follow the declaration order, which is what SQLite uses to

View file

@ -2,6 +2,7 @@ import sqlite3
import pytest import pytest
from sqlite_utils import ANY
from sqlite_utils.db import Check, ForeignKey, TransactionError, TransformError from sqlite_utils.db import Check, ForeignKey, TransactionError, TransformError
from sqlite_utils.utils import OperationalError from sqlite_utils.utils import OperationalError
@ -26,7 +27,7 @@ from sqlite_utils.utils import OperationalError
{"types": {"age": int}}, {"types": {"age": int}},
[ [
'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" INTEGER\n);', 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" INTEGER\n);',
'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";', 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", NULLIF("age", \'\') FROM "dogs";',
'DROP TABLE "dogs";', 'DROP TABLE "dogs";',
"PRAGMA legacy_alter_table=ON;", "PRAGMA legacy_alter_table=ON;",
'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";', 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
@ -62,7 +63,7 @@ from sqlite_utils.utils import OperationalError
{"types": {"age": int}, "rename": {"age": "dog_age"}}, {"types": {"age": int}, "rename": {"age": "dog_age"}},
[ [
'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "dog_age" INTEGER\n);', 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "dog_age" INTEGER\n);',
'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "dog_age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";', 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "dog_age")\n SELECT "rowid", "id", "name", NULLIF("age", \'\') FROM "dogs";',
'DROP TABLE "dogs";', 'DROP TABLE "dogs";',
"PRAGMA legacy_alter_table=ON;", "PRAGMA legacy_alter_table=ON;",
'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";', 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
@ -167,7 +168,7 @@ def test_transform_sql_table_with_primary_key(
{"types": {"age": int}}, {"types": {"age": int}},
[ [
'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER,\n "name" TEXT,\n "age" INTEGER\n);', 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER,\n "name" TEXT,\n "age" INTEGER\n);',
'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";', 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", NULLIF("age", \'\') FROM "dogs";',
'DROP TABLE "dogs";', 'DROP TABLE "dogs";',
"PRAGMA legacy_alter_table=ON;", "PRAGMA legacy_alter_table=ON;",
'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";', 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
@ -823,6 +824,55 @@ def test_transform_to_strict_not_supported(fresh_db, method_name):
assert table.strict is False assert table.strict is False
def test_transform_preserves_any_column_in_strict_table(fresh_db):
if not fresh_db.supports_strict:
pytest.skip("SQLite version does not support strict tables")
fresh_db.execute("create table items (id integer primary key, data any) strict")
fresh_db.conn.executemany(
"insert into items values (?, ?)",
[
(1, 42),
(2, "000123"),
(3, 3.14),
(4, b"bytes"),
(5, None),
],
)
table = fresh_db["items"]
table.transform()
assert table.strict is True
assert table.columns_dict == {"id": int, "data": ANY}
assert fresh_db.execute(
"select typeof(data), data from items order by id"
).fetchall() == [
("integer", 42),
("text", "000123"),
("real", 3.14),
("blob", b"bytes"),
("null", None),
]
def test_transform_any_column_from_strict_to_non_strict(fresh_db):
if not fresh_db.supports_strict:
pytest.skip("SQLite version does not support strict tables")
fresh_db.execute("create table items (data any) strict")
fresh_db.execute("insert into items values (?)", ("000123",))
table = fresh_db["items"]
table.transform(strict=False)
assert table.strict is False
assert table.columns_dict == {"data": ANY}
# Ordinary non-STRICT ANY columns apply NUMERIC affinity
assert fresh_db.execute("select typeof(data), data from items").fetchone() == (
"integer",
123,
)
@pytest.mark.parametrize( @pytest.mark.parametrize(
"indexes, transform_params", "indexes, transform_params",
[ [
@ -893,22 +943,15 @@ def test_transform_retains_indexes_with_foreign_keys(fresh_db):
), f"Indexes before transform: {indexes_before_transform}\nIndexes after transform: {dogs.indexes}" ), f"Indexes before transform: {indexes_before_transform}\nIndexes after transform: {dogs.indexes}"
@pytest.mark.parametrize( def test_transform_with_indexes_errors(fresh_db):
"transform_params", # Should error with a compound (name, age) index if age is dropped
[
{"rename": {"age": "dog_age"}},
{"drop": ["age"]},
],
)
def test_transform_with_indexes_errors(fresh_db, transform_params):
# Should error with a compound (name, age) index if age is renamed or dropped
dogs = fresh_db.table("dogs") dogs = fresh_db.table("dogs")
dogs.insert({"id": 1, "name": "Cleo", "age": 5}, pk="id") dogs.insert({"id": 1, "name": "Cleo", "age": 5}, pk="id")
dogs.create_index(["name", "age"]) dogs.create_index(["name", "age"])
with pytest.raises(TransformError) as excinfo: with pytest.raises(TransformError) as excinfo:
dogs.transform(**transform_params) dogs.transform(drop=["age"])
assert ( assert (
"Index 'idx_dogs_name_age' column 'age' is not in updated table 'dogs'. " "Index 'idx_dogs_name_age' column 'age' is not in updated table 'dogs'. "
@ -917,30 +960,199 @@ def test_transform_with_indexes_errors(fresh_db, transform_params):
) )
@pytest.mark.parametrize(
("table_name", "index_name"),
(("name", "idx_name"), ("t", "name")),
)
def test_transform_rename_column_with_index(fresh_db, table_name, index_name):
# https://github.com/simonw/sqlite-utils/issues/822
# Use the same name for the table, column and index to ensure only the
# indexed column changes.
table = fresh_db.table(table_name)
table.insert({"id": 1, "name": "Cleo"}, pk="id")
table.create_index(["name"], index_name=index_name)
sqls = table.transform_sql(rename={"name": "full_name"}, tmp_suffix="suffix")
drop_index_sql = f'DROP INDEX IF EXISTS "{index_name}";'
assert drop_index_sql in sqls
assert sqls.index(drop_index_sql) < sqls.index(f'DROP TABLE "{table_name}";')
table.transform(rename={"name": "full_name"})
assert [column.name for column in table.columns] == ["id", "full_name"]
assert [(index.name, index.columns) for index in table.indexes] == [
(index_name, ["full_name"])
]
def test_transform_recreates_renamed_index_from_metadata(fresh_db):
table = fresh_db.table("t")
table.insert({"alpha": "one", "beta": "two"})
# Deliberately use unquoted SQL and index details that need to survive the
# reconstruction. Renaming both columns also guards against cascading
# string substitutions.
fresh_db.execute(
"CREATE UNIQUE INDEX swap_idx ON t(alpha COLLATE NOCASE DESC, beta)"
)
table.transform(rename={"alpha": "beta", "beta": "alpha"})
assert table.columns_dict == {"beta": str, "alpha": str}
assert [(index.name, index.unique, index.columns) for index in table.indexes] == [
("swap_idx", 1, ["beta", "alpha"])
]
key_columns = [column for column in table.xindexes[0].columns if column.key]
assert [(column.name, column.desc, column.coll) for column in key_columns] == [
("beta", 1, "NOCASE"),
("alpha", 0, "BINARY"),
]
@pytest.mark.parametrize(
"index_sql",
(
"CREATE INDEX idx_t_name ON t(lower(name))",
"CREATE INDEX idx_t_name ON t(name) WHERE name IS NOT NULL",
),
)
def test_transform_rename_complex_index_errors(fresh_db, index_sql):
table = fresh_db.table("t")
table.insert({"id": 1, "name": "Cleo"}, pk="id")
fresh_db.execute(index_sql)
with pytest.raises(TransformError, match="partial or expression index"):
table.transform(rename={"name": "full_name"})
assert table.columns_dict == {"id": int, "name": str}
assert [index.name for index in table.indexes] == ["idx_t_name"]
def test_transform_with_unique_constraint_implicit_index(fresh_db): def test_transform_with_unique_constraint_implicit_index(fresh_db):
dogs = fresh_db.table("dogs") dogs = fresh_db.table("dogs")
# Create a table with a UNIQUE constraint on 'name', which creates an implicit index # Create a table with a UNIQUE constraint on 'name', which creates an implicit index
fresh_db.execute(""" fresh_db.execute("""
CREATE TABLE dogs ( CREATE TABLE dogs (
id INTEGER PRIMARY KEY, id INTEGER PRIMARY KEY,
name TEXT UNIQUE, name TEXT UNIQUE ON CONFLICT IGNORE,
age INTEGER age INTEGER
); );
""") """)
dogs.insert({"id": 1, "name": "Cleo", "age": 5}) dogs.insert({"id": 1, "name": "Cleo", "age": 5})
# Attempt to transform the table without modifying 'name' dogs.transform(types={"age": str}, rename={"name": "dog_name"})
with pytest.raises(TransformError) as excinfo:
dogs.transform(types={"age": str}) assert 'dog_name" TEXT UNIQUE ON CONFLICT IGNORE' in dogs.schema
dogs.insert({"id": 2, "dog_name": "Cleo", "age": "6"})
assert list(dogs.rows) == [{"id": 1, "dog_name": "Cleo", "age": "5"}]
def test_transform_preserves_composite_unique_constraint(fresh_db):
fresh_db.execute("""
CREATE TABLE memberships (
account_id INTEGER,
email TEXT,
note TEXT,
CONSTRAINT unique_membership
UNIQUE (account_id DESC, email COLLATE NOCASE)
ON CONFLICT ABORT
)
""")
memberships = fresh_db.table("memberships")
memberships.insert({"account_id": 1, "email": "one@example.com", "note": "x"})
memberships.transform(rename={"account_id": "organization_id"}, types={"note": str})
assert ( assert (
"Index 'sqlite_autoindex_dogs_1' on table 'dogs' does not have a CREATE INDEX statement." 'CONSTRAINT "unique_membership" UNIQUE '
in str(excinfo.value) '("organization_id" DESC, "email" COLLATE "NOCASE") ON CONFLICT ABORT'
in memberships.schema
) )
assert ( with pytest.raises(sqlite3.IntegrityError):
"You must manually drop this index prior to running this transformation and manually recreate the new index after running this transformation." memberships.insert(
in str(excinfo.value) {"organization_id": 1, "email": "ONE@example.com", "note": "y"}
)
def test_transform_preserves_column_unique_collation(fresh_db):
fresh_db.execute("""
CREATE TABLE people (
id INTEGER PRIMARY KEY,
name TEXT COLLATE NOCASE UNIQUE
)
""")
people = fresh_db.table("people")
people.insert({"id": 1, "name": "Cleo"})
people.transform(rename={"name": "full_name"})
assert 'UNIQUE ("full_name" COLLATE "NOCASE")' in people.schema
with pytest.raises(sqlite3.IntegrityError):
people.insert({"id": 2, "full_name": "cleo"})
def test_transform_drops_entire_composite_unique_constraint(fresh_db):
fresh_db.execute("""
CREATE TABLE memberships (
account_id INTEGER,
email TEXT,
UNIQUE (account_id, email)
)
""")
memberships = fresh_db.table("memberships")
memberships.insert({"account_id": 1, "email": "one@example.com"})
memberships.transform(drop={"email"})
assert "UNIQUE" not in memberships.schema
memberships.insert({"account_id": 1})
def test_transform_preserves_autoincrement_and_sequence(fresh_db):
fresh_db.execute(
"CREATE TABLE entries (id INTEGER PRIMARY KEY AUTOINCREMENT, value TEXT)"
) )
entries = fresh_db.table("entries")
entries.insert_all(({"value": "one"}, {"value": "two"}))
entries.delete(2)
entries.transform(rename={"value": "label"})
assert "PRIMARY KEY AUTOINCREMENT" in entries.schema
entries.insert({"label": "three"})
assert list(entries.rows) == [
{"id": 1, "label": "one"},
{"id": 3, "label": "three"},
]
@pytest.mark.parametrize(
"new_type,expected_value,expected_type",
[
(int, 42, int),
(float, 42.0, float),
("integer", 42, int),
("float", 42.0, float),
("REAL", 42.0, float),
],
)
def test_transform_empty_string_to_null_for_numeric_types(
fresh_db, new_type, expected_value, expected_type
):
fresh_db["test"].insert_all(
[
{"id": 1, "value": "42"},
{"id": 2, "value": ""},
{"id": 3, "value": None},
{"id": 4, "value": " "},
]
)
fresh_db["test"].transform(types={"value": new_type})
rows = {r["id"]: r["value"] for r in fresh_db["test"].rows}
assert rows[1] == expected_value
assert type(rows[1]) is expected_type
assert rows[2] is None
assert rows[3] is None
assert rows[4] == " "
def test_transform_preserves_view(fresh_db): def test_transform_preserves_view(fresh_db):