from .utils import (

chunks,

hash_record,

sqlite3,

OperationalError,

suggest_column_types,

types_for_column_types,

column_affinity,

progressbar,

find_spatialite,

)

import binascii

from collections import namedtuple

from collections.abc import Mapping

import contextlib

import datetime

import decimal

import inspect

import itertools

import json

import os

import pathlib

import re

import secrets

from sqlite_fts4 import rank_bm25 # type: ignore

import textwrap

from typing import (

cast,

Any,

Callable,

Dict,

Generator,

Iterable,

Sequence,

Set,

Type,

Union,

Optional,

List,

Tuple,

)

import uuid

from sqlite_utils.plugins import pm

try:

from sqlite_dump import iterdump # type: ignore[import-not-found]

except ImportError:

iterdump = None

SQLITE_MAX_VARS = 999

_quote_fts_re = re.compile(r'\s+|(".*?")')

_virtual_table_using_re = re.compile(

r"""

^ # Start of string

\s*CREATE\s+VIRTUAL\s+TABLE\s+ # CREATE VIRTUAL TABLE

(

'(?P<squoted_table>[^']*(?:''[^']*)*)' | # single quoted name

"(?P<dquoted_table>[^"]*(?:""[^"]*)*)" | # double quoted name

`(?P<backtick_table>[^`]+)` | # `backtick` quoted name

\[(?P<squarequoted_table>[^\]]+)\] | # [...] quoted name

(?P<identifier> # SQLite non-quoted identifier

[A-Za-z_\u0080-\uffff] # \u0080-\uffff = "any character larger than u007f"

[A-Za-z_\u0080-\uffff0-9\$]* # zero-or-more alphanemuric or $

)

)

\s+(IF\s+NOT\s+EXISTS\s+)? # IF NOT EXISTS (optional)

USING\s+(?P<using>\w+) # for example USING FTS5

""",

re.VERBOSE | re.IGNORECASE,

)

def quote_identifier(identifier: str) -> str:

"""

Quote an identifier (table name, column name, etc.) using double quotes.

Double quotes inside the identifier are escaped by doubling them.

"""

return '"{}"'.format(identifier.replace('"', '""'))

try:

import pandas as pd # type: ignore

except ImportError:

pd = None # type: ignore

try:

import numpy as np # type: ignore

except ImportError:

np = None # type: ignore

Column = namedtuple(

"Column", ("cid", "name", "type", "notnull", "default_value", "is_pk")

)

Column.__doc__ = """

Describes a SQLite column returned by the :attr:`.Table.columns` property.

``cid``

Column index

``name``

Column name

``type``

Column type

``notnull``

Does the column have a ``not null`` constraint

``default_value``

Default value for this column

``is_pk``

Is this column part of the primary key

"""

ColumnDetails = namedtuple(

"ColumnDetails",

(

"table",

"column",

"total_rows",

"num_null",

"num_blank",

"num_distinct",

"most_common",

"least_common",

),

)

ColumnDetails.__doc__ = """

Summary information about a column, see :ref:`python_api_analyze_column`.

``table``

The name of the table

``column``

The name of the column

``total_rows``

The total number of rows in the table

``num_null``

The number of rows for which this column is null

``num_blank``

The number of rows for which this column is blank (the empty string)

``num_distinct``

The number of distinct values in this column

``most_common``

The ``N`` most common values as a list of ``(value, count)`` tuples, or ``None`` if the table consists entirely of distinct values

``least_common``

The ``N`` least common values as a list of ``(value, count)`` tuples, or ``None`` if the table is entirely distinct

or if the number of distinct values is less than N (since they will already have been returned in ``most_common``)

"""

ForeignKey = namedtuple(

"ForeignKey", ("table", "column", "other_table", "other_column")

)

Index = namedtuple("Index", ("seq", "name", "unique", "origin", "partial", "columns"))

XIndex = namedtuple("XIndex", ("name", "columns"))

XIndexColumn = namedtuple(

"XIndexColumn", ("seqno", "cid", "name", "desc", "coll", "key")

)

Trigger = namedtuple("Trigger", ("name", "table", "sql"))

class TransformError(Exception):

pass

ForeignKeyIndicator = Union[

str,

ForeignKey,

Tuple[str, str],

Tuple[str, str, str],

Tuple[str, str, str, str],

]

ForeignKeysType = Union[Iterable[ForeignKeyIndicator], List[ForeignKeyIndicator]]

class Default:

pass

DEFAULT = Default()

COLUMN_TYPE_MAPPING = {

float: "REAL",

int: "INTEGER",

bool: "INTEGER",

str: "TEXT",

dict: "TEXT",

tuple: "TEXT",

list: "TEXT",

bytes.__class__: "BLOB",

bytes: "BLOB",

memoryview: "BLOB",

datetime.datetime: "TEXT",

datetime.date: "TEXT",

datetime.time: "TEXT",

datetime.timedelta: "TEXT",

decimal.Decimal: "REAL",

None.__class__: "TEXT",

uuid.UUID: "TEXT",

# SQLite explicit types

"TEXT": "TEXT",

"INTEGER": "INTEGER",

"FLOAT": "FLOAT",

"REAL": "REAL",

"BLOB": "BLOB",

"text": "TEXT",

"str": "TEXT",

"integer": "INTEGER",

"int": "INTEGER",

"float": "REAL",

"real": "REAL",

"blob": "BLOB",

"bytes": "BLOB",

}

# If numpy is available, add more types

if np:

try:

COLUMN_TYPE_MAPPING.update(

{

np.int8: "INTEGER",

np.int16: "INTEGER",

np.int32: "INTEGER",

np.int64: "INTEGER",

np.uint8: "INTEGER",

np.uint16: "INTEGER",

np.uint32: "INTEGER",

np.uint64: "INTEGER",

np.float16: "REAL",

np.float32: "REAL",

np.float64: "REAL",

}

)

except AttributeError:

# https://github.com/simonw/sqlite-utils/issues/632

pass

# If pandas is available, add more types

if pd:

COLUMN_TYPE_MAPPING.update({pd.Timestamp: "TEXT"}) # type: ignore

class AlterError(Exception):

"Error altering table"

class NoObviousTable(Exception):

"Could not tell which table this operation refers to"

class NoTable(Exception):

"Specified table does not exist"

class NoView(Exception):

"Specified view does not exist"

class BadPrimaryKey(Exception):

"Table does not have a single obvious primary key"

class NotFoundError(Exception):

"Record not found"

class PrimaryKeyRequired(Exception):

"Primary key needs to be specified"

class InvalidColumns(Exception):

"Specified columns do not exist"

class DescIndex(str):

pass

class BadMultiValues(Exception):

"With multi=True code must return a Python dictionary"

def __init__(self, values: object) -> None:

self.values = values

_COUNTS_TABLE_CREATE_SQL = """

CREATE TABLE IF NOT EXISTS "{}"(

"table" TEXT PRIMARY KEY,

count INTEGER DEFAULT 0

);

""".strip()

class Database:

"""

Wrapper for a SQLite database connection that adds a variety of useful utility methods.

To create an instance::

# create data.db file, or open existing:

db = Database("data.db")

# Create an in-memory database:

dB = Database(memory=True)

:param filename_or_conn: String path to a file, or a ``pathlib.Path`` object, or a

``sqlite3`` connection

:param memory: set to ``True`` to create an in-memory database

:param memory_name: creates a named in-memory database that can be shared across multiple connections

:param recreate: set to ``True`` to delete and recreate a file database (**dangerous**)

:param recursive_triggers: defaults to ``True``, which sets ``PRAGMA recursive_triggers=on;`` -

set to ``False`` to avoid setting this pragma

:param tracer: set a tracer function (``print`` works for this) which will be called with

``sql, parameters`` every time a SQL query is executed

:param use_counts_table: set to ``True`` to use a cached counts table, if available. See

:ref:`python_api_cached_table_counts`

:param use_old_upsert: set to ``True`` to force the older upsert implementation. See

:ref:`python_api_old_upsert`

:param strict: Apply STRICT mode to all created tables (unless overridden)

"""

_counts_table_name = "_counts"

use_counts_table = False

conn: sqlite3.Connection

def __init__(

self,

filename_or_conn: Optional[Union[str, pathlib.Path, sqlite3.Connection]] = None,

memory: bool = False,

memory_name: Optional[str] = None,

recreate: bool = False,

recursive_triggers: bool = True,

tracer: Optional[Callable] = None,

use_counts_table: bool = False,

execute_plugins: bool = True,

use_old_upsert: bool = False,

strict: bool = False,

):

self.memory_name = None

self.memory = False

self.use_old_upsert = use_old_upsert

assert (filename_or_conn is not None and (not memory and not memory_name)) or (

filename_or_conn is None and (memory or memory_name)

), "Either specify a filename_or_conn or pass memory=True"

if memory_name:

uri = "file:{}?mode=memory&cache=shared".format(memory_name)

self.conn = sqlite3.connect(

uri,

uri=True,

check_same_thread=False,

)

self.memory = True

self.memory_name = memory_name

elif memory or filename_or_conn == ":memory:":

self.conn = sqlite3.connect(":memory:")

self.memory = True

elif isinstance(filename_or_conn, (str, pathlib.Path)):

if recreate and os.path.exists(filename_or_conn):

try:

os.remove(filename_or_conn)

except OSError:

# Avoid mypy and __repr__ errors, see:

# https://github.com/simonw/sqlite-utils/issues/503

self.conn = sqlite3.connect(":memory:")

raise

self.conn = sqlite3.connect(str(filename_or_conn))

else:

assert not recreate, "recreate cannot be used with connections, only paths"

self.conn = filename_or_conn

self._tracer = tracer

if recursive_triggers:

self.execute("PRAGMA recursive_triggers=on;")

self._registered_functions: set = set()

self.use_counts_table = use_counts_table

if execute_plugins:

pm.hook.prepare_connection(conn=self.conn)

self.strict = strict

def __enter__(self) -> "Database":

return self

def __exit__(

self,

exc_type: Optional[Type[BaseException]],

exc_val: Optional[BaseException],

exc_tb: Optional[object],

) -> None:

self.close()

def close(self) -> None:

"Close the SQLite connection, and the underlying database file"

self.conn.close()

@contextlib.contextmanager

def ensure_autocommit_off(self) -> Generator[None, None, None]:

"""

Ensure autocommit is off for this database connection.

Example usage::

with db.ensure_autocommit_off():

# do stuff here

This will reset to the previous autocommit state at the end of the block.

"""

old_isolation_level = self.conn.isolation_level

try:

self.conn.isolation_level = None

yield

finally:

self.conn.isolation_level = old_isolation_level

@contextlib.contextmanager

def tracer(

self, tracer: Optional[Callable[[str, Optional[Sequence]], None]] = None

) -> Generator["Database", None, None]:

"""

Context manager to temporarily set a tracer function - all executed SQL queries will

be passed to this.

The tracer function should accept two arguments: ``sql`` and ``parameters``

Example usage::

with db.tracer(print):

db["creatures"].insert({"name": "Cleo"})

See :ref:`python_api_tracing`.

:param tracer: Callable accepting ``sql`` and ``parameters`` arguments

"""

prev_tracer = self._tracer

self._tracer = tracer or print

try:

yield self

finally:

self._tracer = prev_tracer

def __getitem__(self, table_name: str) -> Union["Table", "View"]:

"""

``db[table_name]`` returns a :class:`.Table` object for the table with the specified name.

If the table does not exist yet it will be created the first time data is inserted into it.

:param table_name: The name of the table

"""

if table_name in self.view_names():

return self.view(table_name)

return self.table(table_name)

def __repr__(self) -> str:

return "<Database {}>".format(self.conn)

def register_function(

self,

fn: Optional[Callable] = None,

deterministic: bool = False,

replace: bool = False,

name: Optional[str] = None,

) -> Optional[Callable[[Callable], Callable]]:

"""

``fn`` will be made available as a function within SQL, with the same name and number

of arguments. Can be used as a decorator::

@db.register_function

def upper(value):

return str(value).upper()

The decorator can take arguments::

@db.register_function(deterministic=True, replace=True)

def upper(value):

return str(value).upper()

See :ref:`python_api_register_function`.

:param fn: Function to register

:param deterministic: set ``True`` for functions that always returns the same output for a given input

:param replace: set ``True`` to replace an existing function with the same name - otherwise throw an error

:param name: name of the SQLite function - if not specified, the Python function name will be used

"""

def register(fn: Callable) -> Callable:

fn_name = name or fn.__name__ # type: ignore

arity = len(inspect.signature(fn).parameters)

if not replace and (fn_name, arity) in self._registered_functions:

return fn

kwargs: Dict[str, bool] = {}

registered = False

if deterministic:

# Try this, but fall back if sqlite3.NotSupportedError

try:

self.conn.create_function(

fn_name, arity, fn, **dict(kwargs, deterministic=True)

)

registered = True

except sqlite3.NotSupportedError:

pass

if not registered:

self.conn.create_function(fn_name, arity, fn, **kwargs)

self._registered_functions.add((fn_name, arity))

return fn

if fn is None:

return register

else:

register(fn)

return None

def register_fts4_bm25(self) -> None:

"Register the ``rank_bm25(match_info)`` function used for calculating relevance with SQLite FTS4."

self.register_function(rank_bm25, deterministic=True, replace=True)

def attach(self, alias: str, filepath: Union[str, pathlib.Path]) -> None:

"""

Attach another SQLite database file to this connection with the specified alias, equivalent to::

ATTACH DATABASE 'filepath.db' AS alias

:param alias: Alias name to use

:param filepath: Path to SQLite database file on disk

"""

attach_sql = """

ATTACH DATABASE '{}' AS {};

""".format(

str(pathlib.Path(filepath).resolve()), quote_identifier(alias)

).strip()

self.execute(attach_sql)

def query(

self, sql: str, params: Optional[Union[Sequence, Dict[str, Any]]] = None

) -> Generator[dict, None, None]:

"""

Execute ``sql`` and return an iterable of dictionaries representing each row.

:param sql: SQL query to execute

:param params: Parameters to use in that query - an iterable for ``where id = ?``

parameters, or a dictionary for ``where id = :id``

"""

cursor = self.execute(sql, params or tuple())

keys = [d[0] for d in cursor.description]

for row in cursor:

yield dict(zip(keys, row))

def execute(

self, sql: str, parameters: Optional[Union[Sequence, Dict[str, Any]]] = None

) -> sqlite3.Cursor:

"""

Execute SQL query and return a ``sqlite3.Cursor``.

:param sql: SQL query to execute

:param parameters: Parameters to use in that query - an iterable for ``where id = ?``

parameters, or a dictionary for ``where id = :id``

"""

if self._tracer:

self._tracer(sql, parameters)

if parameters is not None:

return self.conn.execute(sql, parameters)

else:

return self.conn.execute(sql)

def executescript(self, sql: str) -> sqlite3.Cursor:

"""

Execute multiple SQL statements separated by ; and return the ``sqlite3.Cursor``.

:param sql: SQL to execute

"""

if self._tracer:

self._tracer(sql, None)

return self.conn.executescript(sql)

def table(self, table_name: str, **kwargs: Any) -> "Table":

"""

Return a table object, optionally configured with default options.

See :ref:`reference_db_table` for option details.

:param table_name: Name of the table

"""

if table_name in self.view_names():

raise NoTable("Table {} is actually a view".format(table_name))

kwargs.setdefault("strict", self.strict)

return Table(self, table_name, **kwargs)

def view(self, view_name: str) -> "View":

"""

Return a view object.

:param view_name: Name of the view

"""

if view_name not in self.view_names():

raise NoView("View {} does not exist".format(view_name))

return View(self, view_name)

def quote(self, value: str) -> str:

"""

Apply SQLite string quoting to a value, including wrapping it in single quotes.

:param value: String to quote

"""

# Normally we would use .execute(sql, [params]) for escaping, but

# occasionally that isn't available - most notable when we need

# to include a "... DEFAULT 'value'" in a column definition.

return self.execute(

# Use SQLite itself to correctly escape this string:

"SELECT quote(:value)",

{"value": value},

).fetchone()[0]

def quote_fts(self, query: str) -> str:

"""

Escape special characters in a SQLite full-text search query.

This works by surrounding each token within the query with double

quotes, in order to avoid words like ``NOT`` and ``OR`` having

special meaning as defined by the FTS query syntax here:

https://www.sqlite.org/fts5.html#full_text_query_syntax

If the query has unbalanced ``"`` characters, adds one at end.

:param query: String to escape

"""

if query.count('"') % 2:

query += '"'

bits = _quote_fts_re.split(query)

bits = [b for b in bits if b and b != '""']

return " ".join(

'"{}"'.format(bit) if not bit.startswith('"') else bit for bit in bits

)

def quote_default_value(self, value: str) -> str:

if any(

[

str(value).startswith("'") and str(value).endswith("'"),

str(value).startswith('"') and str(value).endswith('"'),

]

):

return value

if str(value).upper() in ("CURRENT_TIME", "CURRENT_DATE", "CURRENT_TIMESTAMP"):

return value

if str(value).endswith(")"):

# Expr

return "({})".format(value)

return self.quote(value)

def table_names(self, fts4: bool = False, fts5: bool = False) -> List[str]:

"""

List of string table names in this database.

:param fts4: Only return tables that are part of FTS4 indexes

:param fts5: Only return tables that are part of FTS5 indexes

"""

where = ["type = 'table'"]

if fts4:

where.append("sql like '%USING FTS4%'")

if fts5:

where.append("sql like '%USING FTS5%'")

sql = "select name from sqlite_master where {}".format(" AND ".join(where))

return [r[0] for r in self.execute(sql).fetchall()]

def view_names(self) -> List[str]:

"List of string view names in this database."

return [

r[0]

for r in self.execute(

"select name from sqlite_master where type = 'view'"

).fetchall()

]

@property

def tables(self) -> List["Table"]:

"List of Table objects in this database."

return [self.table(name) for name in self.table_names()]

@property

def views(self) -> List["View"]:

"List of View objects in this database."

return [self.view(name) for name in self.view_names()]

@property

def triggers(self) -> List[Trigger]:

"List of ``(name, table_name, sql)`` tuples representing triggers in this database."

return [

Trigger(*r)

for r in self.execute(

"select name, tbl_name, sql from sqlite_master where type = 'trigger'"

).fetchall()

]

@property

def triggers_dict(self) -> Dict[str, str]:

"A ``{trigger_name: sql}`` dictionary of triggers in this database."

return {trigger.name: trigger.sql for trigger in self.triggers}

@property

def schema(self) -> str:

"SQL schema for this database."

sqls = []

for row in self.execute(

"select sql from sqlite_master where sql is not null"

).fetchall():

sql = row[0]

if not sql.strip().endswith(";"):

sql += ";"

sqls.append(sql)

return "\n".join(sqls)

@property

def supports_strict(self) -> bool:

"Does this database support STRICT mode?"

if not hasattr(self, "_supports_strict"):

try:

table_name = "t{}".format(secrets.token_hex(16))

with self.conn:

self.conn.execute(

"create table {} (name text) strict".format(table_name)

)

self.conn.execute("drop table {}".format(table_name))

self._supports_strict = True

except Exception:

self._supports_strict = False

return self._supports_strict

@property

def supports_on_conflict(self) -> bool:

# SQLite's upsert is implemented as INSERT INTO ... ON CONFLICT DO ...

if not hasattr(self, "_supports_on_conflict"):

table_name = "t{}".format(secrets.token_hex(16))

try:

with self.conn:

self.conn.execute(

"create table {} (id integer primary key, name text)".format(

table_name

)

)

self.conn.execute(

"insert into {} (id, name) values (1, 'one')".format(table_name)

)

self.conn.execute(

(

"insert into {} (id, name) values (1, 'two') "

"on conflict do update set name = 'two'"

).format(table_name)

)

self._supports_on_conflict = True

except Exception:

self._supports_on_conflict = False

finally:

self.conn.execute("drop table if exists {}".format(table_name))

return self._supports_on_conflict

@property

def sqlite_version(self) -> Tuple[int, ...]:

"Version of SQLite, as a tuple of integers for example ``(3, 36, 0)``."

row = self.execute("select sqlite_version()").fetchall()[0]

return tuple(map(int, row[0].split(".")))

@property

def journal_mode(self) -> str:

"""

Current ``journal_mode`` of this database.

https://www.sqlite.org/pragma.html#pragma_journal_mode

"""

return self.execute("PRAGMA journal_mode;").fetchone()[0]

def enable_wal(self) -> None:

"""

Sets ``journal_mode`` to ``'wal'`` to enable Write-Ahead Log mode.

"""

if self.journal_mode != "wal":

with self.ensure_autocommit_off():

self.execute("PRAGMA journal_mode=wal;")

def disable_wal(self) -> None:

"Sets ``journal_mode`` back to ``'delete'`` to disable Write-Ahead Log mode."

if self.journal_mode != "delete":

with self.ensure_autocommit_off():

self.execute("PRAGMA journal_mode=delete;")

def _ensure_counts_table(self) -> None:

with self.conn:

self.execute(_COUNTS_TABLE_CREATE_SQL.format(self._counts_table_name))

def enable_counts(self) -> None:

"""

Enable trigger-based count caching for every table in the database, see

:ref:`python_api_cached_table_counts`.

"""

self._ensure_counts_table()

for table in self.tables:

if (

table.virtual_table_using is None

and table.name != self._counts_table_name

):

table.enable_counts()

self.use_counts_table = True

def cached_counts(self, tables: Optional[Iterable[str]] = None) -> Dict[str, int]:

"""

Return ``{table_name: count}`` dictionary of cached counts for specified tables, or

all tables if ``tables`` not provided.

:param tables: Subset list of tables to return counts for.

"""

sql = 'select "table", count from {}'.format(self._counts_table_name)

tables_list = list(tables) if tables else None

if tables_list:

sql += ' where "table" in ({})'.format(", ".join("?" for _ in tables_list))

try:

return {r[0]: r[1] for r in self.execute(sql, tables_list).fetchall()}

except OperationalError:

return {}

def reset_counts(self) -> None:

"Re-calculate cached counts for tables."

tables = [table for table in self.tables if table.has_counts_triggers]

with self.conn:

self._ensure_counts_table()

counts_table = self.table(self._counts_table_name)

counts_table.delete_where()

counts_table.insert_all(

{"table": table.name, "count": table.execute_count()}

for table in tables

)

def execute_returning_dicts(

self, sql: str, params: Optional[Union[Sequence, Dict[str, Any]]] = None

) -> List[dict]:

return list(self.query(sql, params))

def resolve_foreign_keys(

self, name: str, foreign_keys: ForeignKeysType

) -> List[ForeignKey]:

"""

Given a list of differing foreign_keys definitions, return a list of

fully resolved ForeignKey() named tuples.

:param name: Name of table that foreign keys are being defined for

:param foreign_keys: List of foreign keys, each of which can be a

string, a ForeignKey() named tuple, a tuple of (column, other_table),

or a tuple of (column, other_table, other_column), or a tuple of

(table, column, other_table, other_column)

"""

table = self.table(name)

if all(isinstance(fk, ForeignKey) for fk in foreign_keys):

return cast(List[ForeignKey], foreign_keys)

if all(isinstance(fk, str) for fk in foreign_keys):

# It's a list of columns

fks = []

for column in foreign_keys:

column = cast(str, column)

other_table = table.guess_foreign_table(column)

other_column = table.guess_foreign_column(other_table)

fks.append(ForeignKey(name, column, other_table, other_column))

return fks

assert all(

isinstance(fk, (tuple, list)) for fk in foreign_keys

), "foreign_keys= should be a list of tuples"

fks = []

for tuple_or_list in foreign_keys:

if len(tuple_or_list) == 4:

assert (

tuple_or_list[0] == name

), "First item in {} should have been {}".format(tuple_or_list, name)

assert len(tuple_or_list) in (

2,

3,

4,

), "foreign_keys= should be a list of tuple pairs or triples"

if len(tuple_or_list) in (3, 4):

if len(tuple_or_list) == 4:

tuple_or_list = cast(Tuple[str, str, str], tuple_or_list[1:])

else:

tuple_or_list = cast(Tuple[str, str, str], tuple_or_list)

fks.append(

ForeignKey(

name, tuple_or_list[0], tuple_or_list[1], tuple_or_list[2]

)

)

else:

# Guess the primary key

fks.append(

ForeignKey(

name,

tuple_or_list[0],

tuple_or_list[1],

table.guess_foreign_column(tuple_or_list[1]),

)

)

return fks

def create_table_sql(

self,

name: str,

columns: Dict[str, Any],

pk: Optional[Any] = None,

foreign_keys: Optional[ForeignKeysType] = None,

column_order: Optional[List[str]] = None,

not_null: Optional[Iterable[str]] = None,

defaults: Optional[Dict[str, Any]] = None,

hash_id: Optional[str] = None,

hash_id_columns: Optional[Iterable[str]] = None,

extracts: Optional[Union[Dict[str, str], List[str]]] = None,

if_not_exists: bool = False,

strict: bool = False,

) -> str:

"""

Returns the SQL ``CREATE TABLE`` statement for creating the specified table.

:param name: Name of table

:param columns: Dictionary mapping column names to their types, for example ``{"name": str, "age": int}``

:param pk: String name of column to use as a primary key, or a tuple of strings for a compound primary key covering multiple columns

:param foreign_keys: List of foreign key definitions for this table

:param column_order: List specifying which columns should come first

:param not_null: List of columns that should be created as ``NOT NULL``

:param defaults: Dictionary specifying default values for columns

:param hash_id: Name of column to be used as a primary key containing a hash of the other columns

:param hash_id_columns: List of columns to be used when calculating the hash ID for a row

:param extracts: List or dictionary of columns to be extracted during inserts, see :ref:`python_api_extracts`

:param if_not_exists: Use ``CREATE TABLE IF NOT EXISTS``

:param strict: Apply STRICT mode to table

"""

if hash_id_columns and (hash_id is None):

hash_id = "id"

foreign_keys = self.resolve_foreign_keys(name, foreign_keys or [])

foreign_keys_by_column = {fk.column: fk for fk in foreign_keys}

# any extracts will be treated as integer columns with a foreign key

extracts = resolve_extracts(extracts)

for extract_column, extract_table in extracts.items():

if isinstance(extract_column, tuple):

assert False

# Ensure other table exists

if not self[extract_table].exists():

self.create_table(extract_table, {"id": int, "value": str}, pk="id")

columns[extract_column] = int

foreign_keys_by_column[extract_column] = ForeignKey(

name, extract_column, extract_table, "id"

)

# Soundness check not_null, and defaults if provided

not_null = not_null or set()

defaults = defaults or {}

assert columns, "Tables must have at least one column"

assert all(

n in columns for n in not_null

), "not_null set {} includes items not in columns {}".format(

repr(not_null), repr(set(columns.keys()))

)

assert all(

n in columns for n in defaults

), "defaults set {} includes items not in columns {}".format(

repr(set(defaults)), repr(set(columns.keys()))

)

column_items = list(columns.items())

if column_order is not None:

def sort_key(p):

return column_order.index(p[0]) if p[0] in column_order else 999

column_items.sort(key=sort_key)

if hash_id:

column_items.insert(0, (hash_id, str))

pk = hash_id

# Soundness check foreign_keys point to existing tables

for fk in foreign_keys:

if fk.other_table == name and columns.get(fk.other_column):

continue

if fk.other_column != "rowid" and not any(

c for c in self[fk.other_table].columns if c.name == fk.other_column

):

raise AlterError(

"No such column: {}.{}".format(fk.other_table, fk.other_column)

)

column_defs = []

# ensure pk is a tuple

single_pk = None

if isinstance(pk, list) and len(pk) == 1 and isinstance(pk[0], str):

pk = pk[0]

if isinstance(pk, str):

single_pk = pk

if pk not in [c[0] for c in column_items]:

column_items.insert(0, (pk, int))

for column_name, column_type in column_items:

column_extras = []

if column_name == single_pk:

column_extras.append("PRIMARY KEY")

if column_name in not_null: