import base64
import csv as csv_std
import difflib
import hashlib
import inspect
import io
import itertools
import json
import os
import pathlib
import pdb
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,
    DescIndex,
    InvalidColumns,
    NoTable,
    NoView,
    PrimaryKeyRequired,
    quote_identifier,
)
from sqlite_utils.plugins import ensure_plugins_loaded, get_plugins, pm
from sqlite_utils.utils import maximize_csv_field_size_limit

from .utils import (
    Format,
    OperationalError,
    TypeTracker,
    _compile_code,
    chunks,
    decode_base64_values,
    dedupe_keys,
    file_progress,
    find_spatialite,
    progressbar,
    rows_from_file,
    sqlite3,
)
from .utils import (
    flatten as _flatten,
)

CONTEXT_SETTINGS = {"help_option_names": ["-h", "--help"]}


def _register_db_for_cleanup(db):
    """Register a database to be closed when the Click context is cleaned up."""
    ctx = click.get_current_context(silent=True)
    if ctx is None:
        return
    if "_databases_to_close" not in ctx.meta:
        ctx.meta["_databases_to_close"] = []
        ctx.call_on_close(lambda: _close_databases(ctx))
    ctx.meta["_databases_to_close"].append(db)


def _close_databases(ctx):
    """Close all databases registered for cleanup."""
    for db in ctx.meta.get("_databases_to_close", []):
        try:
            db.close()
        except Exception:
            pass


VALID_COLUMN_TYPES = ("INTEGER", "TEXT", "FLOAT", "REAL", "BLOB")

UNICODE_ERROR = """
{}

The input you provided uses a character encoding other than utf-8.

You can fix this by passing the --encoding= option with the encoding of the file.

If you do not know the encoding, running 'file filename.csv' may tell you.

It's often worth trying: --encoding=latin-1
""".strip()

maximize_csv_field_size_limit()


class CaseInsensitiveChoice(click.Choice):
    def __init__(self, choices):
        super().__init__([choice.lower() for choice in choices])

    def convert(self, value, param, ctx):
        return super().convert(value.lower(), param, ctx)


def output_options(fn):
    for decorator in reversed(
        (
            click.option(
                "--nl",
                help="Output newline-delimited JSON",
                is_flag=True,
                default=False,
            ),
            click.option(
                "--arrays",
                help="Output rows as arrays instead of objects",
                is_flag=True,
                default=False,
            ),
            click.option("--csv", is_flag=True, help="Output CSV"),
            click.option("--tsv", is_flag=True, help="Output TSV"),
            click.option(
                "--no-headers",
                is_flag=True,
                help="Omit headers from CSV/TSV and table/--fmt output",
            ),
            click.option(
                "-t", "--table", is_flag=True, help="Output as a formatted table"
            ),
            click.option(
                "--fmt",
                help="Table format - one of {}".format(
                    ", ".join(tabulate.tabulate_formats)
                ),
            ),
            click.option(
                "--json-cols",
                help="Detect JSON cols and output them as JSON, not escaped strings",
                is_flag=True,
                default=False,
            ),
            click.option(
                "--ascii",
                "ascii_",
                help="Escape non-ASCII characters in JSON output as \\uXXXX",
                is_flag=True,
                default=False,
            ),
        )
    ):
        fn = decorator(fn)
    return fn


def load_extension_option(fn):
    return click.option(
        "--load-extension",
        multiple=True,
        help="Path to SQLite extension, with optional :entrypoint",
    )(fn)


def functions_option(fn):
    return click.option(
        "--functions",
        help=(
            "Python code or a file path defining custom SQL functions; "
            "can be used multiple times"
        ),
        multiple=True,
    )(fn)


@click.group(
    cls=DefaultGroup,
    default="query",
    default_if_no_args=True,
    context_settings=CONTEXT_SETTINGS,
)
@click.version_option()
def cli():
    "Commands for interacting with a SQLite database"


@cli.command()
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.option(
    "--fts4", help="Just show FTS4 enabled tables", default=False, is_flag=True
)
@click.option(
    "--fts5", help="Just show FTS5 enabled tables", default=False, is_flag=True
)
@click.option(
    "--counts", help="Include row counts per table", default=False, is_flag=True
)
@output_options
@click.option(
    "--columns",
    help="Include list of columns for each table",
    is_flag=True,
    default=False,
)
@click.option(
    "--schema",
    help="Include schema for each table",
    is_flag=True,
    default=False,
)
@load_extension_option
def tables(
    path,
    fts4,
    fts5,
    counts,
    nl,
    arrays,
    csv,
    tsv,
    no_headers,
    table,
    fmt,
    json_cols,
    ascii_,
    columns,
    schema,
    load_extension,
    views=False,
):
    """List the tables in the database

    Example:

    \b
        sqlite-utils tables trees.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    headers = ["view" if views else "table"]
    if counts:
        headers.append("count")
    if columns:
        headers.append("columns")
    if schema:
        headers.append("schema")

    method = db.view if views else db.table

    def _iter():
        if views:
            items = db.view_names()
        else:
            items = db.table_names(fts4=fts4, fts5=fts5)
        for name in items:
            row: list[Any] = [name]
            if counts:
                row.append(method(name).count)
            if columns:
                cols = [c.name for c in method(name).columns]
                if csv:
                    row.append("\n".join(cols))
                else:
                    row.append(cols)
            if schema:
                row.append(method(name).schema)
            yield row

    if table or fmt:
        print(
            tabulate.tabulate(
                _iter(),
                headers=() if no_headers else headers,
                tablefmt=fmt or "simple",
            )
        )
    elif csv or tsv:
        writer = csv_std.writer(sys.stdout, dialect="excel-tab" if tsv else "excel")
        if not no_headers:
            writer.writerow(headers)
        for row in _iter():
            writer.writerow(row)
    else:
        for line in output_rows(_iter(), headers, nl, arrays, json_cols, ascii_):
            click.echo(line)


@cli.command()
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.option(
    "--counts", help="Include row counts per view", default=False, is_flag=True
)
@output_options
@click.option(
    "--columns",
    help="Include list of columns for each view",
    is_flag=True,
    default=False,
)
@click.option(
    "--schema",
    help="Include schema for each view",
    is_flag=True,
    default=False,
)
@load_extension_option
def views(
    path,
    counts,
    nl,
    arrays,
    csv,
    tsv,
    no_headers,
    table,
    fmt,
    json_cols,
    ascii_,
    columns,
    schema,
    load_extension,
):
    """List the views in the database

    Example:

    \b
        sqlite-utils views trees.db
    """
    assert tables.callback is not None
    tables.callback(
        path=path,
        fts4=False,
        fts5=False,
        counts=counts,
        nl=nl,
        arrays=arrays,
        csv=csv,
        tsv=tsv,
        no_headers=no_headers,
        table=table,
        fmt=fmt,
        json_cols=json_cols,
        ascii_=ascii_,
        columns=columns,
        schema=schema,
        load_extension=load_extension,
        views=True,
    )


@cli.command()
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("tables", nargs=-1)
@click.option("--no-vacuum", help="Don't run VACUUM", default=False, is_flag=True)
@load_extension_option
def optimize(path, tables, no_vacuum, load_extension):
    """Optimize all full-text search tables and then run VACUUM - should shrink the database file

    Example:

    \b
        sqlite-utils optimize chickens.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    if not tables:
        tables = db.table_names(fts4=True) + db.table_names(fts5=True)
    with db.conn:
        for table in tables:
            db.table(table).optimize()
    if not no_vacuum:
        db.vacuum()


@cli.command(name="rebuild-fts")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("tables", nargs=-1)
@load_extension_option
def rebuild_fts(path, tables, load_extension):
    """Rebuild all or specific full-text search tables

    Example:

    \b
        sqlite-utils rebuild-fts chickens.db chickens
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    if not tables:
        tables = db.table_names(fts4=True) + db.table_names(fts5=True)
    with db.conn:
        for table in tables:
            db.table(table).rebuild_fts()


@cli.command()
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("names", nargs=-1)
def analyze(path, names):
    """Run ANALYZE against the whole database, or against specific named indexes and tables

    Example:

    \b
        sqlite-utils analyze chickens.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    try:
        if names:
            for name in names:
                db.analyze(name)
        else:
            db.analyze()
    except OperationalError as e:
        raise click.ClickException(str(e))


@cli.command()
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
def vacuum(path):
    """Run VACUUM against the database

    Example:

    \b
        sqlite-utils vacuum chickens.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    db.vacuum()


@cli.command()
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@load_extension_option
def dump(path, load_extension):
    """Output a SQL dump of the schema and full contents of the database

    Example:

    \b
        sqlite-utils dump chickens.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    for line in db.iterdump():
        click.echo(line)


@cli.command(name="add-column")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("col_name")
@click.argument(
    "col_type",
    type=click.Choice(
        ["integer", "int", "float", "real", "text", "str", "blob", "bytes"],
        case_sensitive=False,
    ),
    required=False,
)
@click.option(
    "--fk", type=str, required=False, help="Table to reference as a foreign key"
)
@click.option(
    "--fk-col",
    type=str,
    required=False,
    help="Referenced column on that foreign key table - if omitted will automatically use the primary key",
)
@click.option(
    "--not-null-default",
    type=str,
    required=False,
    help="Add NOT NULL DEFAULT 'TEXT' constraint",
)
@click.option(
    "--ignore",
    is_flag=True,
    help="If column already exists, do nothing",
)
@load_extension_option
def add_column(
    path,
    table,
    col_name,
    col_type,
    fk,
    fk_col,
    not_null_default,
    ignore,
    load_extension,
):
    """Add a column to the specified table

    Example:

    \b
        sqlite-utils add-column chickens.db chickens weight float
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    try:
        db.table(table).add_column(
            col_name, col_type, fk=fk, fk_col=fk_col, not_null_default=not_null_default
        )
    except OperationalError as ex:
        if not ignore:
            raise click.ClickException(str(ex))


@cli.command(name="add-foreign-key")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("column")
@click.argument("other_table", required=False)
@click.argument("other_column", required=False)
@click.option(
    "--ignore",
    is_flag=True,
    help="If foreign key already exists, do nothing",
)
@load_extension_option
def add_foreign_key(
    path, table, column, other_table, other_column, ignore, load_extension
):
    """
    Add a new foreign key constraint to an existing table

    Example:

        sqlite-utils add-foreign-key my.db books author_id authors id
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    try:
        db.table(table).add_foreign_key(
            column, other_table, other_column, ignore=ignore
        )
    except AlterError as e:
        raise click.ClickException(str(e))


@cli.command(name="add-foreign-keys")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("foreign_key", nargs=-1)
@load_extension_option
def add_foreign_keys(path, foreign_key, load_extension):
    """
    Add multiple new foreign key constraints to a database

    Example:

    \b
        sqlite-utils add-foreign-keys my.db \\
            books author_id authors id \\
            authors country_id countries id
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    if len(foreign_key) % 4 != 0:
        raise click.ClickException(
            "Each foreign key requires four values: table, column, other_table, other_column"
        )
    tuples = []
    for i in range(len(foreign_key) // 4):
        tuples.append(tuple(foreign_key[i * 4 : (i * 4) + 4]))
    try:
        db.add_foreign_keys(tuples)
    except AlterError as e:
        raise click.ClickException(str(e))


@cli.command(name="index-foreign-keys")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@load_extension_option
def index_foreign_keys(path, load_extension):
    """
    Ensure every foreign key column has an index on it

    Example:

    \b
        sqlite-utils index-foreign-keys chickens.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    db.index_foreign_keys()


@cli.command(name="create-index")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("column", nargs=-1, required=True)
@click.option("--name", help="Explicit name for the new index")
@click.option("--unique", help="Make this a unique index", default=False, is_flag=True)
@click.option(
    "--if-not-exists",
    "--ignore",
    help="Ignore if index already exists",
    default=False,
    is_flag=True,
)
@click.option(
    "--analyze",
    help="Run ANALYZE after creating the index",
    is_flag=True,
)
@load_extension_option
def create_index(
    path, table, column, name, unique, if_not_exists, analyze, load_extension
):
    """
    Add an index to the specified table for the specified columns

    Example:

    \b
        sqlite-utils create-index chickens.db chickens name

    To create an index in descending order:

    \b
        sqlite-utils create-index chickens.db chickens -- -name
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    # Treat -prefix as descending for columns
    columns = []
    for col in column:
        if col.startswith("-"):
            col = DescIndex(col[1:])
        columns.append(col)
    db.table(table).create_index(
        columns,
        index_name=name,
        unique=unique,
        if_not_exists=if_not_exists,
        analyze=analyze,
    )


@cli.command(name="drop-index")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("index")
@click.option("--ignore", help="Ignore if index does not exist", is_flag=True)
@load_extension_option
def drop_index(path, table, index, ignore, load_extension):
    """
    Drop an index by index name from the specified table

    Example:

    \b
        sqlite-utils drop-index chickens.db chickens idx_chickens_name
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    try:
        db.table(table).drop_index(index, ignore=ignore)
    except OperationalError as ex:
        raise click.ClickException(str(ex))


@cli.command(name="enable-fts")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("column", nargs=-1, required=True)
@click.option("--fts4", help="Use FTS4", default=False, is_flag=True)
@click.option("--fts5", help="Use FTS5", default=False, is_flag=True)
@click.option("--tokenize", help="Tokenizer to use, e.g. porter")
@click.option(
    "--create-triggers",
    help="Create triggers to update the FTS tables when the parent table changes.",
    default=False,
    is_flag=True,
)
@click.option(
    "--replace",
    is_flag=True,
    help="Replace existing FTS configuration if it exists",
)
@load_extension_option
def enable_fts(
    path, table, column, fts4, fts5, tokenize, create_triggers, replace, load_extension
):
    """Enable full-text search for specific table and columns

    Example:

    \b
        sqlite-utils enable-fts chickens.db chickens name
    """
    fts_version = "FTS5"
    if fts4 and fts5:
        click.echo("Can only use one of --fts4 or --fts5", err=True)
        return
    elif fts4:
        fts_version = "FTS4"

    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    try:
        db.table(table).enable_fts(
            column,
            fts_version=fts_version,
            tokenize=tokenize,
            create_triggers=create_triggers,
            replace=replace,
        )
    except (NoTable, OperationalError) as ex:
        raise click.ClickException(str(ex))


@cli.command(name="populate-fts")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("column", nargs=-1, required=True)
@load_extension_option
def populate_fts(path, table, column, load_extension):
    """Re-populate full-text search for specific table and columns

    Example:

    \b
        sqlite-utils populate-fts chickens.db chickens name
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    db.table(table).populate_fts(column)


@cli.command(name="disable-fts")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@load_extension_option
def disable_fts(path, table, load_extension):
    """Disable full-text search for specific table

    Example:

    \b
        sqlite-utils disable-fts chickens.db chickens
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    db.table(table).disable_fts()


@cli.command(name="enable-wal")
@click.argument(
    "path",
    nargs=-1,
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@load_extension_option
def enable_wal(path, load_extension):
    """Enable WAL for database files

    Example:

    \b
        sqlite-utils enable-wal chickens.db
    """
    for path_ in path:
        db = sqlite_utils.Database(path_)
        _register_db_for_cleanup(db)
        _load_extensions(db, load_extension)
        db.enable_wal()


@cli.command(name="disable-wal")
@click.argument(
    "path",
    nargs=-1,
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@load_extension_option
def disable_wal(path, load_extension):
    """Disable WAL for database files

    Example:

    \b
        sqlite-utils disable-wal chickens.db
    """
    for path_ in path:
        db = sqlite_utils.Database(path_)
        _register_db_for_cleanup(db)
        _load_extensions(db, load_extension)
        db.disable_wal()


@cli.command(name="enable-counts")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("tables", nargs=-1)
@load_extension_option
def enable_counts(path, tables, load_extension):
    """Configure triggers to update a _counts table with row counts

    Example:

    \b
        sqlite-utils enable-counts chickens.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    if not tables:
        db.enable_counts()
    else:
        # Check all tables exist
        bad_tables = [table for table in tables if not db[table].exists()]
        if bad_tables:
            raise click.ClickException(f"Invalid tables: {bad_tables}")
        for table in tables:
            db.table(table).enable_counts()


@cli.command(name="reset-counts")
@click.argument(
    "path",
    type=click.Path(exists=True, file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@load_extension_option
def reset_counts(path, load_extension):
    """Reset calculated counts in the _counts table

    Example:

    \b
        sqlite-utils reset-counts chickens.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    db.reset_counts()


_import_options = (
    click.option(
        "--flatten",
        is_flag=True,
        help='Flatten nested JSON objects, so {"a": {"b": 1}} becomes {"a_b": 1}',
    ),
    click.option("--nl", is_flag=True, help="Expect newline-delimited JSON"),
    click.option("-c", "--csv", is_flag=True, help="Expect CSV input"),
    click.option("--tsv", is_flag=True, help="Expect TSV input"),
    click.option("--empty-null", is_flag=True, help="Treat empty strings as NULL"),
    click.option(
        "--lines",
        is_flag=True,
        help="Treat each line as a single value called 'line'",
    ),
    click.option(
        "--text",
        is_flag=True,
        help="Treat input as a single value called 'text'",
    ),
    click.option("--convert", help="Python code to convert each item"),
    click.option(
        "--import",
        "imports",
        type=str,
        multiple=True,
        help="Python modules to import",
    ),
    click.option("--delimiter", help="Delimiter to use for CSV files"),
    click.option("--quotechar", help="Quote character to use for CSV/TSV"),
    click.option("--sniff", is_flag=True, help="Detect delimiter and quote character"),
    click.option("--no-headers", is_flag=True, help="CSV file has no header row"),
    click.option(
        "--encoding",
        help="Character encoding for input, defaults to utf-8",
    ),
)


def import_options(fn):
    for decorator in reversed(_import_options):
        fn = decorator(fn)
    return fn


def insert_upsert_options(*, require_pk=False):
    def inner(fn):
        for decorator in reversed(
            (
                click.argument(
                    "path",
                    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
                    required=True,
                ),
                click.argument("table"),
                click.argument(
                    "file", type=click.File("rb", lazy=True), required=False
                ),
                click.option(
                    "--pk",
                    help="Columns to use as the primary key, e.g. id",
                    multiple=True,
                    required=require_pk,
                ),
                click.option(
                    "--code",
                    help="Python code defining a rows() function or iterable of rows to insert",
                ),
            )
            + _import_options
            + (
                click.option(
                    "--batch-size", type=int, default=100, help="Commit every X records"
                ),
                click.option("--stop-after", type=int, help="Stop after X records"),
                click.option(
                    "--alter",
                    is_flag=True,
                    help="Alter existing table to add any missing columns",
                ),
                click.option(
                    "--not-null",
                    multiple=True,
                    help="Columns that should be created as NOT NULL",
                ),
                click.option(
                    "--default",
                    multiple=True,
                    type=(str, str),
                    help="Default value that should be set for a column",
                ),
                click.option(
                    "--type",
                    "types",
                    type=(
                        str,
                        click.Choice(list(VALID_COLUMN_TYPES), case_sensitive=False),
                    ),
                    multiple=True,
                    help="Column types to use when creating the table",
                ),
                click.option(
                    "--no-detect-types",
                    is_flag=True,
                    help="Treat all CSV/TSV columns as TEXT",
                ),
                click.option(
                    "--analyze",
                    is_flag=True,
                    help="Run ANALYZE at the end of this operation",
                ),
                load_extension_option,
                click.option("--silent", is_flag=True, help="Do not show progress bar"),
                click.option(
                    "--strict",
                    is_flag=True,
                    default=False,
                    help="Apply STRICT mode to created table",
                ),
            )
        ):
            fn = decorator(fn)
        return fn

    return inner


def insert_upsert_implementation(
    path,
    table,
    file,
    pk,
    flatten,
    nl,
    csv,
    tsv,
    empty_null,
    lines,
    text,
    convert,
    imports,
    delimiter,
    quotechar,
    sniff,
    no_headers,
    encoding,
    batch_size,
    stop_after,
    alter,
    upsert,
    ignore=False,
    replace=False,
    truncate=False,
    not_null=None,
    default=None,
    types=None,
    no_detect_types=False,
    analyze=False,
    load_extension=None,
    silent=False,
    bulk_sql=None,
    functions=None,
    strict=False,
    code=None,
):
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    _maybe_register_functions(db, functions)
    column_type_overrides = {column: ctype.upper() for column, ctype in (types or [])}

    def _insert_docs(docs, tracker=None):
        extra_kwargs = {
            "ignore": ignore,
            "replace": replace,
            "truncate": truncate,
            "analyze": analyze,
            "strict": strict,
        }
        if not_null:
            extra_kwargs["not_null"] = set(not_null)
        if default:
            extra_kwargs["defaults"] = dict(default)
        if column_type_overrides:
            extra_kwargs["columns"] = column_type_overrides
        if upsert:
            extra_kwargs["upsert"] = upsert

        # docs should all be dictionaries
        docs = (verify_is_dict(doc) for doc in docs)

        # Apply {"$base64": true, ...} decoding, if needed
        docs = (decode_base64_values(doc) for doc in docs)

        # For bulk_sql= we use cursor.executemany() instead
        if bulk_sql:
            if batch_size:
                doc_chunks = chunks(docs, batch_size)
            else:
                doc_chunks = [docs]
            for doc_chunk in doc_chunks:
                with db.atomic():
                    db.conn.cursor().executemany(bulk_sql, doc_chunk)
            return

        # table_names() rather than db.table(), which raises NoTable for
        # views before the error handling below can deal with them
        table_existed_before_insert = table in db.table_names()
        try:
            db.table(table).insert_all(
                docs, pk=pk, batch_size=batch_size, alter=alter, **extra_kwargs
            )
        except (NoTable, InvalidColumns, PrimaryKeyRequired) as e:
            raise click.ClickException(str(e))
        except Exception as e:
            if (
                isinstance(e, OperationalError)
                and e.args
                and (
                    "has no column named" in e.args[0] or "no such column" in e.args[0]
                )
            ):
                raise click.ClickException(
                    f"{e.args[0]}\n\nTry using --alter to add additional columns"
                )
            # If we can find sql= and parameters= arguments, show those
            variables = _find_variables(e.__traceback__, ["sql", "parameters"])
            if "sql" in variables and "parameters" in variables:
                raise click.ClickException(
                    "{}\n\nsql = {}\nparameters = {}".format(
                        str(e), variables["sql"], variables["parameters"]
                    )
                )
            else:
                raise
        # Apply detected types only to a table this command created -
        # transforming a pre-existing table would rewrite its column types
        # and corrupt values such as TEXT zip codes with leading zeros
        if (
            tracker is not None
            and not table_existed_before_insert
            and db.table(table).exists()
        ):
            detected_types = tracker.types
            detected_types.update(column_type_overrides)
            db.table(table).transform(types=detected_types)

    if code is not None:
        if file is not None:
            raise click.ClickException("--code cannot be used with a FILE argument")
        if any(
            [
                flatten,
                nl,
                csv,
                tsv,
                empty_null,
                lines,
                text,
                convert,
                sniff,
                no_headers,
                delimiter,
                quotechar,
                encoding,
            ]
        ):
            raise click.ClickException(
                "--code cannot be used with input format options"
            )
        _insert_docs(_rows_from_code(code))
        return

    if file is None:
        raise click.ClickException(
            "Provide either a FILE argument or --code to specify rows to insert"
        )

    if (delimiter or quotechar or sniff or no_headers) and not tsv:
        csv = True
    if (nl + csv + tsv) >= 2:
        raise click.ClickException("Use just one of --nl, --csv or --tsv")
    if (csv or tsv) and flatten:
        raise click.ClickException("--flatten cannot be used with --csv or --tsv")
    if empty_null and not (csv or tsv):
        raise click.ClickException("--empty-null can only be used with --csv or --tsv")
    if encoding and not (csv or tsv):
        raise click.ClickException("--encoding must be used with --csv or --tsv")
    if pk and len(pk) == 1:
        pk = pk[0]
    encoding = encoding or "utf-8-sig"

    # The --sniff option needs us to buffer the file to peek ahead
    sniff_buffer = None
    decoded_buffer = None
    if sniff:
        sniff_buffer = io.BufferedReader(file, buffer_size=4096)
        decoded_buffer = io.TextIOWrapper(sniff_buffer, encoding=encoding)
    else:
        decoded_buffer = io.TextIOWrapper(file, encoding=encoding)

    tracker = None
    with file_progress(decoded_buffer, silent=silent) as decoded:
        if csv or tsv:
            if sniff:
                # Read first 2048 bytes and use that to detect
                assert sniff_buffer is not None
                first_bytes = sniff_buffer.peek(2048)
                dialect = csv_std.Sniffer().sniff(
                    first_bytes.decode(encoding, "ignore")
                )
            else:
                dialect = "excel-tab" if tsv else "excel"
            csv_reader_args: dict[str, Any] = {"dialect": dialect}
            if delimiter:
                csv_reader_args["delimiter"] = delimiter
            if quotechar:
                csv_reader_args["quotechar"] = quotechar
            reader = csv_std.reader(decoded, **csv_reader_args)  # type: ignore
            first_row = next(reader)
            if no_headers:
                headers = [f"untitled_{i + 1}" for i in range(len(first_row))]
                reader = itertools.chain([first_row], reader)
            else:
                headers = first_row
            if empty_null:
                docs = (
                    dict(zip(headers, [None if cell == "" else cell for cell in row]))
                    for row in reader
                )
            else:
                docs = (dict(zip(headers, row)) for row in reader)
            # Type detection is the default, unless --no-detect-types is passed
            if not no_detect_types:
                tracker = TypeTracker()
                docs = tracker.wrap(docs)
        elif lines:
            docs = ({"line": line.strip()} for line in decoded)
        elif text:
            docs = ({"text": decoded.read()},)
        else:
            try:
                if nl:
                    docs = (json.loads(line) for line in decoded if line.strip())
                else:
                    docs = json.load(decoded)
                    if isinstance(docs, dict):
                        docs = [docs]
            except json.decoder.JSONDecodeError as ex:
                raise click.ClickException(
                    f"Invalid JSON - use --csv for CSV or --tsv for TSV files\n\nJSON error: {ex}"
                )
            if flatten:
                docs = (_flatten(doc) for doc in docs)

        if stop_after:
            docs = itertools.islice(docs, stop_after)

        if convert:
            variable = "row"
            if lines:
                variable = "line"
            elif text:
                variable = "text"
            fn = _compile_code(convert, imports, variable=variable)
            if lines:
                docs = (fn(doc["line"]) for doc in docs)
            elif text:
                # Special case: this is allowed to be an iterable
                text_value = next(iter(docs))["text"]
                fn_return = fn(text_value)
                if isinstance(fn_return, dict):
                    docs = [fn_return]
                else:
                    try:
                        docs = iter(fn_return)
                    except TypeError:
                        raise click.ClickException(
                            "--convert must return dict or iterator"
                        )
            else:
                docs = (fn(doc) or doc for doc in docs)

        _insert_docs(docs, tracker=tracker)

        # Clean up open file-like objects
        if sniff_buffer:
            sniff_buffer.close()
        if decoded_buffer:
            decoded_buffer.close()


def _find_variables(tb, vars):
    to_find = list(vars)
    found = {}
    for var in to_find:
        if var in tb.tb_frame.f_locals:
            vars.remove(var)
            found[var] = tb.tb_frame.f_locals[var]
    if vars and tb.tb_next:
        found.update(_find_variables(tb.tb_next, vars))
    return found


@cli.command()
@insert_upsert_options()
@click.option(
    "--ignore", is_flag=True, default=False, help="Ignore records if pk already exists"
)
@click.option(
    "--replace",
    is_flag=True,
    default=False,
    help="Replace records if pk already exists",
)
@click.option(
    "--truncate",
    is_flag=True,
    default=False,
    help="Truncate table before inserting records, if table already exists",
)
def insert(
    path,
    table,
    file,
    pk,
    code,
    flatten,
    nl,
    csv,
    tsv,
    empty_null,
    lines,
    text,
    convert,
    imports,
    delimiter,
    quotechar,
    sniff,
    no_headers,
    encoding,
    batch_size,
    stop_after,
    alter,
    no_detect_types,
    analyze,
    load_extension,
    silent,
    ignore,
    replace,
    truncate,
    not_null,
    default,
    types,
    strict,
):
    """
    Insert records from FILE into a table, creating the table if it
    does not already exist.

    Example:

        echo '{"name": "Lila"}' | sqlite-utils insert data.db chickens -

    By default the input is expected to be a JSON object or array of objects.

    \b
    - Use --nl for newline-delimited JSON objects
    - Use --csv or --tsv for comma-separated or tab-separated input
    - Use --lines to write each incoming line to a column called "line"
    - Use --text to write the entire input to a column called "text"

    Use --type column-name type to override the type automatically chosen
    when the table is created.

    You can also use --convert to pass a fragment of Python code that will
    be used to convert each input.

    Your Python code will be passed a "row" variable representing the
    imported row, and can return a modified row.

    This example uses just the name, latitude and longitude columns from
    a CSV file, converting name to upper case and latitude and longitude
    to floating point numbers:

    \b
        sqlite-utils insert plants.db plants plants.csv --csv --convert '
          return {
            "name": row["name"].upper(),
            "latitude": float(row["latitude"]),
            "longitude": float(row["longitude"]),
          }'

    If you are using --lines your code will be passed a "line" variable,
    and for --text a "text" variable.

    When using --text your function can return an iterator of rows to
    insert. This example inserts one record per word in the input:

    \b
        echo 'A bunch of words' | sqlite-utils insert words.db words - \\
          --text --convert '({"word": w} for w in text.split())'

    Instead of a FILE you can use --code to provide a block of Python code
    that defines the rows to insert, as either a rows() function that yields
    dictionaries or a "rows" iterable. --code can also be a path to a .py file:

    \b
        sqlite-utils insert data.db creatures --code '
        def rows():
            yield {"id": 1, "name": "Cleo"}
            yield {"id": 2, "name": "Suna"}
        ' --pk id
    """
    try:
        insert_upsert_implementation(
            path,
            table,
            file,
            pk,
            flatten,
            nl,
            csv,
            tsv,
            empty_null,
            lines,
            text,
            convert,
            imports,
            delimiter,
            quotechar,
            sniff,
            no_headers,
            encoding,
            batch_size,
            stop_after,
            alter=alter,
            upsert=False,
            ignore=ignore,
            replace=replace,
            truncate=truncate,
            no_detect_types=no_detect_types,
            analyze=analyze,
            load_extension=load_extension,
            silent=silent,
            not_null=not_null,
            default=default,
            types=types,
            strict=strict,
            code=code,
        )
    except UnicodeDecodeError as ex:
        raise click.ClickException(UNICODE_ERROR.format(ex))


@cli.command()
@insert_upsert_options()
def upsert(
    path,
    table,
    file,
    pk,
    code,
    flatten,
    nl,
    csv,
    tsv,
    empty_null,
    lines,
    text,
    convert,
    imports,
    batch_size,
    stop_after,
    delimiter,
    quotechar,
    sniff,
    no_headers,
    encoding,
    alter,
    not_null,
    default,
    types,
    no_detect_types,
    analyze,
    load_extension,
    silent,
    strict,
):
    """
    Upsert records based on their primary key. Works like 'insert' but if
    an incoming record has a primary key that matches an existing record
    the existing record will be updated.

    If the table already exists and has a primary key, --pk can be omitted.

    Use --type column-name type to override the type automatically chosen
    when the table is created.

    Example:

    \b
        echo '[
            {"id": 1, "name": "Lila"},
            {"id": 2, "name": "Suna"}
        ]' | sqlite-utils upsert data.db chickens - --pk id
    """
    try:
        insert_upsert_implementation(
            path,
            table,
            file,
            pk,
            flatten,
            nl,
            csv,
            tsv,
            empty_null,
            lines,
            text,
            convert,
            imports,
            delimiter,
            quotechar,
            sniff,
            no_headers,
            encoding,
            batch_size,
            stop_after,
            alter=alter,
            upsert=True,
            not_null=not_null,
            default=default,
            types=types,
            no_detect_types=no_detect_types,
            analyze=analyze,
            load_extension=load_extension,
            silent=silent,
            strict=strict,
            code=code,
        )
    except UnicodeDecodeError as ex:
        raise click.ClickException(UNICODE_ERROR.format(ex))


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("sql")
@click.argument("file", type=click.File("rb"), required=True)
@click.option("--batch-size", type=int, default=100, help="Commit every X records")
@functions_option
@import_options
@load_extension_option
def bulk(
    path,
    sql,
    file,
    batch_size,
    functions,
    flatten,
    nl,
    csv,
    tsv,
    empty_null,
    lines,
    text,
    convert,
    imports,
    delimiter,
    quotechar,
    sniff,
    no_headers,
    encoding,
    load_extension,
):
    """
    Execute parameterized SQL against the provided list of documents.

    Example:

    \b
        echo '[
            {"id": 1, "name": "Lila2"},
            {"id": 2, "name": "Suna2"}
        ]' | sqlite-utils bulk data.db '
            update chickens set name = :name where id = :id
        ' -
    """
    try:
        insert_upsert_implementation(
            path=path,
            table=None,
            file=file,
            pk=None,
            flatten=flatten,
            nl=nl,
            csv=csv,
            tsv=tsv,
            empty_null=empty_null,
            lines=lines,
            text=text,
            convert=convert,
            imports=imports,
            delimiter=delimiter,
            quotechar=quotechar,
            sniff=sniff,
            no_headers=no_headers,
            encoding=encoding,
            batch_size=batch_size,
            stop_after=None,
            alter=False,
            upsert=False,
            not_null=set(),
            default={},
            no_detect_types=True,
            load_extension=load_extension,
            silent=False,
            bulk_sql=sql,
            functions=functions,
        )
    except (OperationalError, sqlite3.IntegrityError) as e:
        raise click.ClickException(str(e))


@cli.command(name="create-database")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.option(
    "--enable-wal", is_flag=True, help="Enable WAL mode on the created database"
)
@click.option(
    "--init-spatialite", is_flag=True, help="Enable SpatiaLite on the created database"
)
@load_extension_option
def create_database(path, enable_wal, init_spatialite, load_extension):
    """Create a new empty database file

    Example:

    \b
        sqlite-utils create-database trees.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    if enable_wal:
        db.enable_wal()

    # load spatialite or another extension from a custom location
    if load_extension:
        _load_extensions(db, load_extension)

    # load spatialite from expected locations and initialize metadata
    if init_spatialite:
        db.init_spatialite()

    db.vacuum()


@cli.command(name="create-table")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("columns", nargs=-1, required=True)
@click.option("pks", "--pk", help="Column to use as primary key", multiple=True)
@click.option(
    "--not-null",
    multiple=True,
    help="Columns that should be created as NOT NULL",
)
@click.option(
    "--default",
    multiple=True,
    type=(str, str),
    help="Default value that should be set for a column",
)
@click.option(
    "--fk",
    multiple=True,
    type=(str, str, str),
    help="Column, other table, other column to set as a foreign key",
)
@click.option(
    "--ignore",
    is_flag=True,
    help="If table already exists, do nothing",
)
@click.option(
    "--replace",
    is_flag=True,
    help="If table already exists, replace it",
)
@click.option(
    "--transform",
    is_flag=True,
    help="If table already exists, try to transform the schema",
)
@load_extension_option
@click.option(
    "--strict",
    is_flag=True,
    help="Apply STRICT mode to created table",
)
def create_table(
    path,
    table,
    columns,
    pks,
    not_null,
    default,
    fk,
    ignore,
    replace,
    transform,
    load_extension,
    strict,
):
    """
    Add a table with the specified columns. Columns should be specified using
    name, type pairs, for example:

    \b
        sqlite-utils create-table my.db people \\
            id integer \\
            name text \\
            height real \\
            photo blob --pk id

    Valid column types are text, integer, real, float and blob.
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    if len(columns) % 2 == 1:
        raise click.ClickException(
            "columns must be an even number of 'name' 'type' pairs"
        )
    coltypes = {}
    columns = list(columns)
    while columns:
        name = columns.pop(0)
        ctype = columns.pop(0)
        if ctype.upper() not in VALID_COLUMN_TYPES:
            raise click.ClickException(
                f"column types must be one of {VALID_COLUMN_TYPES}"
            )
        coltypes[name] = ctype.upper()
    # Does table already exist?
    if table in db.table_names() and not ignore and not replace and not transform:
        raise click.ClickException(
            f'Table "{table}" already exists. Use --replace to delete and replace it.'
        )
    db.table(table).create(
        coltypes,
        pk=pks[0] if len(pks) == 1 else pks,
        not_null=not_null,
        defaults=dict(default),
        foreign_keys=fk,
        ignore=ignore,
        replace=replace,
        transform=transform,
        strict=strict,
    )


@cli.command(name="duplicate")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("new_table")
@click.option("--ignore", is_flag=True, help="If table does not exist, do nothing")
@load_extension_option
def duplicate(path, table, new_table, ignore, load_extension):
    """
    Create a duplicate of this table, copying across the schema and all row data.
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    try:
        db.table(table).duplicate(new_table)
    except NoTable:
        if not ignore:
            raise click.ClickException(f'Table "{table}" does not exist')


@cli.command(name="rename-table")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("new_name")
@click.option("--ignore", is_flag=True, help="If table does not exist, do nothing")
@load_extension_option
def rename_table(path, table, new_name, ignore, load_extension):
    """
    Rename this table.
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    try:
        db.rename_table(table, new_name)
    except sqlite3.OperationalError as ex:
        if not ignore:
            raise click.ClickException(
                f'Table "{table}" could not be renamed. {ex!s}'
            )


@cli.command(name="drop-table")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.option("--ignore", is_flag=True, help="If table does not exist, do nothing")
@load_extension_option
def drop_table(path, table, ignore, load_extension):
    """Drop the specified table

    Example:

    \b
        sqlite-utils drop-table chickens.db chickens
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    try:
        db.table(table).drop(ignore=ignore)
    except NoTable:
        # A view exists with this name
        if not ignore:
            raise click.ClickException(
                f'"{table}" is a view, not a table - use drop-view to drop it'
            )
    except OperationalError:
        raise click.ClickException(f'Table "{table}" does not exist')


@cli.command(name="create-view")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("view")
@click.argument("select")
@click.option(
    "--ignore",
    is_flag=True,
    help="If view already exists, do nothing",
)
@click.option(
    "--replace",
    is_flag=True,
    help="If view already exists, replace it",
)
@load_extension_option
def create_view(path, view, select, ignore, replace, load_extension):
    """Create a view for the provided SELECT query

    Example:

    \b
        sqlite-utils create-view chickens.db heavy_chickens \\
          'select * from chickens where weight > 3'
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    # Does view already exist?
    if view in db.view_names():
        if ignore:
            return
        elif replace:
            db.view(view).drop()
        else:
            raise click.ClickException(
                f'View "{view}" already exists. Use --replace to delete and replace it.'
            )
    db.create_view(view, select)


@cli.command(name="drop-view")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("view")
@click.option("--ignore", is_flag=True, help="If view does not exist, do nothing")
@load_extension_option
def drop_view(path, view, ignore, load_extension):
    """Drop the specified view

    Example:

    \b
        sqlite-utils drop-view chickens.db heavy_chickens
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    try:
        db.view(view).drop(ignore=ignore)
    except NoView:
        if ignore:
            return
        if view in db.table_names():
            raise click.ClickException(
                f'"{view}" is a table, not a view - use drop-table to drop it'
            )
        raise click.ClickException(f'View "{view}" does not exist')


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("sql")
@click.option(
    "--attach",
    type=(str, click.Path(file_okay=True, dir_okay=False, allow_dash=False)),
    multiple=True,
    help="Additional databases to attach - specify alias and filepath",
)
@output_options
@click.option("-r", "--raw", is_flag=True, help="Raw output, first column of first row")
@click.option("--raw-lines", is_flag=True, help="Raw output, first column of each row")
@click.option(
    "-p",
    "--param",
    multiple=True,
    type=(str, str),
    help="Named :parameters for SQL query",
)
@functions_option
@load_extension_option
def query(
    path,
    sql,
    attach,
    nl,
    arrays,
    csv,
    tsv,
    no_headers,
    table,
    fmt,
    json_cols,
    ascii_,
    raw,
    raw_lines,
    param,
    load_extension,
    functions,
):
    """Execute SQL query and return the results as JSON

    Example:

    \b
        sqlite-utils data.db \\
            "select * from chickens where age > :age" \\
            -p age 1

    Pass "-" as the SQL to read the query from standard input:

    \b
        echo "select * from chickens" | sqlite-utils data.db -
    """
    if sql == "-":
        # Read SQL from standard input
        sql = sys.stdin.read()
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    for alias, attach_path in attach:
        db.attach(alias, attach_path)
    _load_extensions(db, load_extension)
    db.register_fts4_bm25()

    _maybe_register_functions(db, functions)

    _execute_query(
        db,
        sql,
        param,
        raw,
        raw_lines,
        table,
        csv,
        tsv,
        no_headers,
        fmt,
        nl,
        arrays,
        json_cols,
        ascii_,
    )


@cli.command()
@click.argument(
    "paths",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=True),
    required=False,
    nargs=-1,
)
@click.argument("sql")
@functions_option
@click.option(
    "--attach",
    type=(str, click.Path(file_okay=True, dir_okay=False, allow_dash=False)),
    multiple=True,
    help="Additional databases to attach - specify alias and filepath",
)
@click.option(
    "--flatten",
    is_flag=True,
    help='Flatten nested JSON objects, so {"foo": {"bar": 1}} becomes {"foo_bar": 1}',
)
@output_options
@click.option("-r", "--raw", is_flag=True, help="Raw output, first column of first row")
@click.option("--raw-lines", is_flag=True, help="Raw output, first column of each row")
@click.option(
    "-p",
    "--param",
    multiple=True,
    type=(str, str),
    help="Named :parameters for SQL query",
)
@click.option(
    "--encoding",
    help="Character encoding for CSV input, defaults to utf-8",
)
@click.option(
    "-n",
    "--no-detect-types",
    is_flag=True,
    help="Treat all CSV/TSV columns as TEXT",
)
@click.option("--schema", is_flag=True, help="Show SQL schema for in-memory database")
@click.option("--dump", is_flag=True, help="Dump SQL for in-memory database")
@click.option(
    "--save",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    help="Save in-memory database to this file",
)
@click.option(
    "--analyze",
    is_flag=True,
    help="Analyze resulting tables and output results",
)
@load_extension_option
def memory(
    paths,
    sql,
    functions,
    attach,
    flatten,
    nl,
    arrays,
    csv,
    tsv,
    no_headers,
    table,
    fmt,
    json_cols,
    ascii_,
    raw,
    raw_lines,
    param,
    encoding,
    no_detect_types,
    schema,
    dump,
    save,
    analyze,
    load_extension,
    return_db=False,
):
    """Execute SQL query against an in-memory database, optionally populated by imported data

    To import data from CSV, TSV or JSON files pass them on the command-line:

    \b
        sqlite-utils memory one.csv two.json \\
            "select * from one join two on one.two_id = two.id"

    For data piped into the tool from standard input, use "-" or "stdin":

    \b
        cat animals.csv | sqlite-utils memory - \\
            "select * from stdin where species = 'dog'"

    The format of the data will be automatically detected. You can specify the format
    explicitly using :json, :csv, :tsv or :nl (for newline-delimited JSON) - for example:

    \b
        cat animals.csv | sqlite-utils memory stdin:csv places.dat:nl \\
            "select * from stdin where place_id in (select id from places)"

    Use --schema to view the SQL schema of any imported files:

    \b
        sqlite-utils memory animals.csv --schema
    """
    db = sqlite_utils.Database(memory=True)
    if not return_db:
        _register_db_for_cleanup(db)

    # If --dump or --save or --analyze used but no paths detected, assume SQL query is a path:
    if (dump or save or schema or analyze) and not paths:
        paths = [sql]
        sql = None
    stem_counts = {}
    for i, path in enumerate(paths):
        # Path may have a :format suffix
        fp = None
        should_close_fp = False
        if ":" in path and path.rsplit(":", 1)[-1].upper() in Format.__members__:
            path, suffix = path.rsplit(":", 1)
            format = Format[suffix.upper()]
        else:
            format = None
        if path in ("-", "stdin"):
            fp = sys.stdin.buffer
            file_table = "stdin"
        else:
            file_path = pathlib.Path(path)
            stem = file_path.stem
            if stem_counts.get(stem):
                file_table = f"{stem}_{stem_counts[stem]}"
            else:
                file_table = stem
            stem_counts[stem] = stem_counts.get(stem, 1) + 1
            fp = file_path.open("rb")
            should_close_fp = True
        try:
            rows, format_used = rows_from_file(fp, format=format, encoding=encoding)
            tracker = None
            if format_used in (Format.CSV, Format.TSV) and not no_detect_types:
                tracker = TypeTracker()
                rows = tracker.wrap(rows)
            if flatten:
                rows = (_flatten(row) for row in rows)

            db.table(file_table).insert_all(rows, alter=True)
            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 = [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,
                        f"select * from {quote_identifier(file_table)}",
                    )
        finally:
            if should_close_fp and fp:
                fp.close()

    if analyze:
        _analyze(db, tables=None, columns=None, save=False)
        return

    if dump:
        for line in db.iterdump():
            click.echo(line)
        return

    if schema:
        click.echo(db.schema)
        return

    if save:
        db2 = sqlite_utils.Database(save)
        _register_db_for_cleanup(db2)
        for line in db.iterdump():
            db2.execute(line)
        return

    for alias, attach_path in attach:
        db.attach(alias, attach_path)
    _load_extensions(db, load_extension)
    db.register_fts4_bm25()

    _maybe_register_functions(db, functions)

    if return_db:
        return db

    _execute_query(
        db,
        sql,
        param,
        raw,
        raw_lines,
        table,
        csv,
        tsv,
        no_headers,
        fmt,
        nl,
        arrays,
        json_cols,
        ascii_,
    )


def _execute_query(
    db,
    sql,
    param,
    raw,
    raw_lines,
    table,
    csv,
    tsv,
    no_headers,
    fmt,
    nl,
    arrays,
    json_cols,
    ascii_,
):
    with db.conn:
        try:
            cursor = db.execute(sql, dict(param))
        except OperationalError as e:
            raise click.ClickException(str(e))
        if cursor.description is None:
            # This was an update/insert
            headers = ["rows_affected"]
            cursor = [[cursor.rowcount]]
        else:
            headers = [c[0] for c in cursor.description]
        cursor_or_rows: Any = cursor
        if raw:
            row = cursor_or_rows.fetchone()
            data = row[0] if row else None
            if isinstance(data, bytes):
                sys.stdout.buffer.write(data)
            else:
                sys.stdout.write(str(data))
        elif raw_lines:
            for row in cursor:
                data = row[0]
                if isinstance(data, bytes):
                    sys.stdout.buffer.write(data + b"\n")
                else:
                    sys.stdout.write(str(data) + "\n")
        elif fmt or table:
            print(
                tabulate.tabulate(
                    list(cursor),
                    headers=() if no_headers else headers,
                    tablefmt=fmt or "simple",
                )
            )
        elif csv or tsv:
            writer = csv_std.writer(sys.stdout, dialect="excel-tab" if tsv else "excel")
            if not no_headers:
                writer.writerow(headers)
            for row in cursor:
                writer.writerow(row)
        else:
            for line in output_rows(cursor, headers, nl, arrays, json_cols, ascii_):
                click.echo(line)


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("dbtable")
@click.argument("q")
@click.option("-o", "--order", type=str, help="Order by ('column' or 'column desc')")
@click.option("-c", "--column", type=str, multiple=True, help="Columns to return")
@click.option(
    "--limit",
    type=int,
    help="Number of rows to return - defaults to everything",
)
@click.option(
    "--sql", "show_sql", is_flag=True, help="Show SQL query that would be run"
)
@click.option("--quote", is_flag=True, help="Apply FTS quoting rules to search term")
@output_options
@load_extension_option
@click.pass_context
def search(
    ctx,
    path,
    dbtable,
    q,
    order,
    show_sql,
    quote,
    column,
    limit,
    nl,
    arrays,
    csv,
    tsv,
    no_headers,
    table,
    fmt,
    json_cols,
    ascii_,
    load_extension,
):
    """Execute a full-text search against this table

    Example:

        sqlite-utils search data.db chickens lila
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    # Check table exists
    table_obj = db.table(dbtable)
    if not table_obj.exists():
        raise click.ClickException(f"Table '{dbtable}' does not exist")
    if not table_obj.detect_fts():
        raise click.ClickException(
            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(
                    f"Table '{dbtable}' has no column '{c}"
                )
    sql = table_obj.search_sql(columns=column, order_by=order, limit=limit)
    if show_sql:
        click.echo(sql)
        return
    if quote:
        q = db.quote_fts(q)
    try:
        ctx.invoke(
            query,
            path=path,
            sql=sql,
            nl=nl,
            arrays=arrays,
            csv=csv,
            tsv=tsv,
            no_headers=no_headers,
            table=table,
            fmt=fmt,
            json_cols=json_cols,
            ascii_=ascii_,
            param=[("query", q)],
            load_extension=load_extension,
        )
    except click.ClickException as e:
        if "malformed MATCH expression" in str(e) or "unterminated string" in str(e):
            raise click.ClickException(
                f"{e!s}\n\nTry running this again with the --quote option"
            )
        else:
            raise


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("dbtable")
@click.option("-c", "--column", type=str, multiple=True, help="Columns to return")
@click.option("--where", help="Optional where clause")
@click.option("-o", "--order", type=str, help="Order by ('column' or 'column desc')")
@click.option(
    "-p",
    "--param",
    multiple=True,
    type=(str, str),
    help="Named :parameters for where clause",
)
@click.option(
    "--limit",
    type=int,
    help="Number of rows to return - defaults to everything",
)
@click.option(
    "--offset",
    type=int,
    help="SQL offset to use",
)
@output_options
@load_extension_option
@click.pass_context
def rows(
    ctx,
    path,
    dbtable,
    column,
    where,
    order,
    param,
    limit,
    offset,
    nl,
    arrays,
    csv,
    tsv,
    no_headers,
    table,
    fmt,
    json_cols,
    ascii_,
    load_extension,
):
    """Output all rows in the specified table

    Example:

    \b
        sqlite-utils rows trees.db Trees
    """
    columns = "*"
    if column:
        columns = ", ".join(quote_identifier(c) for c in column)
    sql = f"select {columns} from {quote_identifier(dbtable)}"
    if where:
        sql += " where " + where
    if order:
        sql += " order by " + order
    if limit:
        sql += f" limit {limit}"
    if offset:
        sql += f" offset {offset}"
    ctx.invoke(
        query,
        path=path,
        sql=sql,
        nl=nl,
        arrays=arrays,
        csv=csv,
        tsv=tsv,
        no_headers=no_headers,
        table=table,
        fmt=fmt,
        param=param,
        json_cols=json_cols,
        ascii_=ascii_,
        load_extension=load_extension,
    )


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("tables", nargs=-1)
@output_options
@load_extension_option
@click.pass_context
def triggers(
    ctx,
    path,
    tables,
    nl,
    arrays,
    csv,
    tsv,
    no_headers,
    table,
    fmt,
    json_cols,
    ascii_,
    load_extension,
):
    """Show triggers configured in this database

    Example:

    \b
        sqlite-utils triggers trees.db
    """
    sql = "select name, tbl_name as \"table\", sql from sqlite_master where type = 'trigger'"
    if tables:
        _quote_db = sqlite_utils.Database(memory=True)
        _register_db_for_cleanup(_quote_db)
        quote = _quote_db.quote
        sql += ' and "table" in ({})'.format(
            ", ".join(quote(table) for table in tables)
        )
    ctx.invoke(
        query,
        path=path,
        sql=sql,
        nl=nl,
        arrays=arrays,
        csv=csv,
        tsv=tsv,
        no_headers=no_headers,
        table=table,
        fmt=fmt,
        json_cols=json_cols,
        ascii_=ascii_,
        load_extension=load_extension,
    )


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("tables", nargs=-1)
@click.option("--aux", is_flag=True, help="Include auxiliary columns")
@output_options
@load_extension_option
@click.pass_context
def indexes(
    ctx,
    path,
    tables,
    aux,
    nl,
    arrays,
    csv,
    tsv,
    no_headers,
    table,
    fmt,
    json_cols,
    ascii_,
    load_extension,
):
    """Show indexes for the whole database or specific tables

    Example:

    \b
        sqlite-utils indexes trees.db Trees
    """
    sql = """
    select
      sqlite_master.name as "table",
      indexes.name as index_name,
      xinfo.*
    from sqlite_master
      join pragma_index_list(sqlite_master.name) indexes
      join pragma_index_xinfo(index_name) xinfo
    where
      sqlite_master.type = 'table'
    """
    if tables:
        _quote_db = sqlite_utils.Database(memory=True)
        _register_db_for_cleanup(_quote_db)
        quote = _quote_db.quote
        sql += " and sqlite_master.name in ({})".format(
            ", ".join(quote(table) for table in tables)
        )
    if not aux:
        sql += " and xinfo.key = 1"
    ctx.invoke(
        query,
        path=path,
        sql=sql,
        nl=nl,
        arrays=arrays,
        csv=csv,
        tsv=tsv,
        no_headers=no_headers,
        table=table,
        fmt=fmt,
        json_cols=json_cols,
        ascii_=ascii_,
        load_extension=load_extension,
    )


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("tables", nargs=-1, required=False)
@load_extension_option
def schema(
    path,
    tables,
    load_extension,
):
    """Show full schema for this database or for specified tables

    Example:

    \b
        sqlite-utils schema trees.db
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    if tables:
        for table in tables:
            click.echo(db[table].schema)
    else:
        click.echo(db.schema)


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.option(
    "--type",
    type=(
        str,
        click.Choice(
            ["INTEGER", "TEXT", "FLOAT", "REAL", "BLOB"], case_sensitive=False
        ),
    ),
    multiple=True,
    help="Change column type to INTEGER, TEXT, FLOAT, REAL or BLOB",
)
@click.option("--drop", type=str, multiple=True, help="Drop this column")
@click.option(
    "--rename", type=(str, str), multiple=True, help="Rename this column to X"
)
@click.option("-o", "--column-order", type=str, multiple=True, help="Reorder columns")
@click.option("--not-null", type=str, multiple=True, help="Set this column to NOT NULL")
@click.option(
    "--not-null-false", type=str, multiple=True, help="Remove NOT NULL from this column"
)
@click.option("--pk", type=str, multiple=True, help="Make this column the primary key")
@click.option(
    "--pk-none", is_flag=True, help="Remove primary key (convert to rowid table)"
)
@click.option(
    "--default",
    type=(str, str),
    multiple=True,
    help="Set default value for this column",
)
@click.option(
    "--default-none", type=str, multiple=True, help="Remove default from this column"
)
@click.option(
    "add_foreign_keys",
    "--add-foreign-key",
    type=(str, str, str),
    multiple=True,
    help="Add a foreign key constraint from a column to another table with another column",
)
@click.option(
    "drop_foreign_keys",
    "--drop-foreign-key",
    type=str,
    multiple=True,
    help="Drop foreign key constraint for this column",
)
@click.option(
    "--strict/--no-strict",
    default=None,
    help="Enable or disable STRICT mode (default: preserve current mode)",
)
@click.option("--sql", is_flag=True, help="Output SQL without executing it")
@load_extension_option
def transform(
    path,
    table,
    type,
    drop,
    rename,
    column_order,
    not_null,
    not_null_false,
    pk,
    pk_none,
    default,
    default_none,
    add_foreign_keys,
    drop_foreign_keys,
    strict,
    sql,
    load_extension,
):
    """Transform a table beyond the capabilities of ALTER TABLE

    Example:

    \b
        sqlite-utils transform mydb.db mytable \\
            --drop column1 \\
            --rename column2 column_renamed
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    types = {}
    for column, ctype in type:
        if ctype.upper() not in VALID_COLUMN_TYPES:
            raise click.ClickException(
                f"column types must be one of {VALID_COLUMN_TYPES}"
            )
        types[column] = ctype.upper()

    not_null_dict = {}
    for column in not_null:
        not_null_dict[column] = True
    for column in not_null_false:
        not_null_dict[column] = False

    default_dict = {}
    for column, value in default:
        default_dict[column] = value
    for column in default_none:
        default_dict[column] = None

    drop_set = set(drop)
    rename_dict = dict(rename)
    column_order_list = list(column_order) or None
    drop_foreign_keys_value = drop_foreign_keys or None
    add_foreign_keys_value = add_foreign_keys or None
    pk_value = DEFAULT
    if pk:
        if len(pk) == 1:
            pk_value = pk[0]
        else:
            pk_value = pk
    elif pk_none:
        pk_value = None

    table_obj = db.table(table)
    if sql:
        for line in table_obj.transform_sql(
            types=types,
            drop=drop_set,
            rename=rename_dict,
            column_order=column_order_list,
            not_null=not_null_dict,
            pk=pk_value,
            defaults=default_dict,
            drop_foreign_keys=drop_foreign_keys_value,
            add_foreign_keys=add_foreign_keys_value,
            strict=strict,
        ):
            click.echo(line)
    else:
        table_obj.transform(
            types=types,
            drop=drop_set,
            rename=rename_dict,
            column_order=column_order_list,
            not_null=not_null_dict,
            pk=pk_value,
            defaults=default_dict,
            drop_foreign_keys=drop_foreign_keys_value,
            add_foreign_keys=add_foreign_keys_value,
            strict=strict,
        )


@cli.command()
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument("columns", nargs=-1, required=True)
@click.option(
    "--table", "other_table", help="Name of the other table to extract columns to"
)
@click.option("--fk-column", help="Name of the foreign key column to add to the table")
@click.option(
    "--rename",
    type=(str, str),
    multiple=True,
    help="Rename this column in extracted table",
)
@load_extension_option
def extract(
    path,
    table,
    columns,
    other_table,
    fk_column,
    rename,
    load_extension,
):
    """Extract one or more columns into a separate table

    Example:

    \b
        sqlite-utils extract trees.db Street_Trees species
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    kwargs: dict[str, Any] = {
        "columns": columns,
        "table": other_table,
        "fk_column": fk_column,
        "rename": dict(rename),
    }
    try:
        db.table(table).extract(**kwargs)
    except (NoTable, InvalidColumns) as e:
        raise click.ClickException(str(e))


@cli.command(name="insert-files")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table")
@click.argument(
    "file_or_dir",
    nargs=-1,
    required=True,
    type=click.Path(file_okay=True, dir_okay=True, allow_dash=True),
)
@click.option(
    "-c",
    "--column",
    type=str,
    multiple=True,
    help="Column definitions for the table",
)
@click.option("pks", "--pk", help="Column to use as primary key", multiple=True)
@click.option("--alter", is_flag=True, help="Alter table to add missing columns")
@click.option("--replace", is_flag=True, help="Replace files with matching primary key")
@click.option("--upsert", is_flag=True, help="Upsert files with matching primary key")
@click.option("--name", type=str, help="File name to use")
@click.option("--text", is_flag=True, help="Store file content as TEXT, not BLOB")
@click.option(
    "--encoding",
    help="Character encoding for input, defaults to utf-8",
)
@click.option("-s", "--silent", is_flag=True, help="Don't show a progress bar")
@load_extension_option
def insert_files(
    path,
    table,
    file_or_dir,
    column,
    pks,
    alter,
    replace,
    upsert,
    name,
    text,
    encoding,
    silent,
    load_extension,
):
    """
    Insert one or more files using BLOB columns in the specified table

    Example:

    \b
        sqlite-utils insert-files pics.db images *.gif \\
            -c name:name \\
            -c content:content \\
            -c content_hash:sha256 \\
            -c created:ctime_iso \\
            -c modified:mtime_iso \\
            -c size:size \\
            --pk name
    """
    if not column:
        if text:
            column = ["path:path", "content_text:content_text", "size:size"]
        else:
            column = ["path:path", "content:content", "size:size"]
        if not pks:
            pks = ["path"]

    def yield_paths_and_relative_paths():
        for f_or_d in file_or_dir:
            path = pathlib.Path(f_or_d)
            if f_or_d == "-":
                yield "-", "-"
            elif path.is_dir():
                for subpath in path.rglob("*"):
                    if subpath.is_file():
                        yield subpath, subpath.relative_to(path)
            elif path.is_file():
                yield path, path

    # Load all paths so we can show a progress bar
    paths_and_relative_paths = list(yield_paths_and_relative_paths())

    with progressbar(paths_and_relative_paths, silent=silent) as bar:

        def to_insert():
            for path, relative_path in bar:
                row = {}
                # content_text is special case as it considers 'encoding'

                def _content_text(p):
                    resolved = p.resolve()
                    try:
                        return resolved.read_text(encoding=encoding)
                    except UnicodeDecodeError as e:
                        raise UnicodeDecodeErrorForPath(e, resolved)

                lookups = dict(FILE_COLUMNS, content_text=_content_text)
                if 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_text": lambda p: stdin_data.decode(
                            encoding or "utf-8"
                        ),
                        "sha256": lambda p: hashlib.sha256(stdin_data).hexdigest(),
                        "md5": lambda p: hashlib.md5(stdin_data).hexdigest(),
                        "size": lambda p: len(stdin_data),
                    }
                for coldef in column:
                    if ":" in coldef:
                        colname, coltype = coldef.rsplit(":", 1)
                    else:
                        colname, coltype = coldef, coldef
                    try:
                        value = lookups[coltype](path)
                        row[colname] = value
                    except KeyError:
                        raise click.ClickException(
                            "'{}' is not a valid column definition - options are {}".format(
                                coltype, ", ".join(lookups.keys())
                            )
                        )
                    # Special case for --name
                    if coltype == "name" and name:
                        row[colname] = name
                yield row

        db = sqlite_utils.Database(path)
        _register_db_for_cleanup(db)
        _load_extensions(db, load_extension)
        try:
            with db.conn:
                db.table(table).insert_all(
                    to_insert(),
                    pk=pks[0] if len(pks) == 1 else pks,
                    alter=alter,
                    replace=replace,
                    upsert=upsert,
                )
        except UnicodeDecodeErrorForPath as e:
            raise click.ClickException(
                UNICODE_ERROR.format(
                    f"Could not read file '{e.path}' as text\n\n{e.exception}"
                )
            )


@cli.command(name="analyze-tables")
@click.argument(
    "path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False, exists=True),
    required=True,
)
@click.argument("tables", nargs=-1)
@click.option(
    "-c",
    "--column",
    "columns",
    type=str,
    multiple=True,
    help="Specific columns to analyze",
)
@click.option("--save", is_flag=True, help="Save results to _analyze_tables table")
@click.option("--common-limit", type=int, default=10, help="How many common values")
@click.option("--no-most", is_flag=True, default=False, help="Skip most common values")
@click.option(
    "--no-least", is_flag=True, default=False, help="Skip least common values"
)
@load_extension_option
def analyze_tables(
    path,
    tables,
    columns,
    save,
    common_limit,
    no_most,
    no_least,
    load_extension,
):
    """Analyze the columns in one or more tables

    Example:

    \b
        sqlite-utils analyze-tables data.db trees
    """
    db = sqlite_utils.Database(path)
    _register_db_for_cleanup(db)
    _load_extensions(db, load_extension)
    _analyze(db, tables, columns, save, common_limit, no_most, no_least)


def _analyze(db, tables, columns, save, common_limit=10, no_most=False, no_least=False):
    if not tables:
        tables = db.table_names()
    todo = []
    table_counts = {}
    seen_columns = set()
    for table in tables:
        table_counts[table] = db[table].count
        for column in db[table].columns:
            if not columns or column.name in columns:
                todo.append((table, column.name))
                seen_columns.add(column.name)
    # Check the user didn't specify a column that doesn't exist
    if columns and (set(columns) - seen_columns):
        raise click.ClickException(
            "These columns were not found: {}".format(
                ", ".join(sorted(set(columns) - seen_columns))
            )
        )
    # Now we now how many we need to do
    for i, (table, column) in enumerate(todo):
        column_details = db[table].analyze_column(
            column,
            common_limit=common_limit,
            total_rows=table_counts[table],
            value_truncate=80,
            most_common=not no_most,
            least_common=not no_least,
        )
        if save:
            db["_analyze_tables_"].insert(
                column_details._asdict(), pk=("table", "column"), replace=True
            )
        most_common_rendered = ""
        if column_details.num_null != column_details.total_rows:
            most_common_rendered = _render_common(
                "\n\n  Most common:", column_details.most_common
            )
        least_common_rendered = _render_common(
            "\n\n  Least common:", column_details.least_common
        )
        details = (
            (
                textwrap.dedent("""
        {table}.{column}: ({i}/{total})

          Total rows: {total_rows}
          Null rows: {num_null}
          Blank rows: {num_blank}

          Distinct values: {num_distinct}{most_common_rendered}{least_common_rendered}
        """)
                .strip()
                .format(
                    i=i + 1,
                    total=len(todo),
                    most_common_rendered=most_common_rendered,
                    least_common_rendered=least_common_rendered,
                    **column_details._asdict(),
                )
            )
            + "\n"
        )
        click.echo(details)


@cli.command()
@click.argument("packages", nargs=-1, required=False)
@click.option(
    "-U", "--upgrade", is_flag=True, help="Upgrade packages to latest version"
)
@click.option(
    "-e",
    "--editable",
    help="Install a project in editable mode from this path",
)
def install(packages, upgrade, editable):
    """Install packages from PyPI into the same environment as sqlite-utils"""
    args = ["pip", "install"]
    if upgrade:
        args += ["--upgrade"]
    if editable:
        args += ["--editable", editable]
    args += list(packages)
    sys.argv = args
    run_module("pip", run_name="__main__")


@cli.command()
@click.argument("packages", nargs=-1, required=True)
@click.option("-y", "--yes", is_flag=True, help="Don't ask for confirmation")
def uninstall(packages, yes):
    """Uninstall Python packages from the sqlite-utils environment"""
    sys.argv = ["pip", "uninstall"] + list(packages) + (["-y"] if yes else [])
    run_module("pip", run_name="__main__")


def _generate_convert_help():
    help = textwrap.dedent("""
    Convert columns using Python code you supply. For example:

    \b
    sqlite-utils convert my.db mytable mycolumn \\
        '"\\n".join(textwrap.wrap(value, 10))' \\
        --import=textwrap

    "value" is a variable with the column value to be converted.

    CODE can also be a reference to a callable that takes the value, for example:

    \b
    sqlite-utils convert my.db mytable date r.parsedate
    sqlite-utils convert my.db mytable data json.loads --import json

    Use "-" for CODE to read Python code from standard input.

    The following common operations are available as recipe functions:
    """).strip()
    recipe_names = [
        n
        for n in dir(recipes)
        if not n.startswith("_")
        and n not in ("json", "parser", "Callable", "Optional")
        and callable(getattr(recipes, n))
    ]
    for name in recipe_names:
        fn = getattr(recipes, name)
        doc = textwrap.dedent(fn.__doc__.rstrip()).replace("\b\n", "")
        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:

    \b
    sqlite-utils convert my.db mytable mycolumn \\
        'r.jsonsplit(value, delimiter=":")'
    """).strip()
    return help


@cli.command(help=_generate_convert_help())
@click.argument(
    "db_path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table", type=str)
@click.argument("columns", type=str, nargs=-1, required=True)
@click.argument("code", type=str)
@click.option(
    "--import", "imports", type=str, multiple=True, help="Python modules to import"
)
@click.option(
    "--dry-run", is_flag=True, help="Show results of running this against first 10 rows"
)
@click.option(
    "--multi", is_flag=True, help="Populate columns for keys in returned dictionary"
)
@click.option("--where", help="Optional where clause")
@click.option(
    "-p",
    "--param",
    multiple=True,
    type=(str, str),
    help="Named :parameters for where clause",
)
@click.option("--output", help="Optional separate column to populate with the output")
@click.option(
    "--output-type",
    help="Column type to use for the output column",
    default="text",
    type=click.Choice(["integer", "float", "blob", "text"]),
)
@click.option("--drop", is_flag=True, help="Drop original column afterwards")
@click.option("-s", "--silent", is_flag=True, help="Don't show a progress bar")
@click.option("pdb_", "--pdb", is_flag=True, help="Open pdb debugger on first error")
def convert(
    db_path,
    table,
    columns,
    code,
    imports,
    dry_run,
    multi,
    where,
    param,
    output,
    output_type,
    drop,
    silent,
    pdb_,
):
    sqlite3.enable_callback_tracebacks(True)
    db = sqlite_utils.Database(db_path)
    _register_db_for_cleanup(db)
    if output is not None and len(columns) > 1:
        raise click.ClickException("Cannot use --output with more than one column")
    if multi and len(columns) > 1:
        raise click.ClickException("Cannot use --multi with more than one column")
    if drop and not (output or multi):
        raise click.ClickException("--drop can only be used with --output or --multi")
    if code == "-":
        # Read code from standard input
        code = sys.stdin.read()
    where_args = dict(param) if param else []
    # Compile the code into a function body called fn(value)
    try:
        fn = _compile_code(code, imports)
    except SyntaxError as e:
        raise click.ClickException(str(e))
    if dry_run:
        # Pull first 20 values for first column and preview them
        if multi:

            def preview(v):
                return json.dumps(fn(v), default=repr, ensure_ascii=False) if v else v

        else:

            def preview(v):
                return fn(v) if v else v

        db.conn.create_function("preview_transform", 1, preview)
        sql = """
            select
                [{column}] as value,
                preview_transform([{column}]) as preview
            from [{table}]{where} limit 10
        """.format(
            column=columns[0],
            table=table,
            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]))
            click.echo(" --- becomes:")
            click.echo(str(row[1]))
            click.echo()
        count = db[table].count_where(
            where=where,
            where_args=where_args,
        )
        click.echo("Would affect {} row{}".format(count, "" if count == 1 else "s"))
    else:
        # Wrap fn with a thing that will catch errors and optionally drop into pdb
        if pdb_:
            fn_ = fn

            def wrapped_fn(value):
                try:
                    return fn_(value)
                except Exception as ex:
                    print("\nException raised, dropping into pdb...:", ex)
                    pdb.post_mortem(ex.__traceback__)
                    sys.exit(1)

            fn = wrapped_fn
        try:
            db.table(table).convert(
                columns,
                fn,
                where=where,
                where_args=where_args,
                output=output,
                output_type=output_type,
                drop=drop,
                multi=multi,
                show_progress=not silent,
            )
        except BadMultiValues as e:
            raise click.ClickException(
                f"When using --multi code must return a Python dictionary - returned: {e.values!r}"
            )


@cli.command("add-geometry-column")
@click.argument(
    "db_path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table", type=str)
@click.argument("column_name", type=str)
@click.option(
    "-t",
    "--type",
    "geometry_type",
    type=click.Choice(
        [
            "point",
            "linestring",
            "polygon",
            "multipoint",
            "multilinestring",
            "multipolygon",
            "geometrycollection",
            "geometry",
        ],
        case_sensitive=False,
    ),
    default="GEOMETRY",
    help="Specify a geometry type for this column.",
    show_default=True,
)
@click.option(
    "--srid",
    type=int,
    default=4326,
    show_default=True,
    help="Spatial Reference ID. See https://spatialreference.org for details on specific projections.",
)
@click.option(
    "--dimensions",
    "coord_dimension",
    type=str,
    default="XY",
    help="Coordinate dimensions. Use XYZ for three-dimensional geometries.",
)
@click.option("--not-null", "not_null", is_flag=True, help="Add a NOT NULL constraint.")
@load_extension_option
def add_geometry_column(
    db_path,
    table,
    column_name,
    geometry_type,
    srid,
    coord_dimension,
    not_null,
    load_extension,
):
    """Add a SpatiaLite geometry column to an existing table. Requires SpatiaLite extension.
    \n\n
    By default, this command will try to load the SpatiaLite extension from usual paths.
    To load it from a specific path, use --load-extension."""
    db = sqlite_utils.Database(db_path)
    _register_db_for_cleanup(db)
    if not db[table].exists():
        raise click.ClickException(
            "You must create a table before adding a geometry column"
        )

    # load spatialite, one way or another
    if load_extension:
        _load_extensions(db, load_extension)
    db.init_spatialite()

    if db.table(table).add_geometry_column(
        column_name, geometry_type, srid, coord_dimension, not_null
    ):
        click.echo(f"Added {geometry_type} column {column_name} to {table}")


@cli.command("create-spatial-index")
@click.argument(
    "db_path",
    type=click.Path(file_okay=True, dir_okay=False, allow_dash=False),
    required=True,
)
@click.argument("table", type=str)
@click.argument("column_name", type=str)
@load_extension_option
def create_spatial_index(db_path, table, column_name, load_extension):
    """Create a spatial index on a SpatiaLite geometry column.
    The table and geometry column must already exist before trying to add a spatial index.
    \n\n
    By default, this command will try to load the SpatiaLite extension from usual paths.
    To load it from a specific path, use --load-extension."""
    db = sqlite_utils.Database(db_path)
    _register_db_for_cleanup(db)
    if not db[table].exists():
        raise click.ClickException(
            "You must create a table and add a geometry column before creating a spatial index"
        )

    # load spatialite
    if load_extension:
        _load_extensions(db, load_extension)
    db.init_spatialite()

    if column_name not in db[table].columns_dict:
        raise click.ClickException(
            "You must add a geometry column before creating a spatial index"
        )

    db.table(table).create_spatial_index(column_name)


def _find_migration_files(migrations):
    if not migrations:
        migrations = [pathlib.Path.cwd()]
    files = set()
    for path_str in migrations:
        path = pathlib.Path(path_str)
        if path.is_dir():
            files.update(path.rglob("migrations.py"))
        else:
            files.add(path)
    return sorted(files)


def _compatible_migration_set(obj):
    return isinstance(obj, sqlite_utils.Migrations) or all(
        hasattr(obj, attr) for attr in ("name", "applied", "pending", "apply")
    )


def _load_migration_sets(files):
    migration_sets = []
    for filepath in files:
        code = filepath.read_text()
        namespace = {
            "__file__": str(filepath),
            "__name__": "__sqlite_utils_migration__",
        }
        exec(code, namespace)
        migration_sets.extend(
            obj for obj in namespace.values() if _compatible_migration_set(obj)
        )
    return migration_sets


def _display_migration_list(db, migration_sets):
    for migration_set in migration_sets:
        click.echo(f"Migrations for: {migration_set.name}")
        click.echo()
        click.echo("  Applied:")
        for migration in migration_set.applied(db):
            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(f"    {migration.name}")
        if not output:
            click.echo("    (none)")
        click.echo()


def _stop_before_for_migration_set(stop_before, migration_set_name):
    matches = []
    for value in stop_before:
        set_name, separator, migration_name = value.partition(":")
        if separator:
            if set_name == migration_set_name:
                matches.append(migration_name)
        else:
            matches.append(value)
    return matches


@click.command()
@click.argument(
    "db_path", type=click.Path(dir_okay=False, readable=True, writable=True)
)
@click.argument("migrations", type=click.Path(dir_okay=True, exists=True), nargs=-1)
@click.option(
    "--stop-before",
    multiple=True,
    help="Stop before applying this migration. Use set:name to target a migration set.",
)
@click.option(
    "list_", "--list", is_flag=True, help="List migrations without running them"
)
@click.option("-v", "--verbose", is_flag=True, help="Show verbose output")
def migrate(db_path, migrations, stop_before, list_, verbose):
    """
    Apply pending database migrations.

    Usage:

        sqlite-utils migrate database.db

    This will find the migrations.py file in the current directory
    or subdirectories and apply any pending migrations.

    Or pass paths to one or more migrations.py files directly:

        sqlite-utils migrate database.db path/to/migrations.py

    Pass --list to see a list of applied and pending migrations
    without applying them.

    Use --stop-before migration_set:name to stop before a
    migration. This option can be used multiple times.
    """
    files = _find_migration_files(migrations)
    migration_sets = _load_migration_sets(files)
    if not migration_sets:
        raise click.ClickException("No migrations.py files found")

    if list_:
        if pathlib.Path(db_path).exists():
            db = sqlite_utils.Database(db_path)
        else:
            # Listing is read-only - don't create the database file
            db = sqlite_utils.Database(memory=True)
        _register_db_for_cleanup(db)
        # Legacy sqlite-migrate classes create the migrations table from
        # their pending()/applied() methods - run the listing inside a
        # transaction and roll it back so --list stays read-only
        db.begin()
        try:
            _display_migration_list(db, migration_sets)
        finally:
            db.rollback()
        return

    db = sqlite_utils.Database(db_path)
    _register_db_for_cleanup(db)

    prev_schema = db.schema
    if verbose:
        click.echo(f"Migrating {db_path}")
        click.echo("\nSchema before:\n")
        click.echo(textwrap.indent(prev_schema, "  ") or "  (empty)")
        click.echo()
    if stop_before:
        # Every --stop-before value must match at least one known migration
        known_names = set()
        for migration_set in migration_sets:
            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(
                f"{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(
                "--stop-before did not match any migrations: {}".format(
                    ", ".join(unknown)
                )
            )
    for migration_set in migration_sets:
        matches = _stop_before_for_migration_set(stop_before, migration_set.name)
        if isinstance(migration_set, sqlite_utils.Migrations):
            try:
                migration_set.apply(db, stop_before=matches)
            except ValueError as e:
                raise click.ClickException(str(e))
        else:
            # Legacy sqlite-migrate Migrations objects take a single string
            # for stop_before, not a list
            distinct = list(dict.fromkeys(matches))
            if len(distinct) > 1:
                raise click.ClickException(
                    "Migration set '{}' uses the older sqlite-migrate class, "
                    "which only supports a single --stop-before value - "
                    "got: {}".format(migration_set.name, ", ".join(distinct))
                )
            migration_set.apply(db, stop_before=distinct[0] if distinct else None)
    if verbose:
        click.echo("Schema after:\n")
        post_schema = db.schema
        if post_schema == prev_schema:
            click.echo("  (unchanged)")
        else:
            click.echo(textwrap.indent(post_schema, "  "))
            click.echo("\nSchema diff:\n")
            diff = list(
                difflib.unified_diff(prev_schema.splitlines(), post_schema.splitlines())
            )
            click.echo("\n".join(diff[3:]))


@cli.command(name="plugins")
def plugins_list():
    "List installed plugins"
    click.echo(json.dumps(get_plugins(), indent=2, ensure_ascii=False))


ensure_plugins_loaded()
pm.hook.register_commands(cli=cli)
cli.add_command(migrate)


def _render_common(title, values):
    if values is None:
        return ""
    lines = [title]
    for value, count in values:
        lines.append(f"    {count}: {value}")
    return "\n".join(lines)


class UnicodeDecodeErrorForPath(Exception):
    def __init__(self, exception, path):
        self.exception = exception
        self.path = path


FILE_COLUMNS = {
    "name": lambda p: p.name,
    "path": lambda p: str(p),
    "fullpath": lambda p: str(p.resolve()),
    "sha256": lambda p: hashlib.sha256(p.resolve().read_bytes()).hexdigest(),
    "md5": lambda p: hashlib.md5(p.resolve().read_bytes()).hexdigest(),
    "mode": lambda p: p.stat().st_mode,
    "content": lambda p: p.resolve().read_bytes(),
    "mtime": lambda p: p.stat().st_mtime,
    "ctime": lambda p: p.stat().st_ctime,
    "mtime_int": lambda p: int(p.stat().st_mtime),
    "ctime_int": lambda p: int(p.stat().st_ctime),
    "mtime_iso": lambda p: datetime.fromtimestamp(p.stat().st_mtime, timezone.utc)
    .replace(tzinfo=None)
    .isoformat(),
    "ctime_iso": lambda p: datetime.fromtimestamp(p.stat().st_ctime, timezone.utc)
    .replace(tzinfo=None)
    .isoformat(),
    "size": lambda p: p.stat().st_size,
    "stem": lambda p: p.stem,
    "suffix": lambda p: p.suffix,
}


def output_rows(iterator, headers, nl, arrays, json_cols, ascii_=False):
    # Duplicate column names would collide as dictionary keys, so rename
    # later occurrences id, id -> id, id_2 - CSV and table output keep
    # the original duplicate headers since they never build dictionaries
    headers = dedupe_keys(headers)
    # We have to iterate two-at-a-time so we can know if we
    # should output a trailing comma or if we have reached
    # the last row.
    current_iter, next_iter = itertools.tee(iterator, 2)
    next(next_iter, None)
    first = True
    for row, next_row in itertools.zip_longest(current_iter, next_iter):
        is_last = next_row is None
        data = row
        if json_cols:
            # Any value that is a valid JSON string should be treated as JSON
            data = [maybe_json(value) for value in data]
        if not arrays:
            data = dict(zip(headers, data))
        line = "{firstchar}{serialized}{maybecomma}{lastchar}".format(
            firstchar=("[" if first else " ") if not nl else "",
            serialized=json.dumps(data, default=json_binary, ensure_ascii=ascii_),
            maybecomma="," if (not nl and not is_last) else "",
            lastchar="]" if (is_last and not nl) else "",
        )
        yield line
        first = False
    if first:
        # We didn't output any rows, so yield the empty list
        yield "[]"


def maybe_json(value):
    if not isinstance(value, str):
        return value
    stripped = value.strip()
    if not (stripped.startswith(("{", "["))):
        return value
    try:
        return json.loads(stripped)
    except ValueError:
        return value


def json_binary(value):
    if isinstance(value, bytes):
        return {"$base64": True, "encoded": base64.b64encode(value).decode("latin-1")}
    else:
        raise TypeError


def verify_is_dict(doc):
    if not isinstance(doc, dict):
        raise click.ClickException(
            f"Rows must all be dictionaries, got: {repr(doc)[:1000]}"
        )
    return doc


def _load_extensions(db, load_extension):
    if load_extension:
        db.conn.enable_load_extension(True)
        for ext in load_extension:
            if ext == "spatialite" and not os.path.exists(ext):
                found = find_spatialite()
                if found is None:
                    raise click.ClickException("Could not find SpatiaLite extension")
                ext = found
            if ":" in ext:
                path, _, entrypoint = ext.partition(":")
                db.conn.execute("SELECT load_extension(?, ?)", [path, entrypoint])
            else:
                db.conn.load_extension(ext)


def _register_functions(db, functions):
    # Register any Python functions as SQL functions:
    # Check if this is a file path
    if "\n" not in functions and functions.endswith(".py"):
        try:
            functions = pathlib.Path(functions).read_text()
        except FileNotFoundError:
            raise click.ClickException(f"File not found: {functions}")

    sqlite3.enable_callback_tracebacks(True)
    globals = {}
    try:
        exec(functions, globals)
    except SyntaxError as 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("_"):
            db.register_function(value, name=name)


def _maybe_register_functions(db, functions_list):
    if not functions_list:
        return
    for functions in functions_list:
        if isinstance(functions, str) and functions.strip():
            _register_functions(db, functions)


def _rows_from_code(code):
    # code may be a path to a .py file
    if "\n" not in code and code.endswith(".py"):
        try:
            code = pathlib.Path(code).read_text()
        except FileNotFoundError:
            raise click.ClickException(f"File not found: {code}")
    namespace = {}
    try:
        exec(code, namespace)
    except SyntaxError as ex:
        raise click.ClickException(f"Error in --code: {ex}")
    rows = namespace.get("rows")
    if callable(rows):
        rows = rows()
    if isinstance(rows, dict):
        rows = [rows]
    error = click.ClickException(
        "--code must define a 'rows' function or iterable of rows to insert"
    )
    if rows is None or isinstance(rows, (str, bytes)):
        raise error
    try:
        iter(rows)
    except TypeError:
        raise error
    return rows