Preserve column before/after comments through .transform()

Refs #762

The before comment comes before the column definition - the after
comment is anything after it but before its trailing comma.

docs/changelog.rst

@@ -9,6 +9,9 @@
 Unreleased
 ----------
 
+- New ``table.checks``, ``table.column_checks`` and ``table.table_checks`` introspection properties expose column-level and table-level ``CHECK`` constraints. (:issue:`834`)
+- ``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 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:

docs/python-api.rst

@@ -1986,6 +1986,17 @@ A bare column name drops any foreign key that column participates in, including
 
 Renaming a column with ``rename=`` updates any foreign keys that use it, and dropping a column with ``drop=`` also drops any foreign keys it participates in - for a compound foreign key this removes the whole constraint.
 
+.. _python_api_transform_check_constraints:
+
+CHECK constraints
+-----------------
+
+``.transform()`` preserves both column-level and table-level ``CHECK`` constraints. If a column is renamed, references to that column in the check expression are renamed too.
+
+A column-level check is removed if its owning column is dropped. Dropping a column referenced by any remaining check raises ``TransformError`` instead of creating an invalid or unexpectedly weakened schema.
+
+Comments immediately before or after a column definition are preserved too. They move with that column if it is renamed or reordered, and are removed if the column is dropped. A comment between two column definitions is treated as belonging to the following column.
+
 .. _python_api_transform_views:
 
 Tables referenced by views
@@ -2480,6 +2491,43 @@ Almost all SQLite tables have a ``rowid`` column, but a table with no explicitly
     False
 
 
+.. _python_api_introspection_checks:
+
+.checks
+-------
+
+The ``.checks`` property returns the column-level and table-level ``CHECK`` constraints defined on a table, as a list of ``Check`` objects. Each object has ``check`` (the expression inside ``CHECK (...)``), ``name``, ``column`` and ``options`` attributes. ``column`` is an empty string for a table-level check. ``options`` contains a list of values only when a column check consists entirely of ``column IN (literal, ...)``. The original constraint fragment is available as ``sql``; ``start`` and ``end`` are its offsets within ``table.schema``.
+
+.. code-block:: python
+
+    >>> db["scores"].checks
+    [Check(check='score > 0', name='positive', column='score', options=None),
+     Check(check='score <= maximum', name='within_maximum', column='', options=None)]
+
+.. _python_api_introspection_column_checks:
+
+.column_checks
+--------------
+
+The ``.column_checks`` property returns the column-level checks grouped by column name:
+
+.. code-block:: python
+
+    >>> db["scores"].column_checks
+    {'score': [Check(check='score > 0', name='positive', column='score', options=None)]}
+
+.. _python_api_introspection_table_checks:
+
+.table_checks
+-------------
+
+The ``.table_checks`` property returns only the table-level checks:
+
+.. code-block:: python
+
+    >>> db["scores"].table_checks
+    [Check(check='score <= maximum', name='within_maximum', column='', options=None)]
+
 .. _python_api_introspection_foreign_keys:
 
 .foreign_keys

sqlite_utils/create_table_parser.py

@@ -0,0 +1,676 @@
+"""Helpers for parsing CHECK constraints from SQLite CREATE TABLE SQL.
+
+SQLite does not expose CHECK constraints through a pragma, so preserving them
+across a table rebuild requires reading ``sqlite_schema.sql``.  This module is
+deliberately small, but it uses a real lexer: strings, quoted identifiers and
+comments are opaque, every token retains its source span and malformed input is
+reported instead of being silently under-parsed.
+"""
+
+import re
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class Check:
+    check: str
+    name: str = ""
+    column: str = ""
+    options: list[Any] | None = None
+    # Source details are excluded from equality and repr so callers can compare
+    # semantic constraints while still having the original SQL available for
+    # diagnostics or future lossless edits.
+    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)
+
+
+@dataclass(frozen=True)
+class ColumnComments:
+    before: str = ""
+    after: str = ""
+
+
+class ParseError(ValueError):
+    pass
+
+
+@dataclass(frozen=True)
+class _Token:
+    kind: str
+    text: str
+    start: int
+    end: int
+
+    def is_keyword(self, keyword: str) -> bool:
+        return self.kind == "word" and self.text.upper() == keyword
+
+
+_PUNCTUATION = frozenset("(),.;+-*/%<>=!~|&?:")
+_TRIVIA = frozenset(("whitespace", "comment"))
+_TABLE_CONSTRAINT_KEYWORDS = frozenset(("PRIMARY", "UNIQUE", "CHECK", "FOREIGN"))
+_OTHER_COLUMN_CONSTRAINT_KEYWORDS = frozenset(
+    ("PRIMARY", "UNIQUE", "REFERENCES", "DEFAULT", "NOT", "COLLATE", "GENERATED")
+)
+_SQLITE_KEYWORDS = frozenset(
+    (
+        "ABORT",
+        "ACTION",
+        "ADD",
+        "AFTER",
+        "ALL",
+        "ALTER",
+        "ANALYZE",
+        "AND",
+        "AS",
+        "ASC",
+        "ATTACH",
+        "AUTOINCREMENT",
+        "BEFORE",
+        "BEGIN",
+        "BETWEEN",
+        "BY",
+        "CASCADE",
+        "CASE",
+        "CAST",
+        "CHECK",
+        "COLLATE",
+        "COLUMN",
+        "COMMIT",
+        "CONFLICT",
+        "CONSTRAINT",
+        "CREATE",
+        "CROSS",
+        "CURRENT_DATE",
+        "CURRENT_TIME",
+        "CURRENT_TIMESTAMP",
+        "DATABASE",
+        "DEFAULT",
+        "DEFERRABLE",
+        "DEFERRED",
+        "DELETE",
+        "DESC",
+        "DETACH",
+        "DISTINCT",
+        "DO",
+        "DROP",
+        "EACH",
+        "ELSE",
+        "END",
+        "ESCAPE",
+        "EXCEPT",
+        "EXCLUDE",
+        "EXCLUSIVE",
+        "EXISTS",
+        "EXPLAIN",
+        "FAIL",
+        "FALSE",
+        "FILTER",
+        "FIRST",
+        "FOLLOWING",
+        "FOR",
+        "FOREIGN",
+        "FROM",
+        "FULL",
+        "GENERATED",
+        "GLOB",
+        "GROUP",
+        "GROUPS",
+        "HAVING",
+        "IF",
+        "IGNORE",
+        "IMMEDIATE",
+        "IN",
+        "INDEX",
+        "INDEXED",
+        "INITIALLY",
+        "INNER",
+        "INSERT",
+        "INSTEAD",
+        "INTERSECT",
+        "INTO",
+        "IS",
+        "ISNULL",
+        "JOIN",
+        "KEY",
+        "LAST",
+        "LEFT",
+        "LIKE",
+        "LIMIT",
+        "MATCH",
+        "MATERIALIZED",
+        "NATURAL",
+        "NO",
+        "NOT",
+        "NOTHING",
+        "NOTNULL",
+        "NULL",
+        "NULLS",
+        "OF",
+        "OFFSET",
+        "ON",
+        "OR",
+        "ORDER",
+        "OTHERS",
+        "OUTER",
+        "OVER",
+        "PARTITION",
+        "PLAN",
+        "PRAGMA",
+        "PRECEDING",
+        "PRIMARY",
+        "QUERY",
+        "RAISE",
+        "RANGE",
+        "RECURSIVE",
+        "REFERENCES",
+        "REGEXP",
+        "REINDEX",
+        "RELEASE",
+        "RENAME",
+        "REPLACE",
+        "RESTRICT",
+        "RETURNING",
+        "RIGHT",
+        "ROLLBACK",
+        "ROW",
+        "ROWS",
+        "SAVEPOINT",
+        "SELECT",
+        "SET",
+        "STRICT",
+        "TABLE",
+        "TEMP",
+        "TEMPORARY",
+        "THEN",
+        "TIES",
+        "TO",
+        "TRANSACTION",
+        "TRIGGER",
+        "TRUE",
+        "UNBOUNDED",
+        "UNION",
+        "UNIQUE",
+        "UPDATE",
+        "USING",
+        "VACUUM",
+        "VALUES",
+        "VIEW",
+        "VIRTUAL",
+        "WHEN",
+        "WHERE",
+        "WINDOW",
+        "WITH",
+        "WITHOUT",
+    )
+)
+_INTEGER_RE = re.compile(r"[+-]?(?:0[xX][0-9a-fA-F]+|[0-9]+)\Z")
+_FLOAT_RE = re.compile(
+    r"[+-]?(?:(?:[0-9]+\.[0-9]*|\.[0-9]+)(?:[eE][+-]?[0-9]+)?|"
+    r"[0-9]+[eE][+-]?[0-9]+)\Z"
+)
+
+
+def _lex(sql: str) -> list[_Token]:
+    tokens: list[_Token] = []
+    i = 0
+    while i < len(sql):
+        start = i
+        char = sql[i]
+        if char.isspace():
+            i += 1
+            while i < len(sql) and sql[i].isspace():
+                i += 1
+            tokens.append(_Token("whitespace", sql[start:i], start, i))
+            continue
+        if sql.startswith("--", i):
+            newline = sql.find("\n", i + 2)
+            i = len(sql) if newline == -1 else newline + 1
+            tokens.append(_Token("comment", sql[start:i], start, i))
+            continue
+        if sql.startswith("/*", i):
+            end = sql.find("*/", i + 2)
+            if end == -1:
+                raise ParseError("Unterminated SQL comment")
+            i = end + 2
+            tokens.append(_Token("comment", sql[start:i], start, i))
+            continue
+        if char in ("'", '"', "`"):
+            quote = char
+            i += 1
+            while i < len(sql):
+                if sql[i] == quote:
+                    if i + 1 < len(sql) and sql[i + 1] == quote:
+                        i += 2
+                        continue
+                    i += 1
+                    break
+                i += 1
+            else:
+                raise ParseError(f"Unterminated {quote} quoted token")
+            kind = "string" if quote == "'" else "identifier"
+            tokens.append(_Token(kind, sql[start:i], start, i))
+            continue
+        if char == "[":
+            end = sql.find("]", i + 1)
+            if end == -1:
+                raise ParseError("Unterminated [ quoted identifier")
+            i = end + 1
+            tokens.append(_Token("identifier", sql[start:i], start, i))
+            continue
+        if char in _PUNCTUATION:
+            i += 1
+            tokens.append(_Token("punct", char, start, i))
+            continue
+        # SQLite accepts any character >= U+0080 in a bare identifier.  More
+        # generally, consume until a lexical delimiter rather than relying on
+        # Python's narrower definition of an alphanumeric character.
+        i += 1
+        while i < len(sql):
+            if sql[i].isspace() or sql[i] in _PUNCTUATION or sql[i] in "'\"`[":
+                break
+            i += 1
+        tokens.append(_Token("word", sql[start:i], start, i))
+    return tokens
+
+
+def _meaningful(tokens: list[_Token]) -> list[_Token]:
+    return [token for token in tokens if token.kind not in _TRIVIA]
+
+
+def _unquote(token: str) -> str:
+    if len(token) >= 2 and token[0] in ("'", '"', "`") and token[-1] == token[0]:
+        return token[1:-1].replace(token[0] * 2, token[0])
+    if len(token) >= 2 and token[0] == "[" and token[-1] == "]":
+        return token[1:-1]
+    return token
+
+
+def _matching_paren(tokens: list[_Token], open_index: int) -> int:
+    if tokens[open_index].text != "(":
+        raise ParseError("Expected an opening parenthesis")
+    depth = 0
+    for index in range(open_index, len(tokens)):
+        if tokens[index].text == "(":
+            depth += 1
+        elif tokens[index].text == ")":
+            depth -= 1
+            if depth == 0:
+                return index
+    raise ParseError("Unbalanced parentheses")
+
+
+def _split_spans(sql: str, tokens: list[_Token]) -> list[tuple[str, int, int]]:
+    if not tokens:
+        return []
+    items: list[tuple[str, int, int]] = []
+    depth = 0
+    start = tokens[0].start
+    for token in tokens:
+        if token.text == "(":
+            depth += 1
+        elif token.text == ")":
+            depth -= 1
+            if depth < 0:
+                raise ParseError("Unbalanced parentheses")
+        elif token.text == "," and depth == 0:
+            raw = sql[start : token.start]
+            item = raw.strip()
+            if item:
+                item_start = start + len(raw) - len(raw.lstrip())
+                items.append((item, item_start, item_start + len(item)))
+            start = token.end
+    if depth:
+        raise ParseError("Unbalanced parentheses")
+    raw = sql[start : tokens[-1].end]
+    item = raw.strip()
+    if item:
+        item_start = start + len(raw) - len(raw.lstrip())
+        items.append((item, item_start, item_start + len(item)))
+    return items
+
+
+def _split_ranges(sql: str, tokens: list[_Token]) -> list[str]:
+    return [item for item, _, _ in _split_spans(sql, tokens)]
+
+
+def _strip_outer_parens(tokens: list[_Token]) -> list[_Token]:
+    while tokens and tokens[0].text == "(":
+        close = _matching_paren(tokens, 0)
+        if close != len(tokens) - 1:
+            break
+        tokens = tokens[1:-1]
+    return tokens
+
+
+_NO_LITERAL = object()
+
+
+def _literal_value(text: str) -> Any:
+    tokens = _meaningful(_lex(text))
+    if len(tokens) == 1 and tokens[0].kind == "string":
+        return _unquote(tokens[0].text)
+    raw = "".join(token.text for token in tokens)
+    if raw.upper() == "NULL":
+        return None
+    if raw.upper() == "TRUE":
+        return True
+    if raw.upper() == "FALSE":
+        return False
+    if _INTEGER_RE.fullmatch(raw):
+        try:
+            return (
+                int(raw, 16) if raw.lower().lstrip("+-").startswith("0x") else int(raw)
+            )
+        except ValueError:
+            return _NO_LITERAL
+    if _FLOAT_RE.fullmatch(raw):
+        try:
+            return float(raw)
+        except ValueError:
+            return _NO_LITERAL
+    return _NO_LITERAL
+
+
+def _ascii_fold(identifier: str) -> str:
+    return identifier.translate(
+        str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz")
+    )
+
+
+def _parse_options(expression: str, column: str) -> list[Any] | None:
+    tokens = _strip_outer_parens(_meaningful(_lex(expression)))
+    if len(tokens) < 4:
+        return None
+    lhs = tokens[0]
+    if lhs.kind not in ("word", "identifier"):
+        return None
+    if column and _ascii_fold(_unquote(lhs.text)) != _ascii_fold(column):
+        return None
+    if not tokens[1].is_keyword("IN") or tokens[2].text != "(":
+        return None
+    close = _matching_paren(tokens, 2)
+    if close != len(tokens) - 1:
+        return None
+    inner = expression[tokens[2].end : tokens[close].start]
+    inner_tokens = _lex(inner)
+    if not _meaningful(inner_tokens):
+        return []
+    values = []
+    for item in _split_ranges(inner, inner_tokens):
+        value = _literal_value(item)
+        if value is _NO_LITERAL:
+            return None
+        values.append(value)
+    return values
+
+
+def _check_after(
+    item: str,
+    tokens: list[_Token],
+    check_index: int,
+    name: str,
+    column: str,
+    constraint_start: int,
+    base_offset: int,
+) -> tuple[Check, int]:
+    if check_index + 1 >= len(tokens) or tokens[check_index + 1].text != "(":
+        raise ParseError("CHECK must be followed by a parenthesized expression")
+    close = _matching_paren(tokens, check_index + 1)
+    expression = item[tokens[check_index + 1].end : tokens[close].start].strip()
+    source_start = tokens[constraint_start].start
+    source_end = tokens[close].end
+    return (
+        Check(
+            expression,
+            name=name,
+            column=column,
+            options=_parse_options(expression, column),
+            sql=item[source_start:source_end],
+            start=base_offset + source_start,
+            end=base_offset + source_end,
+        ),
+        close + 1,
+    )
+
+
+def _column_checks(
+    item: str, tokens: list[_Token], column: str, base_offset: int
+) -> list[Check]:
+    checks: list[Check] = []
+    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("CHECK"):
+            check, index = _check_after(
+                item,
+                tokens,
+                index,
+                pending_name,
+                column,
+                pending_start if pending_start is not None else index,
+                base_offset,
+            )
+            checks.append(check)
+            pending_name = ""
+            pending_start = None
+            continue
+        if (
+            token.kind == "word"
+            and token.text.upper() in _OTHER_COLUMN_CONSTRAINT_KEYWORDS
+        ):
+            pending_name = ""
+            pending_start = None
+        index += 1
+    return checks
+
+
+def _table_body(create_sql: str) -> tuple[str, int] | None:
+    all_tokens = _lex(create_sql)
+    tokens = _meaningful(all_tokens)
+    if not tokens or not tokens[0].is_keyword("CREATE"):
+        raise ParseError("Expected CREATE TABLE")
+    index = 1
+    if index < len(tokens) and (
+        tokens[index].is_keyword("TEMP") or tokens[index].is_keyword("TEMPORARY")
+    ):
+        index += 1
+    if index < len(tokens) and tokens[index].is_keyword("VIRTUAL"):
+        return None
+    if index >= len(tokens) or not tokens[index].is_keyword("TABLE"):
+        raise ParseError("Expected CREATE TABLE")
+    index += 1
+    if (
+        index + 2 < len(tokens)
+        and tokens[index].is_keyword("IF")
+        and tokens[index + 1].is_keyword("NOT")
+        and tokens[index + 2].is_keyword("EXISTS")
+    ):
+        index += 3
+    if index >= len(tokens):
+        raise ParseError("CREATE TABLE is missing its table name")
+    index += 1
+    if index + 1 < len(tokens) and tokens[index].text == ".":
+        index += 2
+    if index < len(tokens) and tokens[index].is_keyword("AS"):
+        return None
+    if index >= len(tokens) or tokens[index].text != "(":
+        raise ParseError("CREATE TABLE is missing its column list")
+    close = _matching_paren(tokens, index)
+    trailing = tokens[close + 1 :]
+    allowed_trailing = {"STRICT", "WITHOUT", "ROWID", ",", ";"}
+    if any(token.text.upper() not in allowed_trailing for token in trailing):
+        raise ParseError("Unexpected SQL after CREATE TABLE column list")
+
+    body_start = tokens[index].end
+    body_end = tokens[close].start
+    return create_sql[body_start:body_end], body_start
+
+
+def parse_checks(create_sql: str) -> list[Check]:
+    """Return CHECK constraints from a valid SQLite CREATE TABLE statement."""
+    body_info = _table_body(create_sql)
+    if body_info is None:
+        return []
+    body, body_start = body_info
+    body_tokens = _lex(body)
+    checks: list[Check] = []
+    for item, item_start, _ in _split_spans(body, body_tokens):
+        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.kind == "word"
+            and head.text.upper() in _TABLE_CONSTRAINT_KEYWORDS
+        ):
+            if head.is_keyword("CHECK"):
+                check, _ = _check_after(
+                    item,
+                    item_tokens,
+                    item_index,
+                    constraint_name,
+                    "",
+                    0,
+                    body_start + item_start,
+                )
+                checks.append(check)
+            continue
+        column = _unquote(item_tokens[0].text)
+        checks.extend(
+            _column_checks(item, item_tokens, column, body_start + item_start)
+        )
+    return checks
+
+
+def parse_column_comments(create_sql: str) -> dict[str, ColumnComments]:
+    """Return comments immediately before and after each column definition."""
+    body_info = _table_body(create_sql)
+    if body_info is None:
+        return {}
+    body, _ = body_info
+    comments: dict[str, ColumnComments] = {}
+    for item, _, _ in _split_spans(body, _lex(body)):
+        item_tokens = _meaningful(_lex(item))
+        if not item_tokens:
+            continue
+        item_index = 0
+        if item_tokens[item_index].is_keyword("CONSTRAINT"):
+            item_index = 2
+        head = item_tokens[item_index] if item_index < len(item_tokens) else None
+        if (
+            head
+            and head.kind == "word"
+            and head.text.upper() in _TABLE_CONSTRAINT_KEYWORDS
+        ):
+            continue
+        column = _unquote(item_tokens[0].text)
+        before = item[: item_tokens[0].start].strip()
+        after = item[item_tokens[-1].end :].strip()
+        if before or after:
+            comments[column] = ColumnComments(before=before, after=after)
+    return comments
+
+
+def _is_identifier_token(tokens: list[_Token], index: int) -> bool:
+    token = tokens[index]
+    if index + 1 < len(tokens) and tokens[index + 1].text in ("(", "."):
+        return False
+    if index and (
+        tokens[index - 1].is_keyword("COLLATE") or tokens[index - 1].is_keyword("AS")
+    ):
+        return False
+    if token.kind == "identifier":
+        return True
+    if token.kind != "word" or token.text.upper() in _SQLITE_KEYWORDS:
+        return False
+    return True
+
+
+def check_references_identifier(expression: str, identifier: str) -> bool:
+    tokens = _meaningful(_lex(expression))
+    folded = _ascii_fold(identifier)
+    return any(
+        _is_identifier_token(tokens, index)
+        and _ascii_fold(_unquote(token.text)) == folded
+        for index, token in enumerate(tokens)
+    )
+
+
+def sql_ends_in_line_comment(sql: str) -> bool:
+    """Return True if appended SQL would be swallowed by a ``--`` comment."""
+    tokens = _lex(sql)
+    if not tokens:
+        return False
+    final = tokens[-1]
+    return (
+        final.kind == "comment"
+        and final.text.startswith("--")
+        and not final.text.endswith(("\n", "\r"))
+    )
+
+
+def _valid_bare_identifier(identifier: str) -> bool:
+    if not identifier or identifier.upper() in _SQLITE_KEYWORDS:
+        return False
+    first = identifier[0]
+    if not (first == "_" or first.isalpha() or ord(first) >= 0x80):
+        return False
+    return all(
+        char == "_" or char == "$" or char.isalnum() or ord(char) >= 0x80
+        for char in identifier[1:]
+    )
+
+
+def _quote_replacement(original: str, replacement: str) -> str:
+    if original.startswith('"'):
+        return '"{}"'.format(replacement.replace('"', '""'))
+    if original.startswith("`"):
+        return "`{}`".format(replacement.replace("`", "``"))
+    if original.startswith("[") and "]" not in replacement:
+        return f"[{replacement}]"
+    if _valid_bare_identifier(replacement):
+        return replacement
+    return '"{}"'.format(replacement.replace('"', '""'))
+
+
+def rewrite_check_expression(expression: str, rename: dict[str, str]) -> str:
+    """Rewrite column identifiers in a CHECK expression, preserving trivia."""
+    if not rename:
+        return expression
+    tokens = _lex(expression)
+    meaningful = _meaningful(tokens)
+    replacements = {_ascii_fold(key): value for key, value in rename.items()}
+    edits: list[tuple[int, int, str]] = []
+    for index, token in enumerate(meaningful):
+        if not _is_identifier_token(meaningful, index):
+            continue
+        replacement = replacements.get(_ascii_fold(_unquote(token.text)))
+        if replacement is not None:
+            edits.append(
+                (token.start, token.end, _quote_replacement(token.text, replacement))
+            )
+    for start, end, replacement in reversed(edits):
+        expression = expression[:start] + replacement + expression[end:]
+    return expression

sqlite_utils/db.py

@@ -27,6 +27,16 @@ from typing_extensions import Self
 
 from sqlite_utils.plugins import ensure_plugins_loaded, pm
 
+from .create_table_parser import (
+    Check,
+    ColumnComments,
+    ParseError,
+    check_references_identifier,
+    parse_checks,
+    parse_column_comments,
+    rewrite_check_expression,
+    sql_ends_in_line_comment,
+)
 from .utils import (
     OperationalError,
     chunks,
@@ -85,6 +95,28 @@ def quote_identifier(identifier: str) -> str:
     return '"{}"'.format(identifier.replace('"', '""'))
 
 
+def _check_constraint_sql(check: Check) -> str:
+    prefix = f"CONSTRAINT {quote_identifier(check.name)} " if check.name else ""
+    newline = "\n" if sql_ends_in_line_comment(check.check) else ""
+    return f"{prefix}CHECK ({check.check}{newline})"
+
+
+def _column_definition_with_comments(
+    definition: str, comments: ColumnComments | None
+) -> str:
+    if comments is None:
+        return definition
+    before = textwrap.dedent(comments.before).strip()
+    after = textwrap.dedent(comments.after).strip()
+    if before:
+        definition = f"{textwrap.indent(before, '   ')}\n{definition}"
+    if after:
+        definition = f"{definition} {after}"
+        if sql_ends_in_line_comment(after):
+            definition += "\n"
+    return definition
+
+
 _IDENTIFIER_CASEFOLD = str.maketrans(
     "ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz"
 )
@@ -1379,6 +1411,8 @@ class Database:
         extracts: dict[str, str] | list[str] | None = None,
         if_not_exists: bool = False,
         strict: bool = False,
+        _checks: Iterable[Check] | None = None,
+        _column_comments: Mapping[str, ColumnComments] | None = None,
     ) -> str:
         """
         Returns the SQL ``CREATE TABLE`` statement for creating the specified table.
@@ -1424,6 +1458,23 @@ class Database:
         defaults = {resolve_casing(n, columns): v for n, v in (defaults or {}).items()}
         if column_order is not None:
             column_order = [resolve_casing(c, columns) for c in column_order]
+        column_comments = {
+            resolve_casing(name, columns): comments
+            for name, comments in (_column_comments or {}).items()
+        }
+        checks = list(_checks or ())
+        checks_by_column: dict[str, list[Check]] = {}
+        table_checks: list[Check] = []
+        for check in checks:
+            if check.column:
+                column = resolve_casing(check.column, columns)
+                if column not in columns:
+                    raise AlterError(
+                        f"No such column for CHECK constraint: {check.column}"
+                    )
+                checks_by_column.setdefault(column, []).append(check)
+            else:
+                table_checks.append(check)
         if not columns:
             raise ValueError("Tables must have at least one column")
         if not all(n in columns for n in not_null):
@@ -1480,18 +1531,25 @@ class Database:
                 column_extras.append(
                     f"REFERENCES {quote_identifier(fk.other_table)}({quote_identifier(cast(str, fk.other_column))}){_fk_actions_sql(fk)}"
                 )
+            column_extras.extend(
+                _check_constraint_sql(check)
+                for check in checks_by_column.get(column_name, ())
+            )
             column_type_str = COLUMN_TYPE_MAPPING[column_type]
             # Special case for strict tables to map FLOAT to REAL
             # Refs https://github.com/simonw/sqlite-utils/issues/644
             if strict and column_type_str == "FLOAT":
                 column_type_str = "REAL"
+            column_definition = "   {} {column_type}{column_extras}".format(
+                quote_identifier(column_name),
+                column_type=column_type_str,
+                column_extras=(
+                    (" " + " ".join(column_extras)) if column_extras else ""
+                ),
+            )
             column_defs.append(
-                "   {} {column_type}{column_extras}".format(
+                _column_definition_with_comments(
-                    quote_identifier(column_name),
+                    column_definition, column_comments.get(column_name)
-                    column_type=column_type_str,
-                    column_extras=(
-                        (" " + " ".join(column_extras)) if column_extras else ""
-                    ),
                 )
             )
         extra_pk = ""
@@ -1519,6 +1577,9 @@ class Database:
                     actions=_fk_actions_sql(fk),
                 )
             )
+        column_defs.extend(
+            f"   {_check_constraint_sql(check)}" for check in table_checks
+        )
         columns_sql = ",\n".join(column_defs)
         sql = """CREATE TABLE {if_not_exists}{table} (
 {columns_sql}{extra_pk}
@@ -2196,6 +2257,27 @@ class Table(Queryable):
         "Does this table use ``rowid`` for its primary key (no other primary keys are specified)?"
         return not any(column for column in self.columns if column.is_pk)
 
+    @property
+    def checks(self) -> list[Check]:
+        "List of column-level and table-level CHECK constraints on this table."
+        if not self.exists() or self.virtual_table_using is not None:
+            return []
+        return parse_checks(self.schema)
+
+    @property
+    def column_checks(self) -> dict[str, list[Check]]:
+        "CHECK constraints grouped by the column on which they are defined."
+        checks: dict[str, list[Check]] = {}
+        for check in self.checks:
+            if check.column:
+                checks.setdefault(check.column, []).append(check)
+        return checks
+
+    @property
+    def table_checks(self) -> list[Check]:
+        "Table-level CHECK constraints on this table."
+        return [check for check in self.checks if not check.column]
+
     def get(self, pk_values: list | tuple | str | int) -> dict:
         """
         Return row (as dictionary) for the specified primary key.
@@ -2655,6 +2737,41 @@ class Table(Queryable):
         if column_order is not None:
             column_order = [resolve_casing(c, existing_columns) for c in column_order]
 
+        try:
+            existing_checks = self.checks
+            existing_column_comments = parse_column_comments(self.schema)
+        except ParseError as ex:
+            raise TransformError(
+                f"Could not parse table schema for table {self.name!r}: {ex}"
+            ) from ex
+        create_table_checks: list[Check] = []
+        for check in existing_checks:
+            owner = (
+                resolve_casing(check.column, existing_columns) if check.column else ""
+            )
+            # A column-level constraint disappears with the column that owns it.
+            if owner and owner in drop:
+                continue
+            for dropped_column in drop:
+                if check_references_identifier(check.check, dropped_column):
+                    raise TransformError(
+                        f"Cannot drop column {dropped_column!r}: it is used by "
+                        f"CHECK constraint {check.name or check.check!r}"
+                    )
+            create_table_checks.append(
+                Check(
+                    rewrite_check_expression(check.check, rename),
+                    name=check.name,
+                    column=rename.get(owner) or owner,
+                )
+            )
+
+        create_table_column_comments: dict[str, ColumnComments] = {}
+        for column, comments in existing_column_comments.items():
+            owner = resolve_casing(column, existing_columns)
+            if owner not in drop:
+                create_table_column_comments[rename.get(owner) or owner] = comments
+
         create_table_foreign_keys: list[ForeignKeyIndicator] = []
 
         if foreign_keys is not None:
@@ -2804,6 +2921,8 @@ class Table(Queryable):
                 foreign_keys=create_table_foreign_keys,
                 column_order=column_order,
                 strict=self.strict if strict is None else strict,
+                _checks=create_table_checks,
+                _column_comments=create_table_column_comments,
             ).strip()
         )
 

tests/test_create_table_parser.py

@@ -0,0 +1,164 @@
+import sqlite3
+
+import hypothesis.strategies as st
+import pytest
+from hypothesis import given
+
+from sqlite_utils.create_table_parser import (
+    Check,
+    ColumnComments,
+    ParseError,
+    parse_checks,
+    parse_column_comments,
+)
+
+
+def test_parse_column_and_table_checks():
+    sql = """
+        CREATE TABLE people (
+            age INTEGER CONSTRAINT positive CHECK (age > 0),
+            status TEXT CHECK(status IN ('active', 'inactive')),
+            CONSTRAINT adult CHECK(age >= 18)
+        )
+    """
+    assert parse_checks(sql) == [
+        Check("age > 0", name="positive", column="age"),
+        Check(
+            "status IN ('active', 'inactive')",
+            column="status",
+            options=["active", "inactive"],
+        ),
+        Check("age >= 18", name="adult"),
+    ]
+    checks = parse_checks(sql)
+    assert checks[0].sql == "CONSTRAINT positive CHECK (age > 0)"
+    assert sql[checks[0].start : checks[0].end] == checks[0].sql
+    assert checks[1].sql == "CHECK(status IN ('active', 'inactive'))"
+    assert sql[checks[2].start : checks[2].end] == checks[2].sql
+
+
+def test_comments_are_trivia_not_constraints():
+    sql = """
+        CREATE /* fake CHECK (nope), ( */ TABLE t (
+            a INTEGER /* CHECK (a < 0), phantom */,
+            b INTEGER CHECK /* between keyword and expression */ (b > 0),
+            /* CHECK (also_fake) */ CONSTRAINT upper CHECK(b < 10)
+        )
+    """
+    sqlite3.connect(":memory:").execute(sql)
+    assert parse_checks(sql) == [
+        Check("b > 0", column="b"),
+        Check("b < 10", name="upper"),
+    ]
+
+
+def test_parse_comments_owned_by_columns():
+    sql = """
+        CREATE TABLE t (
+            -- Before id
+            id /* Between name and type */ INTEGER /* After id */,
+            /* Between column definitions */
+            value TEXT CHECK(value != '') /* After value */,
+            /* Before a table constraint, not a column */
+            CHECK(value != 'forbidden')
+        )
+    """
+    assert parse_column_comments(sql) == {
+        "id": ColumnComments(before="-- Before id", after="/* After id */"),
+        "value": ColumnComments(
+            before="/* Between column definitions */",
+            after="/* After value */",
+        ),
+    }
+
+
+@pytest.mark.parametrize(
+    "expression,expected",
+    [
+        ("value IN ('one', 'two')", ["one", "two"]),
+        ("((value IN ('one', 'two')))", ["one", "two"]),
+        ("value NOT IN ('one', 'two')", None),
+        ("value IN ('one', 'two') OR enabled", None),
+        ("other IN ('one', 'two')", None),
+        ("value IN (lower('one'), 'two')", None),
+        ('value IN ("other")', None),
+    ],
+)
+def test_options_only_for_exact_literal_in_check(expression, expected):
+    sql = f"CREATE TABLE t(value TEXT CHECK({expression}), enabled INTEGER, other TEXT)"
+    sqlite3.connect(":memory:").execute(sql)
+    assert parse_checks(sql)[0].options == expected
+
+
+@pytest.mark.parametrize("column", ["💩x", "e\u0301"])
+def test_unquoted_unicode_identifiers(column):
+    sql = f"CREATE TABLE t({column} INTEGER CHECK({column} > 0))"
+    sqlite3.connect(":memory:").execute(sql)
+    assert parse_checks(sql) == [Check(f"{column} > 0", column=column)]
+
+
+@pytest.mark.parametrize(
+    "sql",
+    [
+        "SELECT CHECK(x > 0)",
+        "CREATE TABLE t(x INTEGER CHECK(x > 0)",
+        "CREATE TABLE t(x TEXT CHECK(x != 'unterminated))",
+        "CREATE TABLE t(x INTEGER /* unterminated)",
+    ],
+)
+def test_invalid_sql_raises_parse_error(sql):
+    with pytest.raises(ParseError):
+        parse_checks(sql)
+
+
+def test_virtual_table_has_no_checks():
+    assert (
+        parse_checks("CREATE /* comment */ VIRTUAL TABLE search USING fts5(text)") == []
+    )
+
+
+comment_or_space = st.sampled_from(
+    [
+        " ",
+        "\n  ",
+        "/* comment with , ( ) and CHECK(fake) */",
+        "-- comment with , ( ) and CHECK(fake)\n",
+    ]
+)
+
+
+@given(gaps=st.lists(comment_or_space, min_size=5, max_size=5))
+def test_comments_and_whitespace_can_separate_check_tokens(gaps):
+    sql = (
+        f"CREATE{gaps[0]}TABLE{gaps[1]}t{gaps[2]}("
+        f"value INTEGER CHECK{gaps[3]}(value{gaps[4]}> 0))"
+    )
+    connection = sqlite3.connect(":memory:")
+    connection.execute(sql)
+    stored_sql = connection.execute(
+        "select sql from sqlite_master where name = 't'"
+    ).fetchone()[0]
+    assert parse_checks(stored_sql) == [Check(f"value{gaps[4]}> 0", column="value")]
+
+
+safe_string_text = st.text(
+    alphabet=st.characters(
+        blacklist_categories=("Cc", "Cs"),
+        blacklist_characters=("'",),
+    ),
+    max_size=40,
+)
+
+
+@given(value=safe_string_text)
+def test_check_like_text_inside_strings_is_opaque(value):
+    sql = f"CREATE TABLE t(value TEXT CHECK(value != '{value}'))"
+    connection = sqlite3.connect(":memory:")
+    connection.execute(sql)
+    stored_sql = connection.execute(
+        "select sql from sqlite_master where name = 't'"
+    ).fetchone()[0]
+    checks = parse_checks(stored_sql)
+    assert len(checks) == 1
+    assert checks[0].column == "value"
+    assert checks[0].check == f"value != '{value}'"

tests/test_introspect.py

@@ -1,6 +1,6 @@
 import pytest
 
-from sqlite_utils.db import Database, Index, View, XIndex, XIndexColumn
+from sqlite_utils.db import Check, Database, Index, View, XIndex, XIndexColumn
 
 
 def _check_supports_strict():
@@ -177,6 +177,31 @@ def test_pks(fresh_db, pk, expected):
     assert expected == fresh_db["foo"].pks
 
 
+def test_checks(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE scores (
+            score INTEGER CONSTRAINT positive CHECK(score > 0),
+            maximum INTEGER,
+            CONSTRAINT within_maximum CHECK(score <= maximum)
+        )
+    """)
+    scores = fresh_db["scores"]
+    expected_column = Check("score > 0", name="positive", column="score")
+    expected_table = Check("score <= maximum", name="within_maximum")
+    assert scores.checks == [expected_column, expected_table]
+    assert scores.column_checks == {"score": [expected_column]}
+    assert scores.table_checks == [expected_table]
+    assert scores.checks[0].sql == "CONSTRAINT positive CHECK(score > 0)"
+
+
+def test_checks_nonexistent_and_virtual_tables(fresh_db):
+    assert fresh_db["does_not_exist"].checks == []
+    fresh_db["searchable"].insert({"text": "hello"}).enable_fts(
+        ["text"], fts_version="FTS5"
+    )
+    assert fresh_db["searchable_fts"].checks == []
+
+
 def test_triggers_and_triggers_dict(fresh_db):
     assert [] == fresh_db.triggers
     authors = fresh_db["authors"]

tests/test_mutator_transactions.py

@@ -0,0 +1,154 @@
+import pytest
+
+from sqlite_utils import Database
+from sqlite_utils.utils import sqlite3
+
+BASELINE_ROWS = [(1, "one"), (2, "two")]
+
+
+def insert(table):
+    table.insert({"id": 3, "value": "three"}, pk="id")
+
+
+def insert_all(table):
+    table.insert_all(
+        [
+            {"id": 3, "value": "three"},
+            {"id": 4, "value": "four"},
+        ],
+        pk="id",
+        batch_size=1,
+    )
+
+
+def upsert(table):
+    table.upsert({"id": 2, "value": "TWO"}, pk="id")
+
+
+def upsert_all(table):
+    table.upsert_all(
+        [
+            {"id": 2, "value": "TWO"},
+            {"id": 3, "value": "three"},
+        ],
+        pk="id",
+        batch_size=1,
+    )
+
+
+def update(table):
+    table.update(2, {"value": "TWO"})
+
+
+def delete(table):
+    table.delete(2)
+
+
+def delete_where(table):
+    table.delete_where("id > ?", [1])
+
+
+MUTATOR_CASES = (
+    pytest.param(
+        insert,
+        [(1, "one"), (2, "two"), (3, "three")],
+        id="insert",
+    ),
+    pytest.param(
+        insert_all,
+        [(1, "one"), (2, "two"), (3, "three"), (4, "four")],
+        id="insert_all",
+    ),
+    pytest.param(
+        upsert,
+        [(1, "one"), (2, "TWO")],
+        id="upsert",
+    ),
+    pytest.param(
+        upsert_all,
+        [(1, "one"), (2, "TWO"), (3, "three")],
+        id="upsert_all",
+    ),
+    pytest.param(
+        update,
+        [(1, "one"), (2, "TWO")],
+        id="update",
+    ),
+    pytest.param(delete, [(1, "one")], id="delete"),
+    pytest.param(delete_where, [(1, "one")], id="delete_where"),
+)
+
+
+class RollbackTest(Exception):
+    pass
+
+
+def seed_database(path):
+    conn = sqlite3.connect(str(path))
+    try:
+        conn.execute("create table items (id integer primary key, value text)")
+        conn.executemany("insert into items values (?, ?)", BASELINE_ROWS)
+        conn.commit()
+    finally:
+        conn.close()
+    return Database(path)
+
+
+def current_rows(db):
+    return db.conn.execute("select id, value from items order by id").fetchall()
+
+
+def persisted_rows(path):
+    conn = sqlite3.connect(str(path))
+    try:
+        return conn.execute("select id, value from items order by id").fetchall()
+    finally:
+        conn.close()
+
+
+@pytest.mark.parametrize("mutate,expected_rows", MUTATOR_CASES)
+def test_mutator_commits_by_default(tmp_path, mutate, expected_rows):
+    path = tmp_path / "default.db"
+    db = seed_database(path)
+
+    assert not db.conn.in_transaction
+    mutate(db["items"])
+    assert current_rows(db) == expected_rows
+    assert not db.conn.in_transaction
+
+    db.close()
+    assert persisted_rows(path) == expected_rows
+
+
+@pytest.mark.parametrize("mutate,expected_rows", MUTATOR_CASES)
+def test_mutator_commits_with_outer_atomic(tmp_path, mutate, expected_rows):
+    path = tmp_path / "atomic.db"
+    db = seed_database(path)
+
+    with db.atomic():
+        assert db.conn.in_transaction
+        mutate(db["items"])
+        assert current_rows(db) == expected_rows
+        assert db.conn.in_transaction
+
+    assert current_rows(db) == expected_rows
+    assert not db.conn.in_transaction
+    db.close()
+    assert persisted_rows(path) == expected_rows
+
+
+@pytest.mark.parametrize("mutate,expected_rows", MUTATOR_CASES)
+def test_mutator_rolls_back_outer_atomic(tmp_path, mutate, expected_rows):
+    path = tmp_path / "rollback.db"
+    db = seed_database(path)
+
+    with pytest.raises(RollbackTest), db.atomic():
+        mutate(db["items"])
+        assert current_rows(db) == expected_rows
+        assert db.conn.in_transaction
+        raise RollbackTest
+
+    assert current_rows(db) == BASELINE_ROWS
+    assert not db.conn.in_transaction
+    db.close()
+    assert persisted_rows(path) == BASELINE_ROWS

tests/test_transform.py

@@ -2,7 +2,7 @@ import sqlite3
 
 import pytest
 
-from sqlite_utils.db import ForeignKey, TransactionError, TransformError
+from sqlite_utils.db import Check, ForeignKey, TransactionError, TransformError
 from sqlite_utils.utils import OperationalError
 
 
@@ -1065,3 +1065,186 @@ def test_transform_restores_legacy_alter_table_setting(fresh_db):
     assert sqls[-1] == "PRAGMA legacy_alter_table=ON;"
     dogs.transform(types={"name": str})
     assert fresh_db.execute("PRAGMA legacy_alter_table").fetchone()[0] == 1
+
+
+def test_transform_preserves_check_constraints(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE scores (
+            id INTEGER PRIMARY KEY,
+            score INTEGER CONSTRAINT valid_score CHECK(score BETWEEN 0 AND 100),
+            CONSTRAINT nonzero_id CHECK(id != 0)
+        )
+    """)
+    scores = fresh_db["scores"]
+    scores.insert({"id": 1, "score": 50})
+    scores.transform()
+    assert scores.checks == [
+        Check("score BETWEEN 0 AND 100", name="valid_score", column="score"),
+        Check("id != 0", name="nonzero_id"),
+    ]
+    with pytest.raises(sqlite3.IntegrityError, match="CHECK constraint failed"):
+        scores.insert({"id": 2, "score": 101})
+
+
+def test_transform_preserves_check_ending_in_line_comment(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE inventory (
+            quantity INTEGER,
+            CHECK (
+                quantity >= 0 -- Quantity cannot be negative
+            )
+        )
+    """)
+    inventory = fresh_db["inventory"]
+    inventory.transform(types={"quantity": float})
+    assert inventory.checks == [Check("quantity >= 0 -- Quantity cannot be negative")]
+    with pytest.raises(sqlite3.IntegrityError, match="CHECK constraint failed"):
+        inventory.insert({"quantity": -1})
+
+
+def test_transform_preserves_comments_owned_by_columns(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE people (
+            -- Primary identifier
+            id INTEGER PRIMARY KEY /* IDs are stable */,
+            /* Displayed to users */
+            name TEXT /* May contain spaces */,
+            -- Age in years
+            age INTEGER -- May be NULL
+        )
+    """)
+    people = fresh_db["people"]
+    people.insert({"id": 1, "name": "Cleo", "age": 5})
+    people.transform(
+        rename={"name": "display_name"},
+        types={"age": float},
+        column_order=("age", "id", "name"),
+    )
+    assert people.get(1) == {"age": 5.0, "id": 1, "display_name": "Cleo"}
+    schema = people.schema
+    assert schema.index("-- Age in years") < schema.index('"age" REAL')
+    assert schema.index('"age" REAL') < schema.index("-- May be NULL")
+    assert schema.index("-- Primary identifier") < schema.index('"id" INTEGER')
+    assert schema.index('"id" INTEGER') < schema.index("/* IDs are stable */")
+    assert schema.index("/* Displayed to users */") < schema.index(
+        '"display_name" TEXT'
+    )
+    assert schema.index('"display_name" TEXT') < schema.index(
+        "/* May contain spaces */"
+    )
+
+
+def test_transform_drops_comments_owned_by_dropped_column(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE t (
+            /* Keep this explanation */
+            id INTEGER,
+            /* Drop this explanation */
+            obsolete TEXT /* Drop this too */
+        )
+    """)
+    fresh_db["t"].transform(drop={"obsolete"})
+    schema = fresh_db["t"].schema
+    assert "Keep this explanation" in schema
+    assert "Drop this explanation" not in schema
+    assert "Drop this too" not in schema
+
+
+def test_transform_renames_columns_inside_check_constraints(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE inventory (
+            quantity INTEGER CONSTRAINT positive
+                CHECK(quantity > 0 AND 'quantity' != ''),
+            maximum INTEGER,
+            CONSTRAINT within_maximum CHECK(quantity <= maximum)
+        )
+    """)
+    inventory = fresh_db["inventory"]
+    inventory.insert({"quantity": 2, "maximum": 3})
+    inventory.transform(rename={"quantity": "amount"})
+    assert inventory.checks == [
+        Check(
+            "amount > 0 AND 'quantity' != ''",
+            name="positive",
+            column="amount",
+        ),
+        Check("amount <= maximum", name="within_maximum"),
+    ]
+    with pytest.raises(sqlite3.IntegrityError, match="CHECK constraint failed"):
+        inventory.insert({"amount": 4, "maximum": 3})
+
+
+def test_transform_check_rewrite_preserves_functions_and_quotes(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE items (
+            length TEXT,
+            "old name" TEXT,
+            CHECK(length("old name") > 0 AND length != '')
+        )
+    """)
+    items = fresh_db["items"]
+    items.insert({"length": "label", "old name": "hello"})
+    items.transform(rename={"length": "description", "old name": "new name"})
+    assert items.checks == [Check("length(\"new name\") > 0 AND description != ''")]
+
+
+def test_transform_check_rewrite_quotes_keyword_column(fresh_db):
+    fresh_db.execute("CREATE TABLE t(old_name TEXT CHECK(old_name != ''))")
+    fresh_db["t"].insert({"old_name": "value"})
+    fresh_db["t"].transform(rename={"old_name": "select"})
+    assert fresh_db["t"].checks == [Check("\"select\" != ''", column="select")]
+
+
+def test_transform_check_rewrite_does_not_rename_collations_or_cast_types(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE t (
+            nocase TEXT,
+            kind TEXT,
+            other TEXT,
+            CHECK(
+                other COLLATE nocase != ''
+                AND CAST(other AS kind) != ''
+                AND nocase != ''
+                AND kind != ''
+            )
+        )
+    """)
+    fresh_db["t"].insert({"nocase": "n", "kind": "k", "other": "o"})
+    fresh_db["t"].transform(rename={"nocase": "label", "kind": "category"})
+    check = fresh_db["t"].checks[0].check
+    assert "COLLATE nocase" in check
+    assert "AS kind" in check
+    assert "AND label != ''" in check
+    assert "AND category != ''" in check
+
+
+def test_transform_drops_check_owned_by_dropped_column(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE t (
+            id INTEGER,
+            obsolete INTEGER CHECK(obsolete > 0),
+            CHECK(id > 0)
+        )
+    """)
+    fresh_db["t"].insert({"id": 1, "obsolete": 2})
+    fresh_db["t"].transform(drop={"obsolete"})
+    assert fresh_db["t"].checks == [Check("id > 0")]
+
+
+def test_transform_refuses_to_drop_column_used_by_remaining_check(fresh_db):
+    fresh_db.execute("""
+        CREATE TABLE ranges (
+            minimum INTEGER,
+            maximum INTEGER,
+            CHECK(minimum <= maximum)
+        )
+    """)
+    ranges = fresh_db["ranges"]
+    ranges.insert({"minimum": 1, "maximum": 2})
+    schema_before = ranges.schema
+    with pytest.raises(
+        TransformError,
+        match="Cannot drop column 'maximum'.*CHECK constraint",
+    ):
+        ranges.transform(drop={"maximum"})
+    assert ranges.schema == schema_before