codespell should check sqlite_utils as well

It did in CI but did not in the Justfile

.gitignore

@@ -15,6 +15,7 @@ venv
 .schema
 .vscode
 .hypothesis
+.claude/
 Pipfile
 Pipfile.lock
 uv.lock

Justfile

@@ -16,6 +16,7 @@
   uv run ty check sqlite_utils
   uv run cog --check README.md docs/*.rst
   uv run --group docs codespell docs/*.rst --ignore-words docs/codespell-ignore-words.txt
+  uv run --group docs codespell sqlite_utils --ignore-words docs/codespell-ignore-words.txt
 
 # Rebuild docs with cog
 @cog:

docs/changelog.rst

@@ -4,6 +4,13 @@
  Changelog
 ===========
 
+.. _v4_1_1:
+
+4.1.1 (2026-07-12)
+------------------
+
+- ``table.transform()`` now raises a ``TransactionError`` if called while a transaction is open with ``PRAGMA foreign_keys`` enabled and the table is referenced by foreign keys with destructive ``ON DELETE`` actions - ``CASCADE``, ``SET NULL`` or ``SET DEFAULT``. The pragma cannot be changed inside a transaction, so previously dropping the old table as part of the transform could fire those actions and silently delete or modify referencing rows. See :ref:`python_api_transform_foreign_keys_transactions` for details and workarounds. (:issue:`794`)
+- The :ref:`CLI <cli>` and :ref:`Python API <python_api>` documentation now cross-reference each other: CLI sections link to the equivalent Python API functionality and Python API sections link back to the corresponding CLI command. (:issue:`791`)
 .. _v4_1:
 
 4.1 (2026-07-11)

docs/cli-reference.rst

@@ -662,7 +662,7 @@ See :ref:`cli_convert`.
       Convert a string like a,b,c into a JSON array ["a", "b", "c"]
 
       r.parsedate(value: 'str', dayfirst: 'bool' = False, yearfirst: 'bool' = False,
-      errors: 'Optional[object]' = None) -> 'Optional[str]'
+      errors: 'object | None' = None) -> 'str | None'
 
       Parse a date and convert it to ISO date format: yyyy-mm-dd
       - dayfirst=True: treat xx as the day in xx/yy/zz
@@ -671,7 +671,7 @@ See :ref:`cli_convert`.
       - errors=r.SET_NULL to set values that cannot be parsed to null
 
       r.parsedatetime(value: 'str', dayfirst: 'bool' = False, yearfirst: 'bool' =
-      False, errors: 'Optional[object]' = None) -> 'Optional[str]'
+      False, errors: 'object | None' = None) -> 'str | None'
 
       Parse a datetime and convert it to ISO datetime format: yyyy-mm-ddTHH:MM:SS
       - dayfirst=True: treat xx as the day in xx/yy/zz

docs/conf.py

@@ -1,10 +1,7 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-
 import inspect
-from pathlib import Path
-from subprocess import Popen, PIPE, check_output
 import sys
+from pathlib import Path
+from subprocess import PIPE, CalledProcessError, Popen, check_output
 
 # This file is execfile()d with the current directory set to its
 # containing dir.
@@ -50,7 +47,7 @@ extlinks = {
 def _linkcode_git_ref():
     try:
         return check_output(["git", "rev-parse", "HEAD"]).decode("utf8").strip()
-    except Exception:
+    except (CalledProcessError, OSError):
         return "main"
 
 
@@ -79,7 +76,7 @@ def linkcode_resolve(domain, info):
         obj = inspect.unwrap(obj)
         source_file = inspect.getsourcefile(obj)
         _, line_number = inspect.getsourcelines(obj)
-    except Exception:
+    except (OSError, TypeError, ValueError):
         return None
 
     if source_file is None:

docs/python-api.rst

@@ -434,9 +434,10 @@ The library will never commit a transaction you opened. If you call write method
 
 Prefer ``db.atomic()`` or ``db.begin()``, ``db.commit()`` and ``db.rollback()`` over mixing sqlite-utils transaction methods with calls to ``db.conn.commit()``, ``db.conn.rollback()`` or raw transaction-control SQL. Mixing the two layers makes it much harder to tell which layer owns the current transaction.
 
-Two related safeguards to be aware of:
+Some related safeguards to be aware of:
 
 - ``db.enable_wal()`` and ``db.disable_wal()`` raise a ``sqlite_utils.db.TransactionError`` if called while a transaction is open, because changing the journal mode would commit it as a side effect.
+- ``table.transform()`` raises a ``sqlite_utils.db.TransactionError`` if called while a transaction is open with ``PRAGMA foreign_keys`` enabled and the table is referenced by foreign keys with destructive ``ON DELETE`` actions, because the pragma cannot be turned off mid-transaction to protect those referencing rows - see :ref:`python_api_transform_foreign_keys_transactions`.
 - Closing the database - explicitly with ``db.close()``, or by exiting a ``with Database(...) as db:`` block - rolls back any transaction that is still open, see :ref:`python_api_close`.
 
 .. _python_api_transactions_modes:
@@ -1996,6 +1997,36 @@ If you want to do something more advanced, you can call the ``table.transform_sq
 
 This method will return a list of SQL statements that should be executed to implement the change. You can then make modifications to that SQL - or add additional SQL statements - before executing it yourself.
 
+.. _python_api_transform_foreign_keys_transactions:
+
+Foreign keys and transactions
+-----------------------------
+
+Because ``.transform()`` drops the old table, running it with ``PRAGMA foreign_keys`` enabled could fire ``ON DELETE`` actions on any tables that reference it - an inbound ``ON DELETE CASCADE`` foreign key would silently delete those referencing rows. To prevent this, ``.transform()`` turns ``PRAGMA foreign_keys`` off for the duration of the operation and restores it afterwards, running ``PRAGMA foreign_key_check`` before committing.
+
+``PRAGMA foreign_keys`` cannot be changed inside a transaction, so this protection is impossible if you call ``.transform()`` while a transaction is already open - for example inside a ``with db.atomic():`` block or after ``db.begin()``. If ``PRAGMA foreign_keys`` is on and another table references the table being transformed with a destructive ``ON DELETE`` action - ``CASCADE``, ``SET NULL`` or ``SET DEFAULT`` - the method will refuse to run and raise a ``sqlite_utils.db.TransactionError``:
+
+.. code-block:: python
+
+    from sqlite_utils.db import TransactionError
+
+    try:
+        with db.atomic():
+            db["authors"].transform(types={"id": str})
+    except TransactionError as ex:
+        print("Could not transform in transaction:", ex)
+
+To transform such a table either call ``.transform()`` outside of the transaction, or execute ``PRAGMA foreign_keys = off`` before opening it:
+
+.. code-block:: python
+
+    db.execute("PRAGMA foreign_keys = off")
+    with db.atomic():
+        db["authors"].transform(types={"id": str})
+    db.execute("PRAGMA foreign_keys = on")
+
+Tables referenced by foreign keys without a destructive action (the default ``NO ACTION``, or ``RESTRICT``) can still be transformed inside a transaction - sqlite-utils uses ``PRAGMA defer_foreign_keys`` to postpone the foreign key checks until the transaction commits.
+
 .. _python_api_extract:
 
 Extracting columns into a separate table

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sqlite-utils"
-version = "4.1"
+version = "4.1.1"
 description = "CLI tool and Python library for manipulating SQLite databases"
 readme = { file = "README.md", content-type = "text/markdown" }
 authors = [
@@ -79,7 +79,14 @@ build-backend = "setuptools.build_meta"
 max-line-length = 160
 # Black compatibility, E203 whitespace before ':':
 extend-ignore = ["E203"]
-extend-exclude = [".venv", "build", "dist", "docs", "sqlite_utils.egg-info"]
+extend-exclude = [
+    ".venv",
+    ".claude",
+    "build",
+    "dist",
+    "docs",
+    "sqlite_utils.egg-info",
+]
 
 [tool.setuptools.package-data]
 sqlite_utils = ["py.typed"]

sqlite_utils/__init__.py

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

sqlite_utils/cli.py

@@ -1,17 +1,30 @@
 import base64
+import csv as csv_std
 import difflib
-from typing import Any
-import click
-from click_default_group import DefaultGroup
-from datetime import datetime, timezone
 import hashlib
+import inspect
+import io
+import itertools
+import json
+import os
 import pathlib
+import pdb  # noqa: T100
+import sys
+import textwrap
+from datetime import datetime, timezone
 from runpy import run_module
+from typing import Any
+
+import click
+import tabulate
+from click_default_group import DefaultGroup
+
 import sqlite_utils
+from sqlite_utils import recipes
 from sqlite_utils.db import (
+    DEFAULT,
     AlterError,
     BadMultiValues,
-    DEFAULT,
     DescIndex,
     InvalidColumns,
     NoTable,
@@ -19,36 +32,28 @@ from sqlite_utils.db import (
     PrimaryKeyRequired,
     quote_identifier,
 )
-from sqlite_utils.plugins import ensure_plugins_loaded, pm, get_plugins
+from sqlite_utils.plugins import ensure_plugins_loaded, get_plugins, pm
 from sqlite_utils.utils import maximize_csv_field_size_limit
-from sqlite_utils import recipes
+
-import textwrap
-import inspect
-import io
-import itertools
-import json
-import os
-import pdb
-import sys
-import csv as csv_std
-import tabulate
 from .utils import (
+    Format,
     OperationalError,
+    TypeTracker,
     _compile_code,
     chunks,
+    decode_base64_values,
     dedupe_keys,
     file_progress,
     find_spatialite,
-    flatten as _flatten,
-    sqlite3,
-    decode_base64_values,
     progressbar,
     rows_from_file,
-    Format,
+    sqlite3,
-    TypeTracker,
+)
+from .utils import (
+    flatten as _flatten,
 )
 
-CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
+CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]}
 
 
 def _register_db_for_cleanup(db):
@@ -67,7 +72,7 @@ def _close_databases(ctx):
     for db in ctx.meta.get("_databases_to_close", []):
         try:
             db.close()
-        except Exception:
+        except sqlite3.Error:
             pass
 
 
@@ -174,7 +179,6 @@ def functions_option(fn):
 @click.version_option()
 def cli():
     "Commands for interacting with a SQLite database"
-    pass
 
 
 @cli.command()
@@ -891,7 +895,7 @@ def enable_counts(path, tables, load_extension):
         # Check all tables exist
         bad_tables = [table for table in tables if not db[table].exists()]
         if bad_tables:
-            raise click.ClickException("Invalid tables: {}".format(bad_tables))
+            raise click.ClickException(f"Invalid tables: {bad_tables}")
         for table in tables:
             db.table(table).enable_counts()
 
@@ -1140,9 +1144,7 @@ def insert_upsert_implementation(
                 )
             ):
                 raise click.ClickException(
-                    "{}\n\nTry using --alter to add additional columns".format(
+                    f"{e.args[0]}\n\nTry using --alter to add additional columns"
-                        e.args[0]
-                    )
                 )
             # If we can find sql= and parameters= arguments, show those
             variables = _find_variables(e.__traceback__, ["sql", "parameters"])
@@ -1240,7 +1242,7 @@ def insert_upsert_implementation(
             reader = csv_std.reader(decoded, **csv_reader_args)  # type: ignore
             first_row = next(reader)
             if no_headers:
-                headers = ["untitled_{}".format(i + 1) for i in range(len(first_row))]
+                headers = [f"untitled_{i + 1}" for i in range(len(first_row))]
                 reader = itertools.chain([first_row], reader)
             else:
                 headers = first_row
@@ -1269,9 +1271,7 @@ def insert_upsert_implementation(
                         docs = [docs]
             except json.decoder.JSONDecodeError as ex:
                 raise click.ClickException(
-                    "Invalid JSON - use --csv for CSV or --tsv for TSV files\n\nJSON error: {}".format(
+                    f"Invalid JSON - use --csv for CSV or --tsv for TSV files\n\nJSON error: {ex}"
-                        ex
-                    )
                 )
             if flatten:
                 docs = (_flatten(doc) for doc in docs)
@@ -1290,7 +1290,7 @@ def insert_upsert_implementation(
                 docs = (fn(doc["line"]) for doc in docs)
             elif text:
                 # Special case: this is allowed to be an iterable
-                text_value = list(docs)[0]["text"]
+                text_value = next(iter(docs))["text"]
                 fn_return = fn(text_value)
                 if isinstance(fn_return, dict):
                     docs = [fn_return]
@@ -1774,17 +1774,14 @@ def create_table(
         ctype = columns.pop(0)
         if ctype.upper() not in VALID_COLUMN_TYPES:
             raise click.ClickException(
-                "column types must be one of {}".format(VALID_COLUMN_TYPES)
+                f"column types must be one of {VALID_COLUMN_TYPES}"
             )
         coltypes[name] = ctype.upper()
     # Does table already exist?
-    if table in db.table_names():
+    if table in db.table_names() and not ignore and not replace and not transform:
-        if not ignore and not replace and not transform:
+        raise click.ClickException(
-            raise click.ClickException(
+            f'Table "{table}" already exists. Use --replace to delete and replace it.'
-                'Table "{}" already exists. Use --replace to delete and replace it.'.format(
+        )
-                    table
-                )
-            )
     db.table(table).create(
         coltypes,
         pk=pks[0] if len(pks) == 1 else pks,
@@ -1819,7 +1816,7 @@ def duplicate(path, table, new_table, ignore, load_extension):
         db.table(table).duplicate(new_table)
     except NoTable:
         if not ignore:
-            raise click.ClickException('Table "{}" does not exist'.format(table))
+            raise click.ClickException(f'Table "{table}" does not exist')
 
 
 @cli.command(name="rename-table")
@@ -1843,9 +1840,7 @@ def rename_table(path, table, new_name, ignore, load_extension):
         db.rename_table(table, new_name)
     except sqlite3.OperationalError as ex:
         if not ignore:
-            raise click.ClickException(
+            raise click.ClickException(f'Table "{table}" could not be renamed. {ex!s}')
-                'Table "{}" could not be renamed. {}'.format(table, str(ex))
-            )
 
 
 @cli.command(name="drop-table")
@@ -1874,10 +1869,10 @@ def drop_table(path, table, ignore, load_extension):
         # A view exists with this name
         if not ignore:
             raise click.ClickException(
-                '"{}" is a view, not a table - use drop-view to drop it'.format(table)
+                f'"{table}" is a view, not a table - use drop-view to drop it'
             )
     except OperationalError:
-        raise click.ClickException('Table "{}" does not exist'.format(table))
+        raise click.ClickException(f'Table "{table}" does not exist')
 
 
 @cli.command(name="create-view")
@@ -1919,9 +1914,7 @@ def create_view(path, view, select, ignore, replace, load_extension):
             db.view(view).drop()
         else:
             raise click.ClickException(
-                'View "{}" already exists. Use --replace to delete and replace it.'.format(
+                f'View "{view}" already exists. Use --replace to delete and replace it.'
-                    view
-                )
             )
     db.create_view(view, select)
 
@@ -1953,9 +1946,9 @@ def drop_view(path, view, ignore, load_extension):
             return
         if view in db.table_names():
             raise click.ClickException(
-                '"{}" is a table, not a view - use drop-table to drop it'.format(view)
+                f'"{view}" is a table, not a view - use drop-table to drop it'
             )
-        raise click.ClickException('View "{}" does not exist'.format(view))
+        raise click.ClickException(f'View "{view}" does not exist')
 
 
 @cli.command()
@@ -2177,7 +2170,7 @@ def memory(
             file_path = pathlib.Path(path)
             stem = file_path.stem
             if stem_counts.get(stem):
-                file_table = "{}_{}".format(stem, stem_counts[stem])
+                file_table = f"{stem}_{stem_counts[stem]}"
             else:
                 file_table = stem
             stem_counts[stem] = stem_counts.get(stem, 1) + 1
@@ -2196,14 +2189,14 @@ def memory(
             if tracker is not None and db.table(file_table).exists():
                 db.table(file_table).transform(types=tracker.types)
             # Add convenient t / t1 / t2 views
-            view_names = ["t{}".format(i + 1)]
+            view_names = [f"t{i + 1}"]
             if i == 0:
                 view_names.append("t")
             for view_name in view_names:
                 if not db[view_name].exists():
                     db.create_view(
                         view_name,
-                        "select * from {}".format(quote_identifier(file_table)),
+                        f"select * from {quote_identifier(file_table)}",
                     )
         finally:
             if should_close_fp and fp:
@@ -2373,19 +2366,17 @@ def search(
     # Check table exists
     table_obj = db.table(dbtable)
     if not table_obj.exists():
-        raise click.ClickException("Table '{}' does not exist".format(dbtable))
+        raise click.ClickException(f"Table '{dbtable}' does not exist")
     if not table_obj.detect_fts():
         raise click.ClickException(
-            "Table '{}' is not configured for full-text search".format(dbtable)
+            f"Table '{dbtable}' is not configured for full-text search"
         )
     if column:
         # Check they all exist
         table_columns = table_obj.columns_dict
         for c in column:
             if c not in table_columns:
-                raise click.ClickException(
+                raise click.ClickException(f"Table '{dbtable}' has no column '{c}")
-                    "Table '{}' has no column '{}".format(dbtable, c)
-                )
     sql = table_obj.search_sql(columns=column, order_by=order, limit=limit)
     if show_sql:
         click.echo(sql)
@@ -2412,7 +2403,7 @@ def search(
     except click.ClickException as e:
         if "malformed MATCH expression" in str(e) or "unterminated string" in str(e):
             raise click.ClickException(
-                "{}\n\nTry running this again with the --quote option".format(str(e))
+                f"{e!s}\n\nTry running this again with the --quote option"
             )
         else:
             raise
@@ -2479,15 +2470,15 @@ def rows(
     columns = "*"
     if column:
         columns = ", ".join(quote_identifier(c) for c in column)
-    sql = "select {} from {}".format(columns, quote_identifier(dbtable))
+    sql = f"select {columns} from {quote_identifier(dbtable)}"
     if where:
         sql += " where " + where
     if order:
         sql += " order by " + order
     if limit:
-        sql += " limit {}".format(limit)
+        sql += f" limit {limit}"
     if offset:
-        sql += " offset {}".format(offset)
+        sql += f" offset {offset}"
     ctx.invoke(
         query,
         path=path,
@@ -2760,7 +2751,7 @@ def transform(
     for column, ctype in type:
         if ctype.upper() not in VALID_COLUMN_TYPES:
             raise click.ClickException(
-                "column types must be one of {}".format(VALID_COLUMN_TYPES)
+                f"column types must be one of {VALID_COLUMN_TYPES}"
             )
         types[column] = ctype.upper()
 
@@ -2858,12 +2849,12 @@ def extract(
     db = sqlite_utils.Database(path)
     _register_db_for_cleanup(db)
     _load_extensions(db, load_extension)
-    kwargs: dict[str, Any] = dict(
+    kwargs: dict[str, Any] = {
-        columns=columns,
+        "columns": columns,
-        table=other_table,
+        "table": other_table,
-        fk_column=fk_column,
+        "fk_column": fk_column,
-        rename=dict(rename),
+        "rename": dict(rename),
-    )
+    }
     try:
         db.table(table).extract(**kwargs)
     except (NoTable, InvalidColumns) as e:
@@ -2958,7 +2949,7 @@ def insert_files(
     with progressbar(paths_and_relative_paths, silent=silent) as bar:
 
         def to_insert():
-            for path, relative_path in bar:
+            for file_path, relative_path in bar:
                 row = {}
                 # content_text is special case as it considers 'encoding'
 
@@ -2970,19 +2961,21 @@ def insert_files(
                         raise UnicodeDecodeErrorForPath(e, resolved)
 
                 lookups = dict(FILE_COLUMNS, content_text=_content_text)
-                if path == "-":
+                if file_path == "-":
                     stdin_data = sys.stdin.buffer.read()
                     # We only support a subset of columns for this case
                     lookups = {
                         "name": lambda p: name or "-",
                         "path": lambda p: name or "-",
-                        "content": lambda p: stdin_data,
+                        "content": lambda p, data=stdin_data: data,
-                        "content_text": lambda p: stdin_data.decode(
+                        "content_text": lambda p, data=stdin_data: data.decode(
                             encoding or "utf-8"
                         ),
-                        "sha256": lambda p: hashlib.sha256(stdin_data).hexdigest(),
+                        "sha256": lambda p, data=stdin_data: hashlib.sha256(
-                        "md5": lambda p: hashlib.md5(stdin_data).hexdigest(),
+                            data
-                        "size": lambda p: len(stdin_data),
+                        ).hexdigest(),
+                        "md5": lambda p, data=stdin_data: hashlib.md5(data).hexdigest(),
+                        "size": lambda p, data=stdin_data: len(data),
                     }
                 for coldef in column:
                     if ":" in coldef:
@@ -2990,7 +2983,7 @@ def insert_files(
                     else:
                         colname, coltype = coldef, coldef
                     try:
-                        value = lookups[coltype](path)
+                        value = lookups[coltype](file_path)
                         row[colname] = value
                     except KeyError:
                         raise click.ClickException(
@@ -3018,7 +3011,7 @@ def insert_files(
         except UnicodeDecodeErrorForPath as e:
             raise click.ClickException(
                 UNICODE_ERROR.format(
-                    "Could not read file '{}' as text\n\n{}".format(e.path, e.exception)
+                    f"Could not read file '{e.path}' as text\n\n{e.exception}"
                 )
             )
 
@@ -3196,7 +3189,7 @@ def _generate_convert_help():
     for name in recipe_names:
         fn = getattr(recipes, name)
         doc = textwrap.dedent(fn.__doc__.rstrip()).replace("\b\n", "")
-        help += "\n\nr.{}{}\n\n\b{}".format(name, str(inspect.signature(fn)), doc)
+        help += f"\n\nr.{name}{inspect.signature(fn)!s}\n\n\b{doc}"
     help += "\n\n"
     help += textwrap.dedent("""
     You can use these recipes like so:
@@ -3299,7 +3292,7 @@ def convert(
         """.format(
             column=columns[0],
             table=table,
-            where=" where {}".format(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():
             click.echo(str(row[0]))
@@ -3319,7 +3312,7 @@ def convert(
             def wrapped_fn(value):
                 try:
                     return fn_(value)
-                except Exception as ex:
+                except Exception as ex:  # noqa: BLE001
                     print("\nException raised, dropping into pdb...:", ex)
                     pdb.post_mortem(ex.__traceback__)
                     sys.exit(1)
@@ -3339,9 +3332,7 @@ def convert(
             )
         except BadMultiValues as e:
             raise click.ClickException(
-                "When using --multi code must return a Python dictionary - returned: {}".format(
+                f"When using --multi code must return a Python dictionary - returned: {e.values!r}"
-                    repr(e.values)
-                )
             )
 
 
@@ -3459,7 +3450,7 @@ def create_spatial_index(db_path, table, column_name, load_extension):
 
 def _find_migration_files(migrations):
     if not migrations:
-        migrations = [pathlib.Path(".").resolve()]
+        migrations = [pathlib.Path.cwd()]
     files = set()
     for path_str in migrations:
         path = pathlib.Path(path_str)
@@ -3484,7 +3475,7 @@ def _load_migration_sets(files):
             "__file__": str(filepath),
             "__name__": "__sqlite_utils_migration__",
         }
-        exec(code, namespace)
+        exec(code, namespace)  # noqa: S102
         migration_sets.extend(
             obj for obj in namespace.values() if _compatible_migration_set(obj)
         )
@@ -3493,17 +3484,17 @@ def _load_migration_sets(files):
 
 def _display_migration_list(db, migration_sets):
     for migration_set in migration_sets:
-        click.echo("Migrations for: {}".format(migration_set.name))
+        click.echo(f"Migrations for: {migration_set.name}")
         click.echo()
         click.echo("  Applied:")
         for migration in migration_set.applied(db):
-            click.echo("    {} - {}".format(migration.name, migration.applied_at))
+            click.echo(f"    {migration.name} - {migration.applied_at}")
         click.echo()
         click.echo("  Pending:")
         output = False
         for migration in migration_set.pending(db):
             output = True
-            click.echo("    {}".format(migration.name))
+            click.echo(f"    {migration.name}")
         if not output:
             click.echo("    (none)")
         click.echo()
@@ -3583,7 +3574,7 @@ def migrate(db_path, migrations, stop_before, list_, verbose):
 
     prev_schema = db.schema
     if verbose:
-        click.echo("Migrating {}".format(db_path))
+        click.echo(f"Migrating {db_path}")
         click.echo("\nSchema before:\n")
         click.echo(textwrap.indent(prev_schema, "  ") or "  (empty)")
         click.echo()
@@ -3594,9 +3585,7 @@ def migrate(db_path, migrations, stop_before, list_, verbose):
             names = {m.name for m in migration_set.pending(db)}
             names.update(m.name for m in migration_set.applied(db))
             known_names.update(names)
-            known_names.update(
+            known_names.update(f"{migration_set.name}:{name}" for name in names)
-                "{}:{}".format(migration_set.name, name) for name in names
-            )
         unknown = [value for value in stop_before if value not in known_names]
         if unknown:
             raise click.ClickException(
@@ -3652,7 +3641,7 @@ def _render_common(title, values):
         return ""
     lines = [title]
     for value, count in values:
-        lines.append("    {}: {}".format(count, value))
+        lines.append(f"    {count}: {value}")
     return "\n".join(lines)
 
 
@@ -3722,7 +3711,7 @@ def maybe_json(value):
     if not isinstance(value, str):
         return value
     stripped = value.strip()
-    if not (stripped.startswith("{") or stripped.startswith("[")):
+    if not (stripped.startswith(("{", "["))):
         return value
     try:
         return json.loads(stripped)
@@ -3740,7 +3729,7 @@ def json_binary(value):
 def verify_is_dict(doc):
     if not isinstance(doc, dict):
         raise click.ClickException(
-            "Rows must all be dictionaries, got: {}".format(repr(doc)[:1000])
+            f"Rows must all be dictionaries, got: {repr(doc)[:1000]}"
         )
     return doc
 
@@ -3768,14 +3757,14 @@ def _register_functions(db, functions):
         try:
             functions = pathlib.Path(functions).read_text()
         except FileNotFoundError:
-            raise click.ClickException("File not found: {}".format(functions))
+            raise click.ClickException(f"File not found: {functions}")
 
     sqlite3.enable_callback_tracebacks(True)
     globals = {}
     try:
-        exec(functions, globals)
+        exec(functions, globals)  # noqa: S102
     except SyntaxError as ex:
-        raise click.ClickException("Error in functions definition: {}".format(ex))
+        raise click.ClickException(f"Error in functions definition: {ex}")
     # Register all callables in the locals dict:
     for name, value in globals.items():
         if callable(value) and not name.startswith("_"):
@@ -3796,12 +3785,12 @@ def _rows_from_code(code):
         try:
             code = pathlib.Path(code).read_text()
         except FileNotFoundError:
-            raise click.ClickException("File not found: {}".format(code))
+            raise click.ClickException(f"File not found: {code}")
     namespace = {}
     try:
-        exec(code, namespace)
+        exec(code, namespace)  # noqa: S102
     except SyntaxError as ex:
-        raise click.ClickException("Error in --code: {}".format(ex))
+        raise click.ClickException(f"Error in --code: {ex}")
     rows = namespace.get("rows")
     if callable(rows):
         rows = rows()

sqlite_utils/hookspecs.py

@@ -1,8 +1,7 @@
 import sqlite3
 
 import click
-from pluggy import HookimplMarker
+from pluggy import HookimplMarker, HookspecMarker
-from pluggy import HookspecMarker
 
 hookspec = HookspecMarker("sqlite_utils")
 hookimpl = HookimplMarker("sqlite_utils")

sqlite_utils/migrations.py

@@ -1,19 +1,28 @@
-from collections.abc import Iterable
-from dataclasses import dataclass
 import datetime
-from typing import Callable, cast, TYPE_CHECKING
+from collections.abc import Callable, Iterable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Protocol, TypeVar, cast
 
 if TYPE_CHECKING:
     from sqlite_utils.db import Database, Table
 
 
+class _MigrationFunction(Protocol):
+    __name__: str
+
+    def __call__(self, db: "Database", /) -> None: ...
+
+
+_MigrationFunctionT = TypeVar("_MigrationFunctionT", bound=_MigrationFunction)
+
+
 class Migrations:
     migrations_table = "_sqlite_migrations"
 
     @dataclass
     class _Migration:
         name: str
-        fn: Callable
+        fn: _MigrationFunction
         transactional: bool = True
 
     @dataclass
@@ -32,7 +41,7 @@ class Migrations:
 
     def __call__(
         self, *, name: str | None = None, transactional: bool = True
-    ) -> Callable:
+    ) -> Callable[[_MigrationFunctionT], _MigrationFunctionT]:
         """
         :param name: The name to use for this migration - if not provided,
           the name of the function will be used.
@@ -43,13 +52,11 @@ class Migrations:
           example those that execute ``VACUUM``.
         """
 
-        def inner(func: Callable) -> Callable:
+        def inner(func: _MigrationFunctionT) -> _MigrationFunctionT:
-            migration_name = name or getattr(func, "__name__")
+            migration_name = name or func.__name__
             if any(m.name == migration_name for m in self._migrations):
                 raise ValueError(
-                    "Migration '{}' is already registered in set '{}'".format(
+                    f"Migration '{migration_name}' is already registered in set '{self.name}'"
-                        migration_name, self.name
-                    )
                 )
             self._migrations.append(
                 self._Migration(migration_name, func, transactional)

sqlite_utils/plugins.py

@@ -1,7 +1,7 @@
-from typing import Dict, List, Union
+import sys
 
 import pluggy
-import sys
+
 from . import hookspecs
 
 pm: pluggy.PluginManager = pluggy.PluginManager("sqlite_utils")
@@ -17,13 +17,13 @@ def ensure_plugins_loaded() -> None:
     _plugins_loaded = True
 
 
-def get_plugins() -> List[Dict[str, Union[str, List[str]]]]:
+def get_plugins() -> list[dict[str, str | list[str]]]:
     ensure_plugins_loaded()
-    plugins: List[Dict[str, Union[str, List[str]]]] = []
+    plugins: list[dict[str, str | list[str]]] = []
     plugin_to_distinfo = dict(pm.list_plugin_distinfo())
     for plugin in pm.get_plugins():
         hookcallers = pm.get_hookcallers(plugin) or []
-        plugin_info: Dict[str, Union[str, List[str]]] = {
+        plugin_info: dict[str, str | list[str]] = {
             "name": plugin.__name__,
             "hooks": [h.name for h in hookcallers],
         }

sqlite_utils/recipes.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
-from typing import Callable, Optional
+import json
+from collections.abc import Callable
 
 from dateutil import parser
-import json
 
 IGNORE: object = object()
 SET_NULL: object = object()
@@ -13,8 +13,8 @@ def parsedate(
     value: str,
     dayfirst: bool = False,
     yearfirst: bool = False,
-    errors: Optional[object] = None,
+    errors: object | None = None,
-) -> Optional[str]:
+) -> str | None:
     """
     Parse a date and convert it to ISO date format: yyyy-mm-dd
     \b
@@ -44,8 +44,8 @@ def parsedatetime(
     value: str,
     dayfirst: bool = False,
     yearfirst: bool = False,
-    errors: Optional[object] = None,
+    errors: object | None = None,
-) -> Optional[str]:
+) -> str | None:
     """
     Parse a datetime and convert it to ISO datetime format: yyyy-mm-ddTHH:MM:SS
     \b

sqlite_utils/utils.py

@@ -9,20 +9,11 @@ import itertools
 import json
 import os
 import sys
+from collections.abc import Callable, Generator, Iterable, Iterator
 from typing import (
+    TYPE_CHECKING,
     Any,
     BinaryIO,
-    Callable,
-    Dict,
-    Generator,
-    Iterable,
-    Iterator,
-    List,
-    Optional,
-    Set,
-    Tuple,
-    Type,
-    TYPE_CHECKING,
     TypeVar,
     Union,
     cast,
@@ -33,8 +24,8 @@ import click
 from . import recipes
 
 if TYPE_CHECKING:
-    import sqlite3  # noqa: F401
+    import sqlite3
-    from sqlite3 import dbapi2  # noqa: F401
+    from sqlite3 import dbapi2
 
     OperationalError = dbapi2.OperationalError
 else:
@@ -44,7 +35,7 @@ else:
         OperationalError = dbapi2.OperationalError
     except ImportError:
         import sqlite3  # noqa: F401
-        from sqlite3 import dbapi2  # noqa: F401
+        from sqlite3 import dbapi2
 
         OperationalError = dbapi2.OperationalError
 
@@ -61,8 +52,8 @@ SPATIALITE_PATHS = (
 ORIGINAL_CSV_FIELD_SIZE_LIMIT = csv.field_size_limit()
 
 # Type alias for row dictionaries - values can be various SQLite-compatible types
-RowValue = Union[None, int, float, str, bytes, bool, List[str]]
+RowValue = None | int | float | str | bytes | bool | list[str]
-Row = Dict[str, RowValue]
+Row = dict[str, RowValue]
 
 T = TypeVar("T")
 
@@ -103,7 +94,7 @@ def maximize_csv_field_size_limit() -> None:
             field_size_limit = int(field_size_limit / 10)
 
 
-def find_spatialite() -> Optional[str]:
+def find_spatialite() -> str | None:
     """
     The ``find_spatialite()`` function searches for the `SpatiaLite <https://www.gaia-gis.it/fossil/libspatialite/index>`__
     SQLite extension in some common places. It returns a string path to the location, or ``None`` if SpatiaLite was not found.
@@ -132,9 +123,9 @@ def find_spatialite() -> Optional[str]:
 
 
 def suggest_column_types(
-    records: Iterable[Dict[str, Any]],
+    records: Iterable[dict[str, Any]],
-) -> Dict[str, type]:
+) -> dict[str, type]:
-    all_column_types: Dict[str, Set[type]] = {}
+    all_column_types: dict[str, set[type]] = {}
     for record in records:
         for key, value in record.items():
             all_column_types.setdefault(key, set()).add(type(value))
@@ -142,9 +133,9 @@ def suggest_column_types(
 
 
 def types_for_column_types(
-    all_column_types: Dict[str, Set[type]],
+    all_column_types: dict[str, set[type]],
-) -> Dict[str, type]:
+) -> dict[str, type]:
-    column_types: Dict[str, type] = {}
+    column_types: dict[str, type] = {}
     for key, types in all_column_types.items():
         # Ignore null values if at least one other type present:
         if len(types) > 1:
@@ -153,7 +144,7 @@ def types_for_column_types(
         if {None.__class__} == types:
             t = str
         elif len(types) == 1:
-            t = list(types)[0]
+            t = next(iter(types))
             # But if it's a subclass of list / tuple / dict, use str
             # instead as we will be storing it as JSON in the table
             for superclass in (list, tuple, dict):
@@ -190,7 +181,7 @@ def column_affinity(column_type: str) -> type:
     return float
 
 
-def decode_base64_values(doc: Dict[str, Any]) -> Dict[str, Any]:
+def decode_base64_values(doc: dict[str, Any]) -> dict[str, Any]:
     # Looks for '{"$base64": true..., "encoded": ...}' values and decodes them
     to_fix = [
         k
@@ -263,9 +254,9 @@ class RowError(Exception):
 
 
 def _extra_key_strategy(
-    reader: Iterable[Dict[Optional[str], object]],
+    reader: Iterable[dict[str | None, object]],
-    ignore_extras: Optional[bool] = False,
+    ignore_extras: bool | None = False,
-    extras_key: Optional[str] = None,
+    extras_key: str | None = None,
 ) -> Iterable[Row]:
     # Logic for handling CSV rows with more values than there are headings
     for row in reader:
@@ -279,9 +270,7 @@ def _extra_key_strategy(
             yield cast(Row, row)
         elif not extras_key:
             extras = row.pop(None)
-            raise RowError(
+            raise RowError(f"Row {row} contained these extra values: {extras}")
-                "Row {} contained these extra values: {}".format(row, extras)
-            )
         else:
             extras_value = row.pop(None)
             row_out = cast(Row, row)
@@ -291,12 +280,12 @@ def _extra_key_strategy(
 
 def rows_from_file(
     fp: BinaryIO,
-    format: Optional[Format] = None,
+    format: Format | None = None,
-    dialect: Optional[Type[csv.Dialect]] = None,
+    dialect: type[csv.Dialect] | None = None,
-    encoding: Optional[str] = None,
+    encoding: str | None = None,
-    ignore_extras: Optional[bool] = False,
+    ignore_extras: bool | None = False,
-    extras_key: Optional[str] = None,
+    extras_key: str | None = None,
-) -> Tuple[Iterable[Row], Format]:
+) -> tuple[Iterable[Row], Format]:
     """
     Load a sequence of dictionaries from a file-like object containing one of four different formats.
 
@@ -363,7 +352,7 @@ def rows_from_file(
         )
         return (
             _extra_key_strategy(
-                cast(Iterable[Dict[Optional[str], object]], rows),
+                cast(Iterable[dict[str | None, object]], rows),
                 ignore_extras,
                 extras_key,
             ),
@@ -379,7 +368,7 @@ def rows_from_file(
             raise TypeError(
                 "rows_from_file() requires a file-like object that supports peek(), such as io.BytesIO"
             )
-        if first_bytes.startswith(b"[") or first_bytes.startswith(b"{"):
+        if first_bytes.startswith((b"[", b"{")):
             # TODO: Detect newline-JSON
             return rows_from_file(buffered, format=Format.JSON)
         else:
@@ -393,7 +382,7 @@ def rows_from_file(
             detected_format = Format.TSV if dialect.delimiter == "\t" else Format.CSV
             return (
                 _extra_key_strategy(
-                    cast(Iterable[Dict[Optional[str], object]], rows),
+                    cast(Iterable[dict[str | None, object]], rows),
                     ignore_extras,
                     extras_key,
                 ),
@@ -425,9 +414,9 @@ class TypeTracker:
     """
 
     def __init__(self) -> None:
-        self.trackers: Dict[str, "ValueTracker"] = {}
+        self.trackers: dict[str, ValueTracker] = {}
 
-    def wrap(self, iterator: Iterable[Dict[str, Any]]) -> Iterable[Dict[str, Any]]:
+    def wrap(self, iterator: Iterable[dict[str, Any]]) -> Iterable[dict[str, Any]]:
         """
         Use this to loop through an existing iterator, tracking the column types
         as part of the iteration.
@@ -441,7 +430,7 @@ class TypeTracker:
             yield row
 
     @property
-    def types(self) -> Dict[str, str]:
+    def types(self) -> dict[str, str]:
         """
         A dictionary mapping column names to their detected types. This can be passed
         to the ``db[table_name].transform(types=tracker.types)`` method.
@@ -450,17 +439,15 @@ class TypeTracker:
 
 
 class ValueTracker:
-    couldbe: Dict[str, Callable[[object], bool]]
+    couldbe: dict[str, Callable[[object], bool]]
 
     def __init__(self) -> None:
         self.couldbe = {key: getattr(self, "test_" + key) for key in self.get_tests()}
 
     @classmethod
-    def get_tests(cls) -> List[str]:
+    def get_tests(cls) -> list[str]:
         return [
-            key.split("test_")[-1]
+            key.split("test_")[-1] for key in cls.__dict__ if key.startswith("test_")
-            for key in cls.__dict__.keys()
-            if key.startswith("test_")
         ]
 
     def test_integer(self, value: object) -> bool:
@@ -492,7 +479,7 @@ class ValueTracker:
     def evaluate(self, value: object) -> None:
         if not value or not self.couldbe:
             return
-        not_these: List[str] = []
+        not_these: list[str] = []
         for name, test in self.couldbe.items():
             if not test(value):
                 not_these.append(name)
@@ -524,14 +511,14 @@ def progressbar(*args: Iterable[T], **kwargs: Any) -> Generator[Any, None, None]
 def _compile_code(
     code: str, imports: Iterable[str], variable: str = "value"
 ) -> Callable[..., Any]:
-    globals_dict: Dict[str, Any] = {"r": recipes, "recipes": recipes}
+    globals_dict: dict[str, Any] = {"r": recipes, "recipes": recipes}
     # Handle imports first so they're available for all approaches
     for import_ in imports:
         globals_dict[import_.split(".")[0]] = __import__(import_)
 
     # If user defined a convert() function, return that
     try:
-        exec(code, globals_dict)
+        exec(code, globals_dict)  # noqa: S102
         return cast(Callable[..., object], globals_dict["convert"])
     except (AttributeError, SyntaxError, NameError, KeyError, TypeError):
         pass
@@ -542,20 +529,20 @@ def _compile_code(
         fn = eval(code, globals_dict)
         if callable(fn):
             return cast(Callable[..., object], fn)
-    except Exception:
+    except Exception:  # noqa: BLE001, S110
         pass
 
     # Try compiling their code as a function instead
     body_variants = [code]
     # If single line and no 'return', try adding the return
     if "\n" not in code and not code.strip().startswith("return "):
-        body_variants.insert(0, "return {}".format(code))
+        body_variants.insert(0, f"return {code}")
 
     code_o = None
     for variant in body_variants:
-        new_code = ["def fn({}):".format(variable)]
+        new_code = [f"def fn({variable}):"]
         for line in variant.split("\n"):
-            new_code.append("    {}".format(line))
+            new_code.append(f"    {line}")
         try:
             code_o = compile("\n".join(new_code), "<string>", "exec")
             break
@@ -566,7 +553,7 @@ def _compile_code(
     if code_o is None:
         raise SyntaxError("Could not compile code")
 
-    exec(code_o, globals_dict)
+    exec(code_o, globals_dict)  # noqa: S102
     return cast(Callable[..., object], globals_dict["fn"])
 
 
@@ -582,7 +569,7 @@ def chunks(sequence: Iterable[T], size: int) -> Iterable[Iterable[T]]:
         yield itertools.chain([item], itertools.islice(iterator, size - 1))
 
 
-def hash_record(record: Dict[str, Any], keys: Optional[Iterable[str]] = None) -> str:
+def hash_record(record: dict[str, Any], keys: Iterable[str] | None = None) -> str:
     """
     ``record`` should be a Python dictionary. Returns a sha1 hash of the
     keys and values in that record.
@@ -603,7 +590,7 @@ def hash_record(record: Dict[str, Any], keys: Optional[Iterable[str]] = None) ->
     :param record: Record to generate a hash for
     :param keys: Subset of keys to use for that hash
     """
-    to_hash: Dict[str, Any] = record
+    to_hash: dict[str, Any] = record
     if keys is not None:
         to_hash = {key: record[key] for key in keys}
     return hashlib.sha1(
@@ -613,7 +600,7 @@ def hash_record(record: Dict[str, Any], keys: Optional[Iterable[str]] = None) ->
     ).hexdigest()
 
 
-def dedupe_keys(keys: Iterable[str]) -> List[str]:
+def dedupe_keys(keys: Iterable[str]) -> list[str]:
     """
     Rename duplicates in a list of column names so every name is unique,
     by appending ``_2``, ``_3``... to later occurrences - skipping any
@@ -636,7 +623,7 @@ def dedupe_keys(keys: Iterable[str]) -> List[str]:
             new_key = key
             suffix = 2
             while new_key in seen or new_key in taken:
-                new_key = "{}_{}".format(key, suffix)
+                new_key = f"{key}_{suffix}"
                 suffix += 1
             key = new_key
         seen.add(key)
@@ -644,7 +631,7 @@ def dedupe_keys(keys: Iterable[str]) -> List[str]:
     return result
 
 
-def _flatten(d: Dict[str, Any]) -> Generator[Tuple[str, Any], None, None]:
+def _flatten(d: dict[str, Any]) -> Generator[tuple[str, Any], None, None]:
     for key, value in d.items():
         if isinstance(value, dict):
             for key2, value2 in _flatten(value):
@@ -653,7 +640,7 @@ def _flatten(d: Dict[str, Any]) -> Generator[Tuple[str, Any], None, None]:
             yield key, value
 
 
-def flatten(row: Dict[str, Any]) -> Dict[str, Any]:
+def flatten(row: dict[str, Any]) -> dict[str, Any]:
     """
     Turn a nested dict e.g. ``{"a": {"b": 1}}`` into a flat dict: ``{"a_b": 1}``
 

tests/conftest.py

@@ -1,6 +1,7 @@
+import pytest
+
 from sqlite_utils import Database
 from sqlite_utils.utils import sqlite3
-import pytest
 
 CREATE_TABLES = """
 create table Gosh (c1 text, c2 text, c3 text);
@@ -55,7 +56,7 @@ def close_all_databases():
     for db in databases:
         try:
             db.close()
-        except Exception:
+        except sqlite3.Error:
             pass
 
 

tests/test_analyze_tables.py

@@ -1,9 +1,11 @@
-from sqlite_utils.db import Database, ColumnDetails
-from sqlite_utils import cli
-from click.testing import CliRunner
-import pytest
 import sqlite3
 
+import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import cli
+from sqlite_utils.db import ColumnDetails, Database
+
 
 @pytest.fixture
 def db_to_analyze(fresh_db):

tests/test_atomic.py

@@ -28,11 +28,13 @@ from sqlite_utils.utils import sqlite3
             END;
             """,
             [
-                "CREATE TRIGGER t_ai AFTER INSERT ON t\n"
+                (
-                "            BEGIN\n"
+                    "CREATE TRIGGER t_ai AFTER INSERT ON t\n"
-                "                UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
+                    "            BEGIN\n"
-                "                INSERT INTO log VALUES ('x;y');\n"
+                    "                UPDATE t SET value = 'a;b' WHERE id = new.id;\n"
-                "            END;"
+                    "                INSERT INTO log VALUES ('x;y');\n"
+                    "            END;"
+                )
             ],
         ),
     ),
@@ -49,10 +51,9 @@ def test_atomic_commits(fresh_db):
 
 
 def test_atomic_rolls_back(fresh_db):
-    with pytest.raises(RuntimeError):
+    with pytest.raises(RuntimeError), fresh_db.atomic():
-        with fresh_db.atomic():
+        fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
-            fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
+        raise RuntimeError("boom")
-            raise RuntimeError("boom")
 
     assert not fresh_db["dogs"].exists()
 
@@ -62,10 +63,9 @@ def test_nested_atomic_rolls_back_to_savepoint(fresh_db):
 
     with fresh_db.atomic():
         fresh_db["dogs"].insert({"id": 1, "name": "Cleo"})
-        with pytest.raises(RuntimeError):
+        with pytest.raises(RuntimeError), fresh_db.atomic():
-            with fresh_db.atomic():
+            fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
-                fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
+            raise RuntimeError("boom")
-                raise RuntimeError("boom")
         fresh_db["dogs"].insert({"id": 3, "name": "Marnie"})
 
     assert list(fresh_db["dogs"].rows) == [
@@ -75,20 +75,18 @@ def test_nested_atomic_rolls_back_to_savepoint(fresh_db):
 
 
 def test_outer_atomic_rolls_back_released_savepoint(fresh_db):
-    with pytest.raises(RuntimeError):
+    with pytest.raises(RuntimeError), fresh_db.atomic():
+        fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
         with fresh_db.atomic():
-            fresh_db["dogs"].insert({"id": 1, "name": "Cleo"}, pk="id")
+            fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
-            with fresh_db.atomic():
+        raise RuntimeError("boom")
-                fresh_db["dogs"].insert({"id": 2, "name": "Pancakes"})
-            raise RuntimeError("boom")
 
     assert not fresh_db["dogs"].exists()
 
 
 def test_executescript_does_not_commit_open_atomic_block(fresh_db):
-    with pytest.raises(RuntimeError):
+    with pytest.raises(RuntimeError), fresh_db.atomic():
-        with fresh_db.atomic():
+        fresh_db.executescript("""
-            fresh_db.executescript("""
                 CREATE TABLE dogs(id INTEGER PRIMARY KEY, name TEXT);
                 CREATE TRIGGER dogs_ai AFTER INSERT ON dogs
                 BEGIN
@@ -97,7 +95,7 @@ def test_executescript_does_not_commit_open_atomic_block(fresh_db):
                 -- This comment has a semicolon;
                 INSERT INTO dogs VALUES (1, 'Cleo; the first');
             """)
-            raise RuntimeError("boom")
+        raise RuntimeError("boom")
 
     assert not fresh_db["dogs"].exists()
 
@@ -105,11 +103,10 @@ def test_executescript_does_not_commit_open_atomic_block(fresh_db):
 def test_transform_does_not_commit_open_atomic_block(fresh_db):
     fresh_db["dogs"].insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
 
-    with pytest.raises(RuntimeError):
+    with pytest.raises(RuntimeError), fresh_db.atomic():
-        with fresh_db.atomic():
+        fresh_db["dogs"].insert({"id": 2, "name": "Pancakes", "age": "6"})
-            fresh_db["dogs"].insert({"id": 2, "name": "Pancakes", "age": "6"})
+        fresh_db["dogs"].transform(rename={"age": "dog_age"})
-            fresh_db["dogs"].transform(rename={"age": "dog_age"})
+        raise RuntimeError("boom")
-            raise RuntimeError("boom")
 
     assert (
         fresh_db["dogs"].schema
@@ -149,10 +146,9 @@ def test_transform_parent_table_with_foreign_keys_rolls_back(fresh_db):
         foreign_keys={"author_id"},
     )
 
-    with pytest.raises(RuntimeError):
+    with pytest.raises(RuntimeError), fresh_db.atomic():
-        with fresh_db.atomic():
+        fresh_db["authors"].transform(rename={"name": "full_name"})
-            fresh_db["authors"].transform(rename={"name": "full_name"})
+        raise RuntimeError("boom")
-            raise RuntimeError("boom")
 
     assert (
         fresh_db["authors"].schema
@@ -354,9 +350,11 @@ def test_atomic_preserves_error_from_transaction_destroying_trigger(fresh_db):
     # with "cannot rollback - no transaction is active"
     fresh_db.execute("create table t (id integer primary key, v text)")
     fresh_db.execute(TRIGGER_SQL)
-    with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
+    with (
-        with fresh_db.atomic():
+        pytest.raises(sqlite3.IntegrityError, match="trigger says no"),
-            fresh_db.execute("insert into t (v) values ('bad')")
+        fresh_db.atomic(),
+    ):
+        fresh_db.execute("insert into t (v) values ('bad')")
     assert not fresh_db.conn.in_transaction
 
 
@@ -367,16 +365,17 @@ def test_nested_atomic_preserves_error_from_transaction_destroying_trigger(
     # "no such savepoint" from ROLLBACK TO SAVEPOINT
     fresh_db.execute("create table t (id integer primary key, v text)")
     fresh_db.execute(TRIGGER_SQL)
-    with pytest.raises(sqlite3.IntegrityError, match="trigger says no"):
+    with (
-        with fresh_db.atomic():
+        pytest.raises(sqlite3.IntegrityError, match="trigger says no"),
-            with fresh_db.atomic():
+        fresh_db.atomic(),
-                fresh_db.execute("insert into t (v) values ('bad')")
+        fresh_db.atomic(),
+    ):
+        fresh_db.execute("insert into t (v) values ('bad')")
     assert not fresh_db.conn.in_transaction
 
 
 def test_atomic_preserves_error_from_insert_or_rollback(fresh_db):
     fresh_db["t"].insert({"id": 1}, pk="id")
-    with pytest.raises(sqlite3.IntegrityError):
+    with pytest.raises(sqlite3.IntegrityError), fresh_db.atomic():
-        with fresh_db.atomic():
+        fresh_db.execute("insert or rollback into t (id) values (1)")
-            fresh_db.execute("insert or rollback into t (id) values (1)")
     assert not fresh_db.conn.in_transaction

tests/test_cli.py

@@ -1,14 +1,16 @@
-from sqlite_utils import cli, Database
-from sqlite_utils.db import Index, ForeignKey
-from click.testing import CliRunner
-from pathlib import Path
-import subprocess
-import sqlite3
-import sys
 import json
 import os
-import pytest
+import sqlite3
+import subprocess
+import sys
 import textwrap
+from pathlib import Path
+
+import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import Database, cli
+from sqlite_utils.db import ForeignKey, Index
 
 
 def write_json(file_path, data):
@@ -21,7 +23,7 @@ def _supports_pragma_function_list():
     try:
         db.execute("select * from pragma_function_list()")
         return True
-    except Exception:
+    except sqlite3.DatabaseError:
         return False
     finally:
         db.close()
@@ -184,9 +186,9 @@ def test_output_table(db_path, options, expected):
         db["rows"].insert_all(
             [
                 {
-                    "c1": "verb{}".format(i),
+                    "c1": f"verb{i}",
-                    "c2": "noun{}".format(i),
+                    "c2": f"noun{i}",
-                    "c3": "adjective{}".format(i),
+                    "c3": f"adjective{i}",
                 }
                 for i in range(4)
             ]
@@ -678,9 +680,9 @@ def test_optimize(db_path, tables):
             db[table].insert_all(
                 [
                     {
-                        "c1": "verb{}".format(i),
+                        "c1": f"verb{i}",
-                        "c2": "noun{}".format(i),
+                        "c2": f"noun{i}",
-                        "c3": "adjective{}".format(i),
+                        "c3": f"adjective{i}",
                     }
                     for i in range(10000)
                 ]
@@ -704,9 +706,9 @@ def test_rebuild_fts_fixes_docsize_error(db_path):
     db = Database(db_path, recursive_triggers=False)
     records = [
         {
-            "c1": "verb{}".format(i),
+            "c1": f"verb{i}",
-            "c2": "noun{}".format(i),
+            "c2": f"noun{i}",
-            "c3": "adjective{}".format(i),
+            "c3": f"adjective{i}",
         }
         for i in range(10000)
     ]
@@ -1019,16 +1021,14 @@ def test_query_json_binary(db_path):
             "data": {
                 "$base64": True,
                 "encoded": (
-                    (
+                    "eJzt0c1xAyEMBeC7q1ABHleR3HxNAQrIjmb4M0gelx+RTY7p4N2WBYT0vmufUknH"
-                        "eJzt0c1xAyEMBeC7q1ABHleR3HxNAQrIjmb4M0gelx+RTY7p4N2WBYT0vmufUknH"
+                    "8kq5lz5pqRFXsTOl3pYkE/NJnHXoStruJEVjc0mOCyTqq/ZMJnXEZW1Js2ZvRm5U+"
-                        "8kq5lz5pqRFXsTOl3pYkE/NJnHXoStruJEVjc0mOCyTqq/ZMJnXEZW1Js2ZvRm5U+"
+                    "DPKk9hRWqjyvTFx0YfzhT6MpGmN2lR1fzxjyfVMD9dFrS+bnkleMpMam/ZGXgrX1I"
-                        "DPKk9hRWqjyvTFx0YfzhT6MpGmN2lR1fzxjyfVMD9dFrS+bnkleMpMam/ZGXgrX1I"
+                    "/K+5Au3S/9lNQRh0k4Gq/RUz8GiKfsQm+7JLsJ6fTo5JhVG00ZU76kZZkxePx49uI"
-                        "/K+5Au3S/9lNQRh0k4Gq/RUz8GiKfsQm+7JLsJ6fTo5JhVG00ZU76kZZkxePx49uI"
+                    "jnpNoJyYlWUsoaSl/CcVATje/Kxu13RANnrHweaH3V5Jh4jvGyKCnxJLiXPKhmW3f"
-                        "jnpNoJyYlWUsoaSl/CcVATje/Kxu13RANnrHweaH3V5Jh4jvGyKCnxJLiXPKhmW3f"
+                    "iCnG7Jql7RR3UvFo8jJ4z039dtOkTFmWzL1be9lt8A5II471m6vXy+l0BR/4wAc+8"
-                        "iCnG7Jql7RR3UvFo8jJ4z039dtOkTFmWzL1be9lt8A5II471m6vXy+l0BR/4wAc+8"
+                    "IEPfOADH/jABz7wgQ984AMf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984A"
-                        "IEPfOADH/jABz7wgQ984AMf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984A"
+                    "Mf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984PuP7xubBoN9"
-                        "Mf+MAHPvCBD3zgAx/4wAc+8IEPfOADH/jABz7wgQ984PuP7xubBoN9"
-                    )
                 ),
             },
         }
@@ -2114,11 +2114,13 @@ _common_other_schema = (
         ),
         (
             ["--rename", "name", "name2"],
-            'CREATE TABLE "trees" (\n'
+            (
-            '   "id" INTEGER PRIMARY KEY,\n'
+                'CREATE TABLE "trees" (\n'
-            '   "address" TEXT,\n'
+                '   "id" INTEGER PRIMARY KEY,\n'
-            '   "species_id" INTEGER REFERENCES "species"("id")\n'
+                '   "address" TEXT,\n'
-            ")",
+                '   "species_id" INTEGER REFERENCES "species"("id")\n'
+                ")"
+            ),
             'CREATE TABLE "species" (\n   "id" INTEGER PRIMARY KEY,\n   "species" TEXT\n)',
         ),
     ],
@@ -2137,9 +2139,9 @@ def test_extract(db_path, args, expected_table_schema, expected_other_schema):
     assert result.exit_code == 0
     schema = db["trees"].schema
     assert schema == expected_table_schema
-    other_schema = [t for t in db.tables if t.name not in ("trees", "Gosh", "Gosh2")][
+    other_schema = next(
-        0
+        t for t in db.tables if t.name not in ("trees", "Gosh", "Gosh2")
-    ].schema
+    ).schema
     assert other_schema == expected_other_schema
 
 
@@ -2431,7 +2433,7 @@ def test_long_csv_column_value(tmpdir):
     with open(csv_path, "w") as csv_file:
         long_string = "a" * 131073
         csv_file.write("id,text\n")
-        csv_file.write("1,{}\n".format(long_string))
+        csv_file.write(f"1,{long_string}\n")
     result = CliRunner().invoke(
         cli.cli,
         ["insert", db_path, "bigtable", csv_path, "--csv"],
@@ -2457,8 +2459,8 @@ def test_import_no_headers(tmpdir, args, tsv):
     csv_path = str(tmpdir / "test.csv")
     with open(csv_path, "w") as csv_file:
         sep = "\t" if tsv else ","
-        csv_file.write("Cleo{sep}Dog{sep}5\n".format(sep=sep))
+        csv_file.write(f"Cleo{sep}Dog{sep}5\n")
-        csv_file.write("Tracy{sep}Spider{sep}7\n".format(sep=sep))
+        csv_file.write(f"Tracy{sep}Spider{sep}7\n")
     result = CliRunner().invoke(
         cli.cli,
         ["insert", db_path, "creatures", csv_path] + args + ["--no-detect-types"],
@@ -2690,7 +2692,9 @@ def test_integer_overflow_error(tmpdir):
 def test_python_dash_m():
     "Tool can be run using python -m sqlite_utils"
     result = subprocess.run(
-        [sys.executable, "-m", "sqlite_utils", "--help"], stdout=subprocess.PIPE
+        [sys.executable, "-m", "sqlite_utils", "--help"],
+        stdout=subprocess.PIPE,
+        check=False,
     )
     assert result.returncode == 0
     assert b"Commands for interacting with a SQLite database" in result.stdout
@@ -2830,14 +2834,14 @@ def test_load_extension(entrypoint, should_pass, should_fail):
     for func in should_pass:
         result = CliRunner().invoke(
             cli.cli,
-            ["memory", "select {}()".format(func), "--load-extension", ext],
+            ["memory", f"select {func}()", "--load-extension", ext],
             catch_exceptions=False,
         )
         assert result.exit_code == 0
     for func in should_fail:
         result = CliRunner().invoke(
             cli.cli,
-            ["memory", "select {}()".format(func), "--load-extension", ext],
+            ["memory", f"select {func}()", "--load-extension", ext],
             catch_exceptions=False,
         )
         assert result.exit_code == 1

tests/test_cli_bulk.py

@@ -1,11 +1,13 @@
-from click.testing import CliRunner
-from sqlite_utils import cli, Database
 import pathlib
-import pytest
 import subprocess
 import sys
 import time
 
+import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import Database, cli
+
 
 @pytest.fixture
 def test_db_and_path(tmpdir):

tests/test_cli_convert.py

@@ -1,10 +1,12 @@
-from click.testing import CliRunner
-from sqlite_utils import cli
-import sqlite_utils
 import json
-import textwrap
 import pathlib
+import textwrap
+
 import pytest
+from click.testing import CliRunner
+
+import sqlite_utils
+from sqlite_utils import cli
 
 
 @pytest.fixture
@@ -50,7 +52,7 @@ def test_convert_code(fresh_db_and_path, code):
         cli.cli, ["convert", db_path, "t", "text", code], catch_exceptions=False
     )
     assert result.exit_code == 0, result.output
-    value = list(db["t"].rows)[0]["text"]
+    value = next(iter(db["t"].rows))["text"]
     assert value == "Spooktober"
 
 
@@ -442,7 +444,7 @@ def test_recipe_jsonsplit(tmpdir, delimiter):
     )
     code = "r.jsonsplit(value)"
     if delimiter:
-        code = 'recipes.jsonsplit(value, delimiter="{}")'.format(delimiter)
+        code = f'recipes.jsonsplit(value, delimiter="{delimiter}")'
     args = ["convert", db_path, "example", "tags", code]
     result = CliRunner().invoke(cli.cli, args)
     assert result.exit_code == 0, result.output
@@ -470,7 +472,7 @@ def test_recipe_jsonsplit_type(fresh_db_and_path, type, expected_array):
     )
     code = "r.jsonsplit(value)"
     if type:
-        code = "recipes.jsonsplit(value, type={})".format(type)
+        code = f"recipes.jsonsplit(value, type={type})"
     args = ["convert", db_path, "example", "records", code]
     result = CliRunner().invoke(cli.cli, args)
     assert result.exit_code == 0, result.output

tests/test_cli_insert.py

@@ -1,11 +1,13 @@
-from sqlite_utils import cli, Database
-from click.testing import CliRunner
 import json
-import pytest
 import subprocess
 import sys
 import time
 
+import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import Database, cli
+
 
 def test_insert_simple(tmpdir):
     json_path = str(tmpdir / "dog.json")
@@ -99,7 +101,7 @@ def test_insert_with_primary_keys(db_path, tmpdir, args, expected_pks):
 
 def test_insert_multiple_with_primary_key(db_path, tmpdir):
     json_path = str(tmpdir / "dogs.json")
-    dogs = [{"id": i, "name": "Cleo {}".format(i), "age": i + 3} for i in range(1, 21)]
+    dogs = [{"id": i, "name": f"Cleo {i}", "age": i + 3} for i in range(1, 21)]
     with open(json_path, "w") as fp:
         fp.write(json.dumps(dogs))
     result = CliRunner().invoke(
@@ -114,7 +116,7 @@ def test_insert_multiple_with_primary_key(db_path, tmpdir):
 def test_insert_multiple_with_compound_primary_key(db_path, tmpdir):
     json_path = str(tmpdir / "dogs.json")
     dogs = [
-        {"breed": "mixed", "id": i, "name": "Cleo {}".format(i), "age": i + 3}
+        {"breed": "mixed", "id": i, "name": f"Cleo {i}", "age": i + 3}
         for i in range(1, 21)
     ]
     with open(json_path, "w") as fp:
@@ -140,8 +142,7 @@ def test_insert_multiple_with_compound_primary_key(db_path, tmpdir):
 def test_insert_not_null_default(db_path, tmpdir):
     json_path = str(tmpdir / "dogs.json")
     dogs = [
-        {"id": i, "name": "Cleo {}".format(i), "age": i + 3, "score": 10}
+        {"id": i, "name": f"Cleo {i}", "age": i + 3, "score": 10} for i in range(1, 21)
-        for i in range(1, 21)
     ]
     with open(json_path, "w") as fp:
         fp.write(json.dumps(dogs))
@@ -587,7 +588,7 @@ def test_insert_streaming_batch_size_1(db_path):
                 return
             tries += 1
             if tries > 10:
-                assert False, "Expected {}, got {}".format(expected, rows)
+                assert False, f"Expected {expected}, got {rows}"
             time.sleep(tries * 0.1)
 
     try_until([{"name": "Azi"}])

tests/test_cli_memory.py

@@ -1,5 +1,6 @@
-import click
 import json
+
+import click
 import pytest
 from click.testing import CliRunner
 
@@ -28,7 +29,7 @@ def test_memory_csv(tmpdir, sql_from, use_stdin):
             fp.write(content)
     result = CliRunner().invoke(
         cli.cli,
-        ["memory", csv_path, "select * from {}".format(sql_from), "--nl"],
+        ["memory", csv_path, f"select * from {sql_from}", "--nl"],
         input=input,
     )
     assert result.exit_code == 0
@@ -53,7 +54,7 @@ def test_memory_tsv(tmpdir, use_stdin):
         sql_from = "chickens"
     result = CliRunner().invoke(
         cli.cli,
-        ["memory", path, "select * from {}".format(sql_from)],
+        ["memory", path, f"select * from {sql_from}"],
         input=input,
     )
     assert result.exit_code == 0, result.output
@@ -79,7 +80,7 @@ def test_memory_json(tmpdir, use_stdin):
         sql_from = "chickens"
     result = CliRunner().invoke(
         cli.cli,
-        ["memory", path, "select * from {}".format(sql_from)],
+        ["memory", path, f"select * from {sql_from}"],
         input=input,
     )
     assert result.exit_code == 0, result.output
@@ -105,7 +106,7 @@ def test_memory_json_nl(tmpdir, use_stdin):
         sql_from = "chickens"
     result = CliRunner().invoke(
         cli.cli,
-        ["memory", path, "select * from {}".format(sql_from)],
+        ["memory", path, f"select * from {sql_from}"],
         input=input,
     )
     assert result.exit_code == 0, result.output
@@ -135,7 +136,7 @@ def test_memory_csv_encoding(tmpdir, use_stdin):
         CliRunner()
         .invoke(
             cli.cli,
-            ["memory", csv_path, "select * from {}".format(sql_from), "--nl"],
+            ["memory", csv_path, f"select * from {sql_from}", "--nl"],
             input=input,
         )
         .exit_code

tests/test_cli_migrate.py

@@ -1,7 +1,8 @@
 import pathlib
 
-from click.testing import CliRunner
 import pytest
+from click.testing import CliRunner
+
 import sqlite_utils
 import sqlite_utils.cli
 

tests/test_column_affinity.py

@@ -1,4 +1,5 @@
 import pytest
+
 from sqlite_utils.utils import column_affinity
 
 EXAMPLES = [
@@ -41,5 +42,5 @@ def test_column_affinity(column_def, expected_type):
 
 @pytest.mark.parametrize("column_def,expected_type", EXAMPLES)
 def test_columns_dict(fresh_db, column_def, expected_type):
-    fresh_db.execute("create table foo (col {})".format(column_def))
+    fresh_db.execute(f"create table foo (col {column_def})")
     assert {"col": expected_type} == fresh_db["foo"].columns_dict

tests/test_constructor.py

@@ -1,8 +1,10 @@
+import sys
+
+import pytest
+
 from sqlite_utils import Database
 from sqlite_utils.db import TransactionError
 from sqlite_utils.utils import sqlite3
-import pytest
-import sys
 
 
 def test_recursive_triggers():

tests/test_convert.py

@@ -1,6 +1,7 @@
-from sqlite_utils.db import BadMultiValues
 import pytest
 
+from sqlite_utils.db import BadMultiValues
+
 
 @pytest.mark.parametrize(
     "columns,fn,expected",

tests/test_create.py

@@ -1,26 +1,28 @@
-from sqlite_utils.db import (
-    Index,
-    Database,
-    DescIndex,
-    AlterError,
-    InvalidColumns,
-    NoObviousTable,
-    OperationalError,
-    ForeignKey,
-    Table,
-    View,
-    NoTable,
-    NoView,
-)
-from sqlite_utils.utils import hash_record, sqlite3
 import collections
 import datetime
 import decimal
 import json
 import pathlib
-import pytest
 import uuid
 
+import pytest
+
+from sqlite_utils.db import (
+    AlterError,
+    Database,
+    DescIndex,
+    ForeignKey,
+    Index,
+    InvalidColumns,
+    NoObviousTable,
+    NoTable,
+    NoView,
+    OperationalError,
+    Table,
+    View,
+)
+from sqlite_utils.utils import hash_record, sqlite3
+
 try:
     import pandas as pd  # type: ignore
 except ImportError:
@@ -699,7 +701,7 @@ def test_bulk_insert_more_than_999_values(fresh_db):
     "num_columns,should_error", ((900, False), (999, False), (1000, True))
 )
 def test_error_if_more_than_999_columns(fresh_db, num_columns, should_error):
-    record = dict([("c{}".format(i), i) for i in range(num_columns)])
+    record = {f"c{i}": i for i in range(num_columns)}
     if should_error:
         with pytest.raises(ValueError):
             fresh_db["big"].insert(record)
@@ -718,17 +720,9 @@ def test_columns_not_in_first_record_should_not_cause_batch_to_be_too_large(fres
     records = [
         {"c0": "first record"},  # one column in first record -> batch size = 999
         # fill out the batch with 99 records with enough columns to exceed THRESHOLD
-        *[
+        *[{f"c{i}": j for i in range(extra_columns)} for j in range(batch_size - 1)],
-            dict([("c{}".format(i), j) for i in range(extra_columns)])
-            for j in range(batch_size - 1)
-        ],
     ]
-    try:
+    fresh_db["too_many_columns"].insert_all(records, alter=True, batch_size=batch_size)
-        fresh_db["too_many_columns"].insert_all(
-            records, alter=True, batch_size=batch_size
-        )
-    except sqlite3.OperationalError:
-        raise
 
 
 @pytest.mark.parametrize(
@@ -910,7 +904,7 @@ def test_insert_list_nested_unicode(fresh_db):
 def test_insert_uuid(fresh_db):
     uuid4 = uuid.uuid4()
     fresh_db["test"].insert({"uuid": uuid4})
-    row = list(fresh_db["test"].rows)[0]
+    row = next(iter(fresh_db["test"].rows))
     assert {"uuid"} == row.keys()
     assert isinstance(row["uuid"], str)
     assert row["uuid"] == str(uuid4)
@@ -918,16 +912,14 @@ def test_insert_uuid(fresh_db):
 
 def test_insert_memoryview(fresh_db):
     fresh_db["test"].insert({"data": memoryview(b"hello")})
-    row = list(fresh_db["test"].rows)[0]
+    row = next(iter(fresh_db["test"].rows))
     assert {"data"} == row.keys()
     assert isinstance(row["data"], bytes)
     assert row["data"] == b"hello"
 
 
 def test_insert_thousands_using_generator(fresh_db):
-    fresh_db["test"].insert_all(
+    fresh_db["test"].insert_all({"i": i, "word": f"word_{i}"} for i in range(10000))
-        {"i": i, "word": "word_{}".format(i)} for i in range(10000)
-    )
     assert [{"name": "i", "type": "INTEGER"}, {"name": "word", "type": "TEXT"}] == [
         {"name": col.name, "type": col.type} for col in fresh_db["test"].columns
     ]
@@ -938,7 +930,7 @@ def test_insert_thousands_raises_exception_with_extra_columns_after_first_100(fr
     # https://github.com/simonw/sqlite-utils/issues/139
     with pytest.raises(Exception, match="table test has no column named extra"):
         fresh_db["test"].insert_all(
-            [{"i": i, "word": "word_{}".format(i)} for i in range(100)]
+            [{"i": i, "word": f"word_{i}"} for i in range(100)]
             + [{"i": 101, "extra": "This extra column should cause an exception"}],
         )
 
@@ -946,7 +938,7 @@ def test_insert_thousands_raises_exception_with_extra_columns_after_first_100(fr
 def test_insert_thousands_adds_extra_columns_after_first_100_with_alter(fresh_db):
     # https://github.com/simonw/sqlite-utils/issues/139
     fresh_db["test"].insert_all(
-        [{"i": i, "word": "word_{}".format(i)} for i in range(100)]
+        [{"i": i, "word": f"word_{i}"} for i in range(100)]
         + [{"i": 101, "extra": "Should trigger ALTER"}],
         alter=True,
     )
@@ -958,7 +950,7 @@ def test_insert_thousands_adds_extra_columns_after_first_100_with_alter(fresh_db
 def test_insert_all_pk_not_in_records_raises(fresh_db, num_rows):
     # https://github.com/simonw/sqlite-utils/issues/732
     fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
-    rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
+    rows = [{"a": f"x{i}", "b": i} for i in range(num_rows)]
 
     with pytest.raises(InvalidColumns) as ex:
         fresh_db["t"].insert_all(rows, pk="not_a_column")
@@ -975,7 +967,7 @@ def test_insert_all_pk_not_in_records_alter_raises(fresh_db, num_rows):
     # known - a pk column that is in neither the table nor the records
     # still raises
     fresh_db.conn.execute("CREATE TABLE t (a TEXT, b INT, PRIMARY KEY (a, b))")
-    rows = [{"a": "x{}".format(i), "b": i} for i in range(num_rows)]
+    rows = [{"a": f"x{i}", "b": i} for i in range(num_rows)]
 
     with pytest.raises(InvalidColumns) as ex:
         fresh_db["t"].insert_all(rows, pk="not_a_column", alter=True)
@@ -1146,7 +1138,7 @@ def test_insert_hash_id_columns(fresh_db, use_table_factory):
         insert_kwargs = {}
     else:
         dogs = fresh_db["dogs"]
-        insert_kwargs = dict(hash_id_columns=("name", "twitter"))
+        insert_kwargs = {"hash_id_columns": ("name", "twitter")}
 
     id = dogs.insert(
         {"name": "Cleo", "twitter": "cleopaws", "age": 5},
@@ -1654,7 +1646,7 @@ def test_upsert_uses_pk_from_prior_insert_655(fresh_db):
     # Upsert should work without specifying pk again
     table.upsert({"id": 1, "name": "Alice Updated"})
     assert table.count == 1
-    assert list(table.rows)[0]["name"] == "Alice Updated"
+    assert next(iter(table.rows))["name"] == "Alice Updated"
 
 
 def test_upsert_all_uses_pk_from_prior_insert_655(fresh_db):

tests/test_create_view.py

@@ -1,4 +1,5 @@
 import pytest
+
 from sqlite_utils.utils import OperationalError
 
 

tests/test_default_value.py

@@ -31,7 +31,7 @@ EXAMPLES = [
 
 @pytest.mark.parametrize("column_def,initial_value,expected_value", EXAMPLES)
 def test_quote_default_value(fresh_db, column_def, initial_value, expected_value):
-    fresh_db.execute("create table foo (col {})".format(column_def))
+    fresh_db.execute(f"create table foo (col {column_def})")
     assert initial_value == fresh_db["foo"].columns[0].default_value
     assert expected_value == fresh_db.quote_default_value(
         fresh_db["foo"].columns[0].default_value

tests/test_delete.py

@@ -3,7 +3,7 @@ import sqlite_utils
 
 def test_delete_rowid_table(fresh_db):
     table = fresh_db["table"]
-    table.insert({"foo": 1}).last_pk
+    table.insert({"foo": 1})
     rowid = table.insert({"foo": 2}).last_pk
     table.delete(rowid)
     assert [{"foo": 1}] == list(table.rows)

tests/test_docs.py

@@ -1,8 +1,10 @@
-from click.testing import CliRunner
-from sqlite_utils import cli, recipes
-from pathlib import Path
-import pytest
 import re
+from pathlib import Path
+
+import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import cli, recipes
 
 docs_path = Path(__file__).parent.parent / "docs"
 commands_re = re.compile(r"(?:\$ |    )sqlite-utils (\S+)")
@@ -34,7 +36,7 @@ def test_commands_are_documented(documented_commands, command):
 
 @pytest.mark.parametrize("command", cli.cli.commands.values())
 def test_commands_have_help(command):
-    assert command.help, "{} is missing its help".format(command)
+    assert command.help, f"{command} is missing its help"
 
 
 def test_convert_help():

tests/test_duplicate.py

@@ -1,7 +1,9 @@
-from sqlite_utils.db import NoTable
 import datetime
+
 import pytest
 
+from sqlite_utils.db import NoTable
+
 
 def test_duplicate(fresh_db):
     # Create table using native Sqlite statement:
@@ -12,7 +14,7 @@ def test_duplicate(fresh_db):
     "bool_col" INTEGER,
     "datetime_col" TEXT)""")
     # Insert one row of mock data:
-    dt = datetime.datetime.now()
+    dt = datetime.datetime.now(datetime.timezone.utc)
     data = {
         "text_col": "Cleo",
         "real_col": 3.14,

tests/test_enable_counts.py

@@ -1,14 +1,14 @@
-from sqlite_utils import Database
-from sqlite_utils import cli
-from click.testing import CliRunner
 import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import Database, cli
 
 
 def test_enable_counts_specific_table(fresh_db):
     foo = fresh_db["foo"]
     assert fresh_db.table_names() == []
     for i in range(10):
-        foo.insert({"name": "item {}".format(i)})
+        foo.insert({"name": f"item {i}"})
     assert fresh_db.table_names() == ["foo"]
     assert foo.count == 10
     # Now enable counts
@@ -44,7 +44,7 @@ def test_enable_counts_specific_table(fresh_db):
     assert list(fresh_db["_counts"].rows) == [{"count": 10, "table": "foo"}]
     # Add some items to test the triggers
     for i in range(5):
-        foo.insert({"name": "item {}".format(10 + i)})
+        foo.insert({"name": f"item {10 + i}"})
     assert foo.count == 15
     assert list(fresh_db["_counts"].rows) == [{"count": 15, "table": "foo"}]
     # Delete some items

tests/test_extract.py

@@ -1,19 +1,21 @@
-from sqlite_utils.db import InvalidColumns
 import itertools
+
 import pytest
 
+from sqlite_utils.db import InvalidColumns
+
 
 @pytest.mark.parametrize("table", [None, "Species"])
 @pytest.mark.parametrize("fk_column", [None, "species"])
 def test_extract_single_column(fresh_db, table, fk_column):
     expected_table = table or "species"
-    expected_fk = fk_column or "{}_id".format(expected_table)
+    expected_fk = fk_column or f"{expected_table}_id"
     iter_species = itertools.cycle(["Palm", "Spruce", "Mangrove", "Oak"])
     fresh_db["tree"].insert_all(
         (
             {
                 "id": i,
-                "name": "Tree {}".format(i),
+                "name": f"Tree {i}",
                 "species": next(iter_species),
                 "end": 1,
             }
@@ -26,13 +28,12 @@ def test_extract_single_column(fresh_db, table, fk_column):
         'CREATE TABLE "tree" (\n'
         '   "id" INTEGER PRIMARY KEY,\n'
         '   "name" TEXT,\n'
-        '   "{}" INTEGER REFERENCES "{}"("id"),\n'.format(expected_fk, expected_table)
+        f'   "{expected_fk}" INTEGER REFERENCES "{expected_table}"("id"),\n'
         + '   "end" INTEGER\n'
         + ")"
     )
     assert fresh_db[expected_table].schema == (
-        'CREATE TABLE "{}" (\n'.format(expected_table)
+        f'CREATE TABLE "{expected_table}" (\n' + '   "id" INTEGER PRIMARY KEY,\n'
-        + '   "id" INTEGER PRIMARY KEY,\n'
         '   "species" TEXT\n'
         ")"
     )
@@ -57,7 +58,7 @@ def test_extract_multiple_columns_with_rename(fresh_db):
         (
             {
                 "id": i,
-                "name": "Tree {}".format(i),
+                "name": f"Tree {i}",
                 "common_name": next(iter_common),
                 "latin_name": next(iter_latin),
             }

tests/test_extracts.py

@@ -1,13 +1,14 @@
-from sqlite_utils.db import Index
 import pytest
 
+from sqlite_utils.db import Index
+
 
 @pytest.mark.parametrize(
     "kwargs,expected_table",
     [
-        (dict(extracts={"species_id": "Species"}), "Species"),
+        ({"extracts": {"species_id": "Species"}}, "Species"),
-        (dict(extracts=["species_id"]), "species_id"),
+        ({"extracts": ["species_id"]}, "species_id"),
-        (dict(extracts=("species_id",)), "species_id"),
+        ({"extracts": ("species_id",)}, "species_id"),
     ],
 )
 @pytest.mark.parametrize("use_table_factory", [True, False])
@@ -30,15 +31,11 @@ def test_extracts(fresh_db, kwargs, expected_table, use_table_factory):
     # Should now have two tables: Trees and Species
     assert {expected_table, "Trees"} == set(fresh_db.table_names())
     assert (
-        'CREATE TABLE "{}" (\n   "id" INTEGER PRIMARY KEY,\n   "value" TEXT\n)'.format(
+        f'CREATE TABLE "{expected_table}" (\n   "id" INTEGER PRIMARY KEY,\n   "value" TEXT\n)'
-            expected_table
-        )
         == fresh_db[expected_table].schema
     )
     assert (
-        'CREATE TABLE "Trees" (\n   "id" INTEGER,\n   "species_id" INTEGER REFERENCES "{}"("id")\n)'.format(
+        f'CREATE TABLE "Trees" (\n   "id" INTEGER,\n   "species_id" INTEGER REFERENCES "{expected_table}"("id")\n)'
-            expected_table
-        )
         == fresh_db["Trees"].schema
     )
     # Should have a foreign key reference
@@ -51,7 +48,7 @@ def test_extracts(fresh_db, kwargs, expected_table, use_table_factory):
     assert [
         Index(
             seq=0,
-            name="idx_{}_value".format(expected_table),
+            name=f"idx_{expected_table}_value",
             unique=1,
             origin="c",
             partial=0,

tests/test_foreign_keys.py

@@ -1,6 +1,7 @@
 """Tests for compound (multi-column) foreign keys - issue #594."""
 
 import pytest
+
 from sqlite_utils import Database
 from sqlite_utils.db import AlterError, ForeignKey
 from sqlite_utils.utils import sqlite3
@@ -64,7 +65,7 @@ def test_foreign_key_no_longer_unpacks_as_tuple(fresh_db):
     fresh_db["books"].add_foreign_key("author_id", "authors", "id")
     fk = fresh_db["books"].foreign_keys[0]
     with pytest.raises(TypeError):
-        table, column, other_table, other_column = fk
+        _table, _column, _other_table, _other_column = fk
     with pytest.raises(TypeError):
         fk[0]
 

tests/test_fts.py

@@ -1,7 +1,9 @@
+from unittest.mock import ANY
+
 import pytest
+
 from sqlite_utils import Database
 from sqlite_utils.utils import sqlite3
-from unittest.mock import ANY
 
 search_records = [
     {
@@ -103,9 +105,10 @@ def test_search_limit_offset(fresh_db):
     table.enable_fts(["text", "country"], fts_version="FTS4")
     assert len(list(table.search("are"))) == 2
     assert len(list(table.search("are", limit=1))) == 1
-    assert list(table.search("are", limit=1, order_by="rowid"))[0]["rowid"] == 1
+    assert next(iter(table.search("are", limit=1, order_by="rowid")))["rowid"] == 1
     assert (
-        list(table.search("are", limit=1, offset=1, order_by="rowid"))[0]["rowid"] == 2
+        next(iter(table.search("are", limit=1, offset=1, order_by="rowid")))["rowid"]
+        == 2
     )
 
 
@@ -223,20 +226,20 @@ def test_populate_fts_escape_table_names(fresh_db):
 
 @pytest.mark.parametrize("fts_version", ("4", "5"))
 def test_fts_tokenize(fresh_db, fts_version):
-    table_name = "searchable_{}".format(fts_version)
+    table_name = f"searchable_{fts_version}"
     table = fresh_db[table_name]
     table.insert_all(search_records)
     # Test without porter stemming
     table.enable_fts(
         ["text", "country"],
-        fts_version="FTS{}".format(fts_version),
+        fts_version=f"FTS{fts_version}",
     )
     assert [] == list(table.search("bite"))
     # Test WITH stemming
     table.disable_fts()
     table.enable_fts(
         ["text", "country"],
-        fts_version="FTS{}".format(fts_version),
+        fts_version=f"FTS{fts_version}",
         tokenize="porter",
     )
     rows = list(table.search("bite", order_by="rowid"))
@@ -251,10 +254,10 @@ def test_fts_tokenize(fresh_db, fts_version):
 
 def test_optimize_fts(fresh_db):
     for fts_version in ("4", "5"):
-        table_name = "searchable_{}".format(fts_version)
+        table_name = f"searchable_{fts_version}"
         table = fresh_db[table_name]
         table.insert_all(search_records)
-        table.enable_fts(["text", "country"], fts_version="FTS{}".format(fts_version))
+        table.enable_fts(["text", "country"], fts_version=f"FTS{fts_version}")
     # You can call optimize successfully against the tables OR their _fts equivalents:
     for table_name in (
         "searchable_4",
@@ -310,12 +313,12 @@ def test_disable_fts(fresh_db, create_triggers):
         expected_triggers = {"searchable_ai", "searchable_ad", "searchable_au"}
     else:
         expected_triggers = set()
-    assert expected_triggers == set(
+    assert expected_triggers == {
         r[0]
         for r in fresh_db.execute(
             "select name from sqlite_master where type = 'trigger'"
         ).fetchall()
-    )
+    }
     # Now run .disable_fts() and confirm it worked
     table.disable_fts()
     assert (
@@ -424,7 +427,7 @@ def test_enable_fts_replace(kwargs):
     db["books"].enable_fts(**kwargs, replace=True)
     # Check that the new configuration is correct
     if should_have_changed_columns:
-        assert db["books_fts"].columns_dict.keys() == set(["title"])
+        assert db["books_fts"].columns_dict.keys() == {"title"}
     if "create_triggers" in kwargs:
         assert db["books"].triggers
     if "fts_version" in kwargs:
@@ -741,6 +744,7 @@ def test_enable_fts_cli_on_view_errors(tmpdir):
     db.create_view("v", "select * from t")
     db.close()
     from click.testing import CliRunner
+
     from sqlite_utils import cli as cli_module
 
     result = CliRunner().invoke(cli_module.cli, ["enable-fts", db_path, "v", "text"])

tests/test_get.py

@@ -1,4 +1,5 @@
 import pytest
+
 from sqlite_utils.db import NotFoundError
 
 

tests/test_gis.py

@@ -1,7 +1,8 @@
 import json
-import pytest
 
+import pytest
 from click.testing import CliRunner
+
 from sqlite_utils.cli import cli
 from sqlite_utils.db import Database
 from sqlite_utils.utils import find_spatialite, sqlite3
@@ -104,7 +105,7 @@ def test_query_load_extension(use_spatialite_shortcut):
         [
             ":memory:",
             "select spatialite_version()",
-            "--load-extension={}".format(load_extension),
+            f"--load-extension={load_extension}",
         ],
     )
     assert result.exit_code == 0, result.stdout

tests/test_hypothesis.py

@@ -1,5 +1,6 @@
-from hypothesis import given
 import hypothesis.strategies as st
+from hypothesis import given
+
 import sqlite_utils
 
 

tests/test_insert_files.py

@@ -1,10 +1,12 @@
-from sqlite_utils import cli, Database
-from click.testing import CliRunner
 import os
 import pathlib
-import pytest
 import sys
 
+import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import Database, cli
+
 
 @pytest.mark.parametrize("silent", (False, True))
 @pytest.mark.parametrize(
@@ -44,7 +46,7 @@ def test_insert_files(silent, pk_args, expected_pks):
         )
         cols = []
         for coltype in coltypes:
-            cols += ["-c", "{}:{}".format(coltype, coltype)]
+            cols += ["-c", f"{coltype}:{coltype}"]
         result = runner.invoke(
             cli.cli,
             ["insert-files", db_path, "files", str(tmpdir)]
@@ -142,7 +144,7 @@ def test_insert_files_stdin(use_text, encoding, input, expected):
         )
         assert result.exit_code == 0, result.stdout
         db = Database(db_path)
-        row = list(db["files"].rows)[0]
+        row = next(iter(db["files"].rows))
         key = "content"
         if use_text:
             key = "content_text"
@@ -167,5 +169,5 @@ def test_insert_files_bad_text_encoding_error():
         )
         assert result.exit_code == 1, result.output
         assert result.output.strip().startswith(
-            "Error: Could not read file '{}' as text".format(str(latin.resolve()))
+            f"Error: Could not read file '{latin.resolve()!s}' as text"
         )

tests/test_introspect.py

@@ -1,6 +1,7 @@
-from sqlite_utils.db import Index, View, Database, XIndex, XIndexColumn
 import pytest
 
+from sqlite_utils.db import Database, Index, View, XIndex, XIndexColumn
+
 
 def _check_supports_strict():
     """Check if SQLite supports strict tables without leaking the database."""
@@ -57,8 +58,8 @@ def test_detect_fts_similar_tables(fresh_db, reverse_order):
     fresh_db[table2].insert({"title": "Hello"}).enable_fts(
         ["title"], fts_version="FTS4"
     )
-    assert fresh_db[table1].detect_fts() == "{}_fts".format(table1)
+    assert fresh_db[table1].detect_fts() == f"{table1}_fts"
-    assert fresh_db[table2].detect_fts() == "{}_fts".format(table2)
+    assert fresh_db[table2].detect_fts() == f"{table2}_fts"
 
 
 def test_tables(existing_db):

tests/test_list_mode.py

@@ -3,6 +3,7 @@ Tests for list-based iteration in insert_all and upsert_all
 """
 
 import pytest
+
 from sqlite_utils import Database
 
 

tests/test_lookup.py

@@ -1,6 +1,7 @@
-from sqlite_utils.db import Index
 import pytest
 
+from sqlite_utils.db import Index
+
 
 def test_lookup_new_table(fresh_db):
     species = fresh_db["species"]

tests/test_m2m.py

@@ -1,6 +1,7 @@
-from sqlite_utils.db import ForeignKey, NoObviousTable
 import pytest
 
+from sqlite_utils.db import ForeignKey, NoObviousTable
+
 
 def test_insert_m2m_single(fresh_db):
     dogs = fresh_db["dogs"]
@@ -65,8 +66,7 @@ def test_insert_m2m_iterable(fresh_db):
     iterable_records = ({"id": 1, "name": "Phineas"}, {"id": 2, "name": "Ferb"})
 
     def iterable():
-        for record in iterable_records:
+        yield from iterable_records
-            yield record
 
     platypuses = fresh_db["platypuses"]
     platypuses.insert({"id": 1, "name": "Perry"}, pk="id").m2m(

tests/test_migrations.py

@@ -1,4 +1,5 @@
 import pytest
+
 import sqlite_utils
 from sqlite_utils import Migrations
 
@@ -154,10 +155,9 @@ def test_non_transactional_migration_allows_vacuum(tmpdir):
 
 def test_apply_composes_inside_outer_transaction(migrations):
     db = sqlite_utils.Database(memory=True)
-    with pytest.raises(ZeroDivisionError):
+    with pytest.raises(ZeroDivisionError), db.atomic():
-        with db.atomic():
+        migrations.apply(db)
-            migrations.apply(db)
+        raise ZeroDivisionError
-            raise ZeroDivisionError
     # The outer transaction rolled back, taking the migrations with it
     assert db.table_names() == []
 

tests/test_plugins.py

@@ -1,9 +1,12 @@
-from click.testing import CliRunner
-import click
 import importlib
-import pytest
+import sqlite3
 import sys
-from sqlite_utils import cli, Database, hookimpl, plugins
+
+import click
+import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import Database, cli, hookimpl, plugins
 
 
 def _supports_pragma_function_list():
@@ -11,7 +14,7 @@ def _supports_pragma_function_list():
     try:
         db.execute("select * from pragma_function_list()")
         return True
-    except Exception:
+    except sqlite3.DatabaseError:
         return False
     finally:
         db.close()

tests/test_query.py

@@ -1,6 +1,7 @@
-import pytest
 import types
 
+import pytest
+
 from sqlite_utils.utils import sqlite3
 
 

tests/test_recipes.py

@@ -1,7 +1,9 @@
+import json
+
+import pytest
+
 from sqlite_utils import recipes
 from sqlite_utils.utils import sqlite3
-import json
-import pytest
 
 
 @pytest.fixture

tests/test_recreate.py

@@ -1,8 +1,10 @@
-from sqlite_utils import Database
-import sqlite3
 import pathlib
+import sqlite3
+
 import pytest
 
+from sqlite_utils import Database
+
 
 def test_recreate_ignored_for_in_memory():
     # None of these should raise an exception:

tests/test_rows_from_file.py

@@ -1,7 +1,9 @@
-from sqlite_utils.utils import rows_from_file, Format, RowError
 from io import BytesIO, StringIO
+
 import pytest
 
+from sqlite_utils.utils import Format, RowError, rows_from_file
+
 
 @pytest.mark.parametrize(
     "input,expected_format",
@@ -29,7 +31,7 @@ def test_rows_from_file_detect_format(input, expected_format):
 )
 def test_rows_from_file_extra_fields_strategies(ignore_extras, extras_key, expected):
     try:
-        rows, format = rows_from_file(
+        rows, _format = rows_from_file(
             BytesIO(b"id,name\r\n1,Cleo,oops"),
             format=Format.CSV,
             ignore_extras=ignore_extras,

tests/test_sniff.py

@@ -1,7 +1,9 @@
-from sqlite_utils import cli, Database
-from click.testing import CliRunner
 import pathlib
+
 import pytest
+from click.testing import CliRunner
+
+from sqlite_utils import Database, cli
 
 sniff_dir = pathlib.Path(__file__).parent / "sniff"
 

tests/test_suggest_column_types.py

@@ -1,5 +1,7 @@
-import pytest
 from collections import OrderedDict
+
+import pytest
+
 from sqlite_utils.utils import suggest_column_types
 
 

tests/test_tracer.py

@@ -53,16 +53,18 @@ def test_with_tracer():
     assert len(collected) == 4
     assert collected == [
         (
-            "SELECT name FROM sqlite_master\n"
+            (
-            "    WHERE rootpage = 0\n"
+                "SELECT name FROM sqlite_master\n"
-            "    AND (\n"
+                "    WHERE rootpage = 0\n"
-            "        sql LIKE :like\n"
+                "    AND (\n"
-            "        OR sql LIKE :like2\n"
+                "        sql LIKE :like\n"
-            "        OR (\n"
+                "        OR sql LIKE :like2\n"
-            "            tbl_name = :table\n"
+                "        OR (\n"
-            "            AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
+                "            tbl_name = :table\n"
-            "        )\n"
+                "            AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
-            "    )",
+                "        )\n"
+                "    )"
+            ),
             {
                 "like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%",
                 "like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
@@ -72,21 +74,23 @@ def test_with_tracer():
         ("select name from sqlite_master where type = 'view'", None),
         ("select sql from sqlite_master where name = ?", ("dogs_fts",)),
         (
-            'with "original" as (\n'
+            (
-            "    select\n"
+                'with "original" as (\n'
-            "        rowid,\n"
+                "    select\n"
-            "        *\n"
+                "        rowid,\n"
-            '    from "dogs"\n'
+                "        *\n"
-            ")\n"
+                '    from "dogs"\n'
-            "select\n"
+                ")\n"
-            '    "original".*\n'
+                "select\n"
-            "from\n"
+                '    "original".*\n'
-            '    "original"\n'
+                "from\n"
-            '    join "dogs_fts" on "original".rowid = "dogs_fts".rowid\n'
+                '    "original"\n'
-            "where\n"
+                '    join "dogs_fts" on "original".rowid = "dogs_fts".rowid\n'
-            '    "dogs_fts" match :query\n'
+                "where\n"
-            "order by\n"
+                '    "dogs_fts" match :query\n'
-            '    "dogs_fts".rank',
+                "order by\n"
+                '    "dogs_fts".rank'
+            ),
             {"query": "Cleopaws"},
         ),
     ]

tests/test_transform.py

@@ -1,9 +1,10 @@
 import sqlite3
 
-from sqlite_utils.db import ForeignKey, TransformError
-from sqlite_utils.utils import OperationalError
 import pytest
 
+from sqlite_utils.db import ForeignKey, TransactionError, TransformError
+from sqlite_utils.utils import OperationalError
+
 
 @pytest.mark.parametrize(
     "params,expected_sql",
@@ -113,7 +114,7 @@ def test_transform_sql_table_with_primary_key(
     if use_pragma_foreign_keys:
         fresh_db.conn.execute("PRAGMA foreign_keys=ON")
     dogs.insert({"id": 1, "name": "Cleo", "age": "5"}, pk="id")
-    sql = dogs.transform_sql(**{**params, **{"tmp_suffix": "suffix"}})
+    sql = dogs.transform_sql(**{**params, "tmp_suffix": "suffix"})
     assert sql == expected_sql
     # Check that .transform() runs without exceptions:
     with fresh_db.tracer(tracer):
@@ -186,7 +187,7 @@ def test_transform_sql_table_with_no_primary_key(
     if use_pragma_foreign_keys:
         fresh_db.conn.execute("PRAGMA foreign_keys=ON")
     dogs.insert({"id": 1, "name": "Cleo", "age": "5"})
-    sql = dogs.transform_sql(**{**params, **{"tmp_suffix": "suffix"}})
+    sql = dogs.transform_sql(**{**params, "tmp_suffix": "suffix"})
     assert sql == expected_sql
     # Check that .transform() runs without exceptions:
     with fresh_db.tracer(tracer):
@@ -432,6 +433,163 @@ def test_transform_verify_foreign_keys(fresh_db):
     assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
 
 
+@pytest.mark.parametrize("use_pragma_foreign_keys", [False, True])
+def test_transform_on_delete_cascade_does_not_delete_records(
+    fresh_db, use_pragma_foreign_keys
+):
+    # Transforming a table drops and recreates it - if another table references
+    # it with ON DELETE CASCADE and PRAGMA foreign_keys is on, that drop must
+    # not cascade and delete the referencing records
+    if use_pragma_foreign_keys:
+        fresh_db.conn.execute("PRAGMA foreign_keys=ON")
+    fresh_db.executescript("""
+        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
+        CREATE TABLE books (
+            id INTEGER PRIMARY KEY,
+            title TEXT,
+            author_id INTEGER REFERENCES authors(id) ON DELETE CASCADE
+        );
+        """)
+    fresh_db["authors"].insert({"id": 1, "name": "Ursula K. Le Guin"})
+    fresh_db["books"].insert({"id": 1, "title": "The Dispossessed", "author_id": 1})
+    # Transform the table on the other end of the cascading foreign key
+    fresh_db["authors"].transform(rename={"name": "author_name"})
+    assert list(fresh_db["authors"].rows) == [
+        {"id": 1, "author_name": "Ursula K. Le Guin"}
+    ]
+    assert list(fresh_db["books"].rows) == [
+        {"id": 1, "title": "The Dispossessed", "author_id": 1}
+    ]
+    # Transforming the table with the cascading foreign key should not
+    # delete its records either
+    fresh_db["books"].transform(rename={"title": "book_title"})
+    assert list(fresh_db["books"].rows) == [
+        {"id": 1, "book_title": "The Dispossessed", "author_id": 1}
+    ]
+    if use_pragma_foreign_keys:
+        assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
+
+
+@pytest.mark.parametrize("on_delete", ["CASCADE", "SET NULL", "SET DEFAULT", "cascade"])
+def test_transform_in_transaction_refuses_destructive_on_delete(fresh_db, on_delete):
+    # PRAGMA foreign_keys is a no-op inside a transaction, so transforming a
+    # table referenced by ON DELETE CASCADE / SET NULL / SET DEFAULT foreign
+    # keys inside an open transaction would fire those actions when the old
+    # table is dropped - transform() should refuse instead
+    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
+    fresh_db.executescript(f"""
+        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
+        CREATE TABLE books (
+            id INTEGER PRIMARY KEY,
+            title TEXT,
+            author_id INTEGER REFERENCES authors(id) ON DELETE {on_delete}
+        );
+        """)
+    fresh_db["authors"].insert({"id": 1, "name": "Ursula K. Le Guin"})
+    fresh_db["books"].insert({"id": 1, "title": "The Dispossessed", "author_id": 1})
+    previous_schema = fresh_db["authors"].schema
+    with fresh_db.atomic(), pytest.raises(TransactionError) as excinfo:
+        fresh_db["authors"].transform(rename={"name": "author_name"})
+    message = str(excinfo.value)
+    assert "books" in message
+    assert f"ON DELETE {on_delete.upper()}" in message
+    # Nothing should have changed
+    assert fresh_db["authors"].schema == previous_schema
+    assert list(fresh_db["books"].rows) == [
+        {"id": 1, "title": "The Dispossessed", "author_id": 1}
+    ]
+    assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
+
+
+def test_transform_in_transaction_refuses_self_referential_cascade(fresh_db):
+    # The copied table carries a foreign key referencing the original table
+    # name, so a self-referential cascade would wipe the copy too
+    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
+    fresh_db.executescript("""
+        CREATE TABLE categories (
+            id INTEGER PRIMARY KEY,
+            name TEXT,
+            parent_id INTEGER REFERENCES categories(id) ON DELETE CASCADE
+        );
+        """)
+    fresh_db["categories"].insert_all(
+        [
+            {"id": 1, "name": "Fiction", "parent_id": None},
+            {"id": 2, "name": "Science Fiction", "parent_id": 1},
+        ]
+    )
+    with fresh_db.atomic(), pytest.raises(TransactionError) as excinfo:
+        fresh_db["categories"].transform(rename={"name": "title"})
+    assert "categories" in str(excinfo.value)
+    assert fresh_db["categories"].count == 2
+
+
+def test_transform_in_transaction_allowed_with_no_action_foreign_key(fresh_db):
+    # An inbound foreign key without a destructive ON DELETE action is safe
+    # inside a transaction thanks to PRAGMA defer_foreign_keys
+    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
+    fresh_db.executescript("""
+        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
+        CREATE TABLE books (
+            id INTEGER PRIMARY KEY,
+            title TEXT,
+            author_id INTEGER REFERENCES authors(id)
+        );
+        """)
+    fresh_db["authors"].insert({"id": 1, "name": "Ursula K. Le Guin"})
+    fresh_db["books"].insert({"id": 1, "title": "The Dispossessed", "author_id": 1})
+    with fresh_db.atomic():
+        fresh_db["authors"].transform(rename={"name": "author_name"})
+    assert list(fresh_db["authors"].rows) == [
+        {"id": 1, "author_name": "Ursula K. Le Guin"}
+    ]
+    assert list(fresh_db["books"].rows) == [
+        {"id": 1, "title": "The Dispossessed", "author_id": 1}
+    ]
+    assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
+
+
+def test_transform_in_transaction_allowed_for_child_table(fresh_db):
+    # The table being transformed only has an outbound foreign key - dropping
+    # it fires no ON DELETE actions, so this is allowed inside a transaction
+    fresh_db.conn.execute("PRAGMA foreign_keys=ON")
+    fresh_db.executescript("""
+        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
+        CREATE TABLE books (
+            id INTEGER PRIMARY KEY,
+            title TEXT,
+            author_id INTEGER REFERENCES authors(id) ON DELETE CASCADE
+        );
+        """)
+    fresh_db["authors"].insert({"id": 1, "name": "Ursula K. Le Guin"})
+    fresh_db["books"].insert({"id": 1, "title": "The Dispossessed", "author_id": 1})
+    with fresh_db.atomic():
+        fresh_db["books"].transform(rename={"title": "book_title"})
+    assert list(fresh_db["books"].rows) == [
+        {"id": 1, "book_title": "The Dispossessed", "author_id": 1}
+    ]
+
+
+def test_transform_in_transaction_allowed_with_foreign_keys_off(fresh_db):
+    # With PRAGMA foreign_keys off (the default) no cascades can fire, so
+    # transform inside a transaction is safe even with a CASCADE schema
+    fresh_db.executescript("""
+        CREATE TABLE authors (id INTEGER PRIMARY KEY, name TEXT);
+        CREATE TABLE books (
+            id INTEGER PRIMARY KEY,
+            title TEXT,
+            author_id INTEGER REFERENCES authors(id) ON DELETE CASCADE
+        );
+        """)
+    fresh_db["authors"].insert({"id": 1, "name": "Ursula K. Le Guin"})
+    fresh_db["books"].insert({"id": 1, "title": "The Dispossessed", "author_id": 1})
+    with fresh_db.atomic():
+        fresh_db["authors"].transform(rename={"name": "author_name"})
+    assert list(fresh_db["books"].rows) == [
+        {"id": 1, "title": "The Dispossessed", "author_id": 1}
+    ]
+
+
 def test_transform_add_foreign_keys_from_scratch(fresh_db):
     _add_country_city_continent(fresh_db)
     fresh_db["places"].insert(_CAVEAU)
@@ -556,15 +714,15 @@ def test_transform_preserves_rowids(fresh_db, table_type):
     # Now delete and insert a row to mix up the `rowid` sequence
     fresh_db["places"].delete_where("id = ?", ["2"])
     fresh_db["places"].insert({"id": "4", "name": "London", "country": "UK"})
-    previous_rows = list(
+    previous_rows = [
         tuple(row) for row in fresh_db.execute("select rowid, id, name from places")
-    )
+    ]
     # Transform it
     fresh_db["places"].transform(column_order=("country", "name"))
     # Should be the same
-    next_rows = list(
+    next_rows = [
         tuple(row) for row in fresh_db.execute("select rowid, id, name from places")
-    )
+    ]
     assert previous_rows == next_rows
 
 

tests/test_update.py

@@ -43,7 +43,7 @@ def test_update_compound_pk_table(fresh_db):
 )
 def test_update_invalid_pk(fresh_db, pk, update_pk):
     table = fresh_db["table"]
-    table.insert({"id1": 5, "id2": 3, "v": 1}, pk=pk).last_pk
+    table.insert({"id1": 5, "id2": 3, "v": 1}, pk=pk)
     with pytest.raises(NotFoundError):
         table.update(update_pk, {"v": 2})
 

tests/test_upsert.py

@@ -1,7 +1,8 @@
-from sqlite_utils.db import PrimaryKeyRequired
-from sqlite_utils import Database
 import pytest
 
+from sqlite_utils import Database
+from sqlite_utils.db import PrimaryKeyRequired
+
 
 @pytest.mark.parametrize("use_old_upsert", (False, True))
 def test_upsert(use_old_upsert):

tests/test_utils.py

@@ -1,8 +1,10 @@
-from sqlite_utils import utils
 import csv
 import io
+
 import pytest
 
+from sqlite_utils import utils
+
 
 @pytest.mark.parametrize(
     "input,expected,should_be_is",
@@ -57,7 +59,7 @@ def test_maximize_csv_field_size_limit():
     # Reset to default in case other tests have changed it
     csv.field_size_limit(utils.ORIGINAL_CSV_FIELD_SIZE_LIMIT)
     long_value = "a" * 131073
-    long_csv = "id,text\n1,{}".format(long_value)
+    long_csv = f"id,text\n1,{long_value}"
     fp = io.BytesIO(long_csv.encode("utf-8"))
     # Using rows_from_file should error
     with pytest.raises(csv.Error):

tests/test_wal.py

@@ -1,4 +1,5 @@
 import pytest
+
 from sqlite_utils import Database
 from sqlite_utils.db import TransactionError
 
@@ -11,7 +12,7 @@ def db_path_tmpdir(tmpdir):
 
 
 def test_enable_disable_wal(db_path_tmpdir):
-    db, path, tmpdir = db_path_tmpdir
+    db, _path, tmpdir = db_path_tmpdir
     assert len(tmpdir.listdir()) == 1
     assert "delete" == db.journal_mode
     assert "test.db-wal" not in [f.basename for f in tmpdir.listdir()]
@@ -25,12 +26,11 @@ def test_enable_disable_wal(db_path_tmpdir):
 
 
 def test_enable_wal_inside_transaction_raises(db_path_tmpdir):
-    db, path, tmpdir = db_path_tmpdir
+    db, _path, _tmpdir = db_path_tmpdir
     db["test"].insert({"id": 1}, pk="id")
-    with pytest.raises(TransactionError):
+    with pytest.raises(TransactionError), db.atomic():
-        with db.atomic():
+        db["test"].insert({"id": 2}, pk="id")
-            db["test"].insert({"id": 2}, pk="id")
+        db.enable_wal()
-            db.enable_wal()
     # The atomic() block must have rolled back cleanly and the
     # journal mode must be unchanged
     assert db.journal_mode == "delete"
@@ -38,19 +38,18 @@ def test_enable_wal_inside_transaction_raises(db_path_tmpdir):
 
 
 def test_disable_wal_inside_transaction_raises(db_path_tmpdir):
-    db, path, tmpdir = db_path_tmpdir
+    db, _path, _tmpdir = db_path_tmpdir
     db.enable_wal()
     db["test"].insert({"id": 1}, pk="id")
-    with pytest.raises(TransactionError):
+    with pytest.raises(TransactionError), db.atomic():
-        with db.atomic():
+        db["test"].insert({"id": 2}, pk="id")
-            db["test"].insert({"id": 2}, pk="id")
+        db.disable_wal()
-            db.disable_wal()
     assert db.journal_mode == "wal"
     assert [r["id"] for r in db["test"].rows] == [1]
 
 
 def test_ensure_autocommit_on(db_path_tmpdir):
-    db, path, tmpdir = db_path_tmpdir
+    db, _path, _tmpdir = db_path_tmpdir
     previous_isolation_level = db.conn.isolation_level
     assert previous_isolation_level is not None
     with db.ensure_autocommit_on():
@@ -63,7 +62,7 @@ def test_ensure_autocommit_on(db_path_tmpdir):
 def test_enable_wal_noop_inside_transaction_is_allowed(db_path_tmpdir):
     # Calling enable_wal() when WAL is already enabled is a no-op,
     # so it is fine inside a transaction
-    db, path, tmpdir = db_path_tmpdir
+    db, _path, _tmpdir = db_path_tmpdir
     db.enable_wal()
     with db.atomic():
         db["test"].insert({"id": 1}, pk="id")
@@ -75,13 +74,12 @@ def test_ensure_autocommit_on_inside_transaction_raises(db_path_tmpdir):
     # Setting isolation_level commits any pending transaction as a side
     # effect, silently breaking the caller's rollback guarantee - so
     # entering autocommit mode with a transaction open is an error
-    db, path, tmpdir = db_path_tmpdir
+    db, _path, _tmpdir = db_path_tmpdir
     db["test"].insert({"id": 1}, pk="id")
     db.begin()
     db.execute("insert into test (id) values (2)")
-    with pytest.raises(TransactionError):
+    with pytest.raises(TransactionError), db.ensure_autocommit_on():
-        with db.ensure_autocommit_on():
+        pass
-            pass
     # The transaction is still open and can still be rolled back
     assert db.conn.in_transaction
     db.rollback()