remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-09-27 20:34:12 +02:00
Code Issues Projects Releases Packages Wiki Activity

Use double quotes not braces for tables and columns

Browse source 
Refs #677
This commit is contained in:
Simon Willison 2025-11-23 17:11:50 -08:00
commit b2078ce6a1
17 changed files with 682 additions and 609 deletions
Download patch file Download diff file Expand all files Collapse all files

20
sqlite_utils/cli.py
View file

@ -6,7 +6,13 @@ import hashlib
import pathlib import pathlib
from runpy import run_module from runpy import run_module
import sqlite_utils import sqlite_utils
from sqlite_utils.db import AlterError, BadMultiValues, DescIndex, NoTable from sqlite_utils.db import (
AlterError,
BadMultiValues,
DescIndex,
NoTable,
quote_identifier,
)
from sqlite_utils.plugins import pm, get_plugins from sqlite_utils.plugins import pm, get_plugins
from sqlite_utils.utils import maximize_csv_field_size_limit from sqlite_utils.utils import maximize_csv_field_size_limit
from sqlite_utils import recipes from sqlite_utils import recipes
@ -1967,7 +1973,9 @@ def memory(
view_names.append("t") view_names.append("t")
for view_name in view_names: for view_name in view_names:
if not db[view_name].exists(): if not db[view_name].exists():
db.create_view(view_name, "select * from [{}]".format(file_table)) db.create_view(
view_name, "select * from {}".format(quote_identifier(file_table))
)
if fp: if fp:
fp.close() fp.close()
@ -2230,8 +2238,8 @@ def rows(
""" """
columns = "*" columns = "*"
if column: if column:
columns = ", ".join("[{}]".format(c) for c in column) columns = ", ".join(quote_identifier(c) for c in column)
sql = "select {} from [{}]".format(columns, dbtable) sql = "select {} from {}".format(columns, quote_identifier(dbtable))
if where: if where:
sql += " where " + where sql += " where " + where
if order: if order:
@ -2288,10 +2296,10 @@ def triggers(
\b \b
sqlite-utils triggers trees.db sqlite-utils triggers trees.db
""" """
sql = "select name, tbl_name as [table], sql from sqlite_master where type = 'trigger'" sql = "select name, tbl_name as \"table\", sql from sqlite_master where type = 'trigger'"
if tables: if tables:
quote = sqlite_utils.Database(memory=True).quote quote = sqlite_utils.Database(memory=True).quote
sql += " and [table] in ({})".format( sql += ' and "table" in ({})'.format(
", ".join(quote(table) for table in tables) ", ".join(quote(table) for table in tables)
) )
ctx.invoke( ctx.invoke(

386
sqlite_utils/db.py
View file

@ -70,6 +70,16 @@ USING\s+(?P<using>\w+) # for example USING FTS5
re.VERBOSE | re.IGNORECASE, 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: try:
import pandas as pd # type: ignore import pandas as pd # type: ignore
except ImportError: except ImportError:
@ -280,8 +290,8 @@ class BadMultiValues(Exception):
_COUNTS_TABLE_CREATE_SQL = """ _COUNTS_TABLE_CREATE_SQL = """
CREATE TABLE IF NOT EXISTS [{}]( CREATE TABLE IF NOT EXISTS "{}"(
[table] TEXT PRIMARY KEY, "table" TEXT PRIMARY KEY,
count INTEGER DEFAULT 0 count INTEGER DEFAULT 0
); );
""".strip() """.strip()
@ -500,9 +510,9 @@ class Database:
:param filepath: Path to SQLite database file on disk :param filepath: Path to SQLite database file on disk
""" """
attach_sql = """ attach_sql = """
ATTACH DATABASE '{}' AS [{}]; ATTACH DATABASE '{}' AS {};
""".format( """.format(
str(pathlib.Path(filepath).resolve()), alias str(pathlib.Path(filepath).resolve()), quote_identifier(alias)
).strip() ).strip()
self.execute(attach_sql) self.execute(attach_sql)
@ -786,9 +796,9 @@ class Database:
:param tables: Subset list of tables to return counts for. :param tables: Subset list of tables to return counts for.
""" """
sql = "select [table], count from {}".format(self._counts_table_name) sql = 'select "table", count from {}'.format(self._counts_table_name)
if tables: if tables:
sql += " where [table] in ({})".format(", ".join("?" for table in tables)) sql += ' where "table" in ({})'.format(", ".join("?" for table in tables))
try: try:
return {r[0]: r[1] for r in self.execute(sql, tables).fetchall()} return {r[0]: r[1] for r in self.execute(sql, tables).fetchall()}
except OperationalError: except OperationalError:
@ -976,9 +986,13 @@ class Database:
) )
if column_name in foreign_keys_by_column: if column_name in foreign_keys_by_column:
column_extras.append( column_extras.append(
"REFERENCES [{other_table}]([{other_column}])".format( "REFERENCES {}({})".format(
other_table=foreign_keys_by_column[column_name].other_table, quote_identifier(
other_column=foreign_keys_by_column[column_name].other_column, foreign_keys_by_column[column_name].other_table
),
quote_identifier(
foreign_keys_by_column[column_name].other_column
),
) )
) )
column_type_str = COLUMN_TYPE_MAPPING[column_type] column_type_str = COLUMN_TYPE_MAPPING[column_type]
@ -987,8 +1001,8 @@ class Database:
if strict and column_type_str == "FLOAT": if strict and column_type_str == "FLOAT":
column_type_str = "REAL" column_type_str = "REAL"
column_defs.append( column_defs.append(
" [{column_name}] {column_type}{column_extras}".format( " {} {column_type}{column_extras}".format(
column_name=column_name, quote_identifier(column_name),
column_type=column_type_str, column_type=column_type_str,
column_extras=( column_extras=(
(" " + " ".join(column_extras)) if column_extras else "" (" " + " ".join(column_extras)) if column_extras else ""
@ -998,15 +1012,15 @@ class Database:
extra_pk = "" extra_pk = ""
if single_pk is None and pk and len(pk) > 1: if single_pk is None and pk and len(pk) > 1:
extra_pk = ",\n PRIMARY KEY ({pks})".format( extra_pk = ",\n PRIMARY KEY ({pks})".format(
pks=", ".join(["[{}]".format(p) for p in pk]) pks=", ".join([quote_identifier(p) for p in pk])
) )
columns_sql = ",\n".join(column_defs) columns_sql = ",\n".join(column_defs)
sql = """CREATE TABLE {if_not_exists}[{table}] ( sql = """CREATE TABLE {if_not_exists}{table} (
{columns_sql}{extra_pk} {columns_sql}{extra_pk}
){strict}; ){strict};
""".format( """.format(
if_not_exists="IF NOT EXISTS " if if_not_exists else "", if_not_exists="IF NOT EXISTS " if if_not_exists else "",
table=name, table=quote_identifier(name),
columns_sql=columns_sql, columns_sql=columns_sql,
extra_pk=extra_pk, extra_pk=extra_pk,
strict=" STRICT" if strict and self.supports_strict else "", strict=" STRICT" if strict and self.supports_strict else "",
@ -1144,8 +1158,8 @@ class Database:
:param new_name: Name to rename it to :param new_name: Name to rename it to
""" """
self.execute( self.execute(
"ALTER TABLE [{name}] RENAME TO [{new_name}]".format( "ALTER TABLE {} RENAME TO {}".format(
name=name, new_name=new_name quote_identifier(name), quote_identifier(new_name)
) )
) )
@ -1270,7 +1284,7 @@ class Database:
""" """
sql = "ANALYZE" sql = "ANALYZE"
if name is not None: if name is not None:
sql += " [{}]".format(name) sql += " {}".format(quote_identifier(name))
self.execute(sql) self.execute(sql)
def iterdump(self) -> Generator[str, None, None]: def iterdump(self) -> Generator[str, None, None]:
@ -1348,7 +1362,7 @@ class Queryable:
:param where_args: Parameters to use with that fragment - an iterable for ``id > ?`` :param where_args: Parameters to use with that fragment - an iterable for ``id > ?``
parameters, or a dictionary for ``id > :id`` parameters, or a dictionary for ``id > :id``
""" """
sql = "select count(*) from [{}]".format(self.name) sql = "select count(*) from {}".format(quote_identifier(self.name))
if where is not None: if where is not None:
sql += " where " + where sql += " where " + where
return self.db.execute(sql, where_args or []).fetchone()[0] return self.db.execute(sql, where_args or []).fetchone()[0]
@ -1391,7 +1405,7 @@ class Queryable:
""" """
if not self.exists(): if not self.exists():
return return
sql = "select {} from [{}]".format(select, self.name) sql = "select {} from {}".format(select, quote_identifier(self.name))
if where is not None: if where is not None:
sql += " where " + where sql += " where " + where
if order_by is not None: if order_by is not None:
@ -1429,7 +1443,7 @@ class Queryable:
if not pks: if not pks:
column_names.insert(0, "rowid") column_names.insert(0, "rowid")
pks = ["rowid"] pks = ["rowid"]
select = ",".join("[{}]".format(column_name) for column_name in column_names) select = ",".join(quote_identifier(column_name) for column_name in column_names)
for row in self.rows_where( for row in self.rows_where(
select=select, select=select,
where=where, where=where,
@ -1448,7 +1462,9 @@ class Queryable:
"List of :ref:`Columns <reference_db_other_column>` representing the columns in this table or view." "List of :ref:`Columns <reference_db_other_column>` representing the columns in this table or view."
if not self.exists(): if not self.exists():
return [] return []
rows = self.db.execute("PRAGMA table_info([{}])".format(self.name)).fetchall() rows = self.db.execute(
"PRAGMA table_info({})".format(quote_identifier(self.name))
).fetchall()
return [Column(*row) for row in rows] return [Column(*row) for row in rows]
@property @property
@ -1588,7 +1604,7 @@ class Table(Queryable):
) )
) )
wheres = ["[{}] = ?".format(pk_name) for pk_name in pks] wheres = ["{} = ?".format(quote_identifier(pk_name)) for pk_name in pks]
rows = self.rows_where(" and ".join(wheres), pk_values) rows = self.rows_where(" and ".join(wheres), pk_values)
try: try:
row = list(rows)[0] row = list(rows)[0]
@ -1602,7 +1618,7 @@ class Table(Queryable):
"List of foreign keys defined on this table." "List of foreign keys defined on this table."
fks = [] fks = []
for row in self.db.execute( for row in self.db.execute(
"PRAGMA foreign_key_list([{}])".format(self.name) "PRAGMA foreign_key_list({})".format(quote_identifier(self.name))
).fetchall(): ).fetchall():
if row is not None: if row is not None:
id, seq, table_name, from_, to_, on_update, on_delete, match = row id, seq, table_name, from_, to_, on_update, on_delete, match = row
@ -1799,9 +1815,9 @@ class Table(Queryable):
if not self.exists(): if not self.exists():
raise NoTable(f"Table {self.name} does not exist") raise NoTable(f"Table {self.name} does not exist")
with self.db.conn: with self.db.conn:
sql = "CREATE TABLE [{new_table}] AS SELECT * FROM [{table}];".format( sql = "CREATE TABLE {} AS SELECT * FROM {};".format(
new_table=new_name, quote_identifier(new_name),
table=self.name, quote_identifier(self.name),
) )
self.db.execute(sql) self.db.execute(sql)
return self.db[new_name] return self.db[new_name]
@ -2034,23 +2050,27 @@ class Table(Queryable):
if "rowid" not in new_cols: if "rowid" not in new_cols:
new_cols.insert(0, "rowid") new_cols.insert(0, "rowid")
old_cols.insert(0, "rowid") old_cols.insert(0, "rowid")
copy_sql = "INSERT INTO [{new_table}] ({new_cols})\n SELECT {old_cols} FROM [{old_table}];".format( copy_sql = "INSERT INTO {} ({new_cols})\n SELECT {old_cols} FROM {};".format(
new_table=new_table_name, quote_identifier(new_table_name),
old_table=self.name, quote_identifier(self.name),
old_cols=", ".join("[{}]".format(col) for col in old_cols), old_cols=", ".join(quote_identifier(col) for col in old_cols),
new_cols=", ".join("[{}]".format(col) for col in new_cols), new_cols=", ".join(quote_identifier(col) for col in new_cols),
) )
sqls.append(copy_sql) sqls.append(copy_sql)
# Drop (or keep) the old table # Drop (or keep) the old table
if keep_table: if keep_table:
sqls.append( sqls.append(
"ALTER TABLE [{}] RENAME TO [{}];".format(self.name, keep_table) "ALTER TABLE {} RENAME TO {};".format(
quote_identifier(self.name), quote_identifier(keep_table)
)
) )
else: else:
sqls.append("DROP TABLE [{}];".format(self.name)) sqls.append("DROP TABLE {};".format(quote_identifier(self.name)))
# Rename the new one # Rename the new one
sqls.append( sqls.append(
"ALTER TABLE [{}] RENAME TO [{}];".format(new_table_name, self.name) "ALTER TABLE {} RENAME TO {};".format(
quote_identifier(new_table_name), quote_identifier(self.name)
)
) )
# Re-add existing indexes # Re-add existing indexes
for index in self.indexes: for index in self.indexes:
@ -2066,7 +2086,7 @@ class Table(Queryable):
"transformation and manually recreate the new index after running this transformation." "transformation and manually recreate the new index after running this transformation."
) )
if keep_table: if keep_table:
sqls.append(f"DROP INDEX IF EXISTS [{index.name}];") sqls.append(f"DROP INDEX IF EXISTS {quote_identifier(index.name)};")
for col in index.columns: for col in index.columns:
if col in rename.keys() or col in drop: if col in rename.keys() or col in drop:
raise TransformError( raise TransformError(
@ -2137,11 +2157,11 @@ class Table(Queryable):
lookup_columns = [(rename.get(col) or col) for col in columns] lookup_columns = [(rename.get(col) or col) for col in columns]
lookup_table.create_index(lookup_columns, unique=True, if_not_exists=True) lookup_table.create_index(lookup_columns, unique=True, if_not_exists=True)
self.db.execute( self.db.execute(
"INSERT OR IGNORE INTO [{lookup_table}] ({lookup_columns}) SELECT DISTINCT {table_cols} FROM [{table}]".format( "INSERT OR IGNORE INTO {} ({lookup_columns}) SELECT DISTINCT {table_cols} FROM {}".format(
lookup_table=table, quote_identifier(table),
lookup_columns=", ".join("[{}]".format(c) for c in lookup_columns), quote_identifier(self.name),
table_cols=", ".join("[{}]".format(c) for c in columns), lookup_columns=", ".join(quote_identifier(c) for c in lookup_columns),
table=self.name, table_cols=", ".join(quote_identifier(c) for c in columns),
) )
) )
@ -2150,16 +2170,16 @@ class Table(Queryable):
# And populate it # And populate it
self.db.execute( self.db.execute(
"UPDATE [{table}] SET [{magic_lookup_column}] = (SELECT id FROM [{lookup_table}] WHERE {where})".format( "UPDATE {} SET {} = (SELECT id FROM {} WHERE {where})".format(
table=self.name, quote_identifier(self.name),
magic_lookup_column=magic_lookup_column, quote_identifier(magic_lookup_column),
lookup_table=table, quote_identifier(table),
where=" AND ".join( where=" AND ".join(
"[{table}].[{column}] IS [{lookup_table}].[{lookup_column}]".format( "{}.{} IS {}.{}".format(
table=self.name, quote_identifier(self.name),
lookup_table=table, quote_identifier(column),
column=column, quote_identifier(table),
lookup_column=rename.get(column) or column, quote_identifier(rename.get(column) or column),
) )
for column in columns for column in columns
), ),
@ -2216,10 +2236,9 @@ class Table(Queryable):
columns_sql = [] columns_sql = []
for column in columns: for column in columns:
if isinstance(column, DescIndex): if isinstance(column, DescIndex):
fmt = "[{}] desc" columns_sql.append("{} desc".format(quote_identifier(column)))
else: else:
fmt = "[{}]" columns_sql.append(quote_identifier(column))
columns_sql.append(fmt.format(column))
suffix = None suffix = None
created_index_name = None created_index_name = None
@ -2230,14 +2249,14 @@ class Table(Queryable):
sql = ( sql = (
textwrap.dedent( textwrap.dedent(
""" """
CREATE {unique}INDEX {if_not_exists}[{index_name}] CREATE {unique}INDEX {if_not_exists}{index_name}
ON [{table_name}] ({columns}); ON {table_name} ({columns});
""" """
) )
.strip() .strip()
.format( .format(
index_name=created_index_name, index_name=quote_identifier(created_index_name),
table_name=self.name, table_name=quote_identifier(self.name),
columns=", ".join(columns_sql), columns=", ".join(columns_sql),
unique="UNIQUE " if unique else "", unique="UNIQUE " if unique else "",
if_not_exists="IF NOT EXISTS " if if_not_exists else "", if_not_exists="IF NOT EXISTS " if if_not_exists else "",
@ -2307,9 +2326,9 @@ class Table(Queryable):
not_null_sql = "NOT NULL DEFAULT {}".format( not_null_sql = "NOT NULL DEFAULT {}".format(
self.db.quote_default_value(not_null_default) self.db.quote_default_value(not_null_default)
) )
sql = "ALTER TABLE [{table}] ADD COLUMN [{col_name}] {col_type}{not_null_default};".format( sql = "ALTER TABLE {} ADD COLUMN {} {col_type}{not_null_default};".format(
table=self.name, quote_identifier(self.name),
col_name=col_name, quote_identifier(col_name),
col_type=fk_col_type or COLUMN_TYPE_MAPPING[col_type], col_type=fk_col_type or COLUMN_TYPE_MAPPING[col_type],
not_null_default=(" " + not_null_sql) if not_null_sql else "", not_null_default=(" " + not_null_sql) if not_null_sql else "",
) )
@ -2325,7 +2344,7 @@ class Table(Queryable):
:param ignore: Set to ``True`` to ignore the error if the table does not exist :param ignore: Set to ``True`` to ignore the error if the table does not exist
""" """
try: try:
self.db.execute("DROP TABLE [{}]".format(self.name)) self.db.execute("DROP TABLE {}".format(quote_identifier(self.name)))
except sqlite3.OperationalError: except sqlite3.OperationalError:
if not ignore: if not ignore:
raise raise
@ -2431,29 +2450,29 @@ class Table(Queryable):
textwrap.dedent( textwrap.dedent(
""" """
{create_counts_table} {create_counts_table}
CREATE TRIGGER IF NOT EXISTS [{table}{counts_table}_insert] AFTER INSERT ON [{table}] CREATE TRIGGER IF NOT EXISTS {trigger_insert} AFTER INSERT ON {table}
BEGIN BEGIN
INSERT OR REPLACE INTO [{counts_table}] INSERT OR REPLACE INTO {counts_table}
VALUES ( VALUES (
{table_quoted}, {table_quoted},
COALESCE( COALESCE(
(SELECT count FROM [{counts_table}] WHERE [table] = {table_quoted}), (SELECT count FROM {counts_table} WHERE "table" = {table_quoted}),
0 0
) + 1 ) + 1
); );
END; END;
CREATE TRIGGER IF NOT EXISTS [{table}{counts_table}_delete] AFTER DELETE ON [{table}] CREATE TRIGGER IF NOT EXISTS {trigger_delete} AFTER DELETE ON {table}
BEGIN BEGIN
INSERT OR REPLACE INTO [{counts_table}] INSERT OR REPLACE INTO {counts_table}
VALUES ( VALUES (
{table_quoted}, {table_quoted},
COALESCE( COALESCE(
(SELECT count FROM [{counts_table}] WHERE [table] = {table_quoted}), (SELECT count FROM {counts_table} WHERE "table" = {table_quoted}),
0 0
) - 1 ) - 1
); );
END; END;
INSERT OR REPLACE INTO _counts VALUES ({table_quoted}, (select count(*) from [{table}])); INSERT OR REPLACE INTO _counts VALUES ({table_quoted}, (select count(*) from {table}));
""" """
) )
.strip() .strip()
@ -2461,9 +2480,15 @@ class Table(Queryable):
create_counts_table=_COUNTS_TABLE_CREATE_SQL.format( create_counts_table=_COUNTS_TABLE_CREATE_SQL.format(
self.db._counts_table_name self.db._counts_table_name
), ),
counts_table=self.db._counts_table_name, counts_table=quote_identifier(self.db._counts_table_name),
table=self.name, table=quote_identifier(self.name),
table_quoted=self.db.quote(self.name), table_quoted=self.db.quote(self.name),
trigger_insert=quote_identifier(
self.name + self.db._counts_table_name + "_insert"
),
trigger_delete=quote_identifier(
self.name + self.db._counts_table_name + "_delete"
),
) )
) )
with self.db.conn: with self.db.conn:
@ -2503,16 +2528,17 @@ class Table(Queryable):
create_fts_sql = ( create_fts_sql = (
textwrap.dedent( textwrap.dedent(
""" """
CREATE VIRTUAL TABLE [{table}_fts] USING {fts_version} ( CREATE VIRTUAL TABLE {table_fts} USING {fts_version} (
{columns},{tokenize} {columns},{tokenize}
content=[{table}] content={table}
) )
""" """
) )
.strip() .strip()
.format( .format(
table=self.name, table=quote_identifier(self.name),
columns=", ".join("[{}]".format(c) for c in columns), table_fts=quote_identifier(self.name + "_fts"),
columns=", ".join(quote_identifier(c) for c in columns),
fts_version=fts_version, fts_version=fts_version,
tokenize="\n tokenize='{}',".format(tokenize) if tokenize else "", tokenize="\n tokenize='{}',".format(tokenize) if tokenize else "",
) )
@ -2539,27 +2565,34 @@ class Table(Queryable):
self.populate_fts(columns) self.populate_fts(columns)
if create_triggers: if create_triggers:
old_cols = ", ".join("old.[{}]".format(c) for c in columns) old_cols = ", ".join("old.{}".format(quote_identifier(c)) for c in columns)
new_cols = ", ".join("new.[{}]".format(c) for c in columns) new_cols = ", ".join("new.{}".format(quote_identifier(c)) for c in columns)
columns_quoted = ", ".join(quote_identifier(c) for c in columns)
table = quote_identifier(self.name)
table_fts = quote_identifier(self.name + "_fts")
triggers = ( triggers = (
textwrap.dedent( textwrap.dedent(
""" """
CREATE TRIGGER [{table}_ai] AFTER INSERT ON [{table}] BEGIN CREATE TRIGGER {table_ai} AFTER INSERT ON {table} BEGIN
INSERT INTO [{table}_fts] (rowid, {columns}) VALUES (new.rowid, {new_cols}); INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
END; END;
CREATE TRIGGER [{table}_ad] AFTER DELETE ON [{table}] BEGIN CREATE TRIGGER {table_ad} AFTER DELETE ON {table} BEGIN
INSERT INTO [{table}_fts] ([{table}_fts], rowid, {columns}) VALUES('delete', old.rowid, {old_cols}); INSERT INTO {table_fts} ({table_fts}, rowid, {columns}) VALUES('delete', old.rowid, {old_cols});
END; END;
CREATE TRIGGER [{table}_au] AFTER UPDATE ON [{table}] BEGIN CREATE TRIGGER {table_au} AFTER UPDATE ON {table} BEGIN
INSERT INTO [{table}_fts] ([{table}_fts], rowid, {columns}) VALUES('delete', old.rowid, {old_cols}); INSERT INTO {table_fts} ({table_fts}, rowid, {columns}) VALUES('delete', old.rowid, {old_cols});
INSERT INTO [{table}_fts] (rowid, {columns}) VALUES (new.rowid, {new_cols}); INSERT INTO {table_fts} (rowid, {columns}) VALUES (new.rowid, {new_cols});
END; END;
""" """
) )
.strip() .strip()
.format( .format(
table=self.name, table=table,
columns=", ".join("[{}]".format(c) for c in columns), table_fts=table_fts,
table_ai=quote_identifier(self.name + "_ai"),
table_ad=quote_identifier(self.name + "_ad"),
table_au=quote_identifier(self.name + "_au"),
columns=columns_quoted,
old_cols=old_cols, old_cols=old_cols,
new_cols=new_cols, new_cols=new_cols,
) )
@ -2574,16 +2607,19 @@ class Table(Queryable):
:param columns: Columns to populate the data for :param columns: Columns to populate the data for
""" """
columns_quoted = ", ".join(quote_identifier(c) for c in columns)
sql = ( sql = (
textwrap.dedent( textwrap.dedent(
""" """
INSERT INTO [{table}_fts] (rowid, {columns}) INSERT INTO {table_fts} (rowid, {columns})
SELECT rowid, {columns} FROM [{table}]; SELECT rowid, {columns} FROM {table};
""" """
) )
.strip() .strip()
.format( .format(
table=self.name, columns=", ".join("[{}]".format(c) for c in columns) table=quote_identifier(self.name),
table_fts=quote_identifier(self.name + "_fts"),
columns=columns_quoted,
) )
) )
self.db.executescript(sql) self.db.executescript(sql)
@ -2600,18 +2636,20 @@ class Table(Queryable):
""" """
SELECT name FROM sqlite_master SELECT name FROM sqlite_master
WHERE type = 'trigger' WHERE type = 'trigger'
AND sql LIKE '% INSERT INTO [{}]%' AND (sql LIKE '% INSERT INTO [{}]%' OR sql LIKE '% INSERT INTO "{}"%')
""" """
) )
.strip() .strip()
.format(fts_table) .format(fts_table, fts_table)
) )
trigger_names = [] trigger_names = []
for row in self.db.execute(sql).fetchall(): for row in self.db.execute(sql).fetchall():
trigger_names.append(row[0]) trigger_names.append(row[0])
with self.db.conn: with self.db.conn:
for trigger_name in trigger_names: for trigger_name in trigger_names:
self.db.execute("DROP TRIGGER IF EXISTS [{}]".format(trigger_name)) self.db.execute(
"DROP TRIGGER IF EXISTS {}".format(quote_identifier(trigger_name))
)
return self return self
def rebuild_fts(self): def rebuild_fts(self):
@ -2621,8 +2659,8 @@ class Table(Queryable):
# Assume this is itself an FTS table # Assume this is itself an FTS table
fts_table = self.name fts_table = self.name
self.db.execute( self.db.execute(
"INSERT INTO [{table}]([{table}]) VALUES('rebuild');".format( "INSERT INTO {table}({table}) VALUES('rebuild');".format(
table=fts_table table=quote_identifier(fts_table)
) )
) )
return self return self
@ -2644,7 +2682,7 @@ class Table(Queryable):
""" """
).strip() ).strip()
args = { args = {
"like": "%VIRTUAL TABLE%USING FTS%content=[{}]%".format(self.name), "like": '%VIRTUAL TABLE%USING FTS%content="{}"%'.format(self.name),
"like2": '%VIRTUAL TABLE%USING FTS%content="{}"%'.format(self.name), "like2": '%VIRTUAL TABLE%USING FTS%content="{}"%'.format(self.name),
"table": self.name, "table": self.name,
} }
@ -2660,9 +2698,9 @@ class Table(Queryable):
if fts_table is not None: if fts_table is not None:
self.db.execute( self.db.execute(
""" """
INSERT INTO [{table}] ([{table}]) VALUES ("optimize"); INSERT INTO {table} ({table}) VALUES ("optimize");
""".strip().format( """.strip().format(
table=fts_table table=quote_identifier(fts_table)
) )
) )
return self return self
@ -2688,17 +2726,19 @@ class Table(Queryable):
""" """
# Pick names for table and rank column that don't clash # Pick names for table and rank column that don't clash
original = "original_" if self.name == "original" else "original" original = "original_" if self.name == "original" else "original"
original_quoted = quote_identifier(original)
columns_sql = "*" columns_sql = "*"
columns_with_prefix_sql = "[{}].*".format(original) columns_with_prefix_sql = "{}.*".format(original_quoted)
if columns: if columns:
columns_sql = ",\n ".join("[{}]".format(c) for c in columns) columns_sql = ",\n ".join(quote_identifier(c) for c in columns)
columns_with_prefix_sql = ",\n ".join( columns_with_prefix_sql = ",\n ".join(
"[{}].[{}]".format(original, c) for c in columns "{}.{}".format(original_quoted, quote_identifier(c)) for c in columns
) )
fts_table = self.detect_fts() fts_table = self.detect_fts()
assert fts_table, "Full-text search is not configured for table '{}'".format( assert fts_table, "Full-text search is not configured for table '{}'".format(
self.name self.name
) )
fts_table_quoted = quote_identifier(fts_table)
virtual_table_using = self.db[fts_table].virtual_table_using virtual_table_using = self.db[fts_table].virtual_table_using
sql = textwrap.dedent( sql = textwrap.dedent(
""" """
@ -2706,26 +2746,26 @@ class Table(Queryable):
select select
rowid, rowid,
{columns} {columns}
from [{dbtable}]{where_clause} from {dbtable}{where_clause}
) )
select select
{columns_with_prefix} {columns_with_prefix}
from from
[{original}] {original}
join [{fts_table}] on [{original}].rowid = [{fts_table}].rowid join {fts_table} on {original}.rowid = {fts_table}.rowid
where where
[{fts_table}] match :query {fts_table} match :query
order by order by
{order_by} {order_by}
{limit_offset} {limit_offset}
""" """
).strip() ).strip()
if virtual_table_using == "FTS5": if virtual_table_using == "FTS5":
rank_implementation = "[{}].rank".format(fts_table) rank_implementation = "{}.rank".format(fts_table_quoted)
else: else:
self.db.register_fts4_bm25() self.db.register_fts4_bm25()
rank_implementation = "rank_bm25(matchinfo([{}], 'pcnalx'))".format( rank_implementation = "rank_bm25(matchinfo({}, 'pcnalx'))".format(
fts_table fts_table_quoted
) )
if include_rank: if include_rank:
columns_with_prefix_sql += ",\n " + rank_implementation + " rank" columns_with_prefix_sql += ",\n " + rank_implementation + " rank"
@ -2735,12 +2775,12 @@ class Table(Queryable):
if offset is not None: if offset is not None:
limit_offset += " offset {}".format(offset) limit_offset += " offset {}".format(offset)
return sql.format( return sql.format(
dbtable=self.name, dbtable=quote_identifier(self.name),
where_clause="\n where {}".format(where) if where else "", where_clause="\n where {}".format(where) if where else "",
original=original, original=original_quoted,
columns=columns_sql, columns=columns_sql,
columns_with_prefix=columns_with_prefix_sql, columns_with_prefix=columns_with_prefix_sql,
fts_table=fts_table, fts_table=fts_table_quoted,
order_by=order_by or rank_implementation, order_by=order_by or rank_implementation,
limit_offset=limit_offset.strip(), limit_offset=limit_offset.strip(),
).strip() ).strip()
@ -2808,9 +2848,9 @@ class Table(Queryable):
if not isinstance(pk_values, (list, tuple)): if not isinstance(pk_values, (list, tuple)):
pk_values = [pk_values] pk_values = [pk_values]
self.get(pk_values) self.get(pk_values)
wheres = ["[{}] = ?".format(pk_name) for pk_name in self.pks] wheres = ["{} = ?".format(quote_identifier(pk_name)) for pk_name in self.pks]
sql = "delete from [{table}] where {wheres}".format( sql = "delete from {} where {wheres}".format(
table=self.name, wheres=" and ".join(wheres) quote_identifier(self.name), wheres=" and ".join(wheres)
) )
with self.db.conn: with self.db.conn:
self.db.execute(sql, pk_values) self.db.execute(sql, pk_values)
@ -2834,7 +2874,7 @@ class Table(Queryable):
""" """
if not self.exists(): if not self.exists():
return self return self
sql = "delete from [{}]".format(self.name) sql = "delete from {}".format(quote_identifier(self.name))
if where is not None: if where is not None:
sql += " where " + where sql += " where " + where
self.db.execute(sql, where_args or []) self.db.execute(sql, where_args or [])
@ -2875,12 +2915,16 @@ class Table(Queryable):
pks = self.pks pks = self.pks
validate_column_names(updates.keys()) validate_column_names(updates.keys())
for key, value in updates.items(): for key, value in updates.items():
sets.append("[{}] = {}".format(key, conversions.get(key, "?"))) sets.append(
"{} = {}".format(quote_identifier(key), conversions.get(key, "?"))
)
args.append(jsonify_if_needed(value)) args.append(jsonify_if_needed(value))
wheres = ["[{}] = ?".format(pk_name) for pk_name in pks] wheres = ["{} = ?".format(quote_identifier(pk_name)) for pk_name in pks]
args.extend(pk_values) args.extend(pk_values)
sql = "update [{table}] set {sets} where {wheres}".format( sql = "update {} set {sets} where {wheres}".format(
table=self.name, sets=", ".join(sets), wheres=" and ".join(wheres) quote_identifier(self.name),
sets=", ".join(sets),
wheres=" and ".join(wheres),
) )
with self.db.conn: with self.db.conn:
try: try:
@ -2957,14 +3001,14 @@ class Table(Queryable):
if fn_name == "<lambda>": if fn_name == "<lambda>":
fn_name = f"lambda_{abs(hash(fn))}" fn_name = f"lambda_{abs(hash(fn))}"
self.db.register_function(convert_value, name=fn_name) self.db.register_function(convert_value, name=fn_name)
sql = "update [{table}] set {sets}{where};".format( sql = "update {} set {sets}{where};".format(
table=self.name, quote_identifier(self.name),
sets=", ".join( sets=", ".join(
[ [
"[{output_column}] = {fn_name}([{column}])".format( "{} = {}({})".format(
output_column=output or column, quote_identifier(output or column),
column=column, fn_name,
fn_name=fn_name, quote_identifier(column),
) )
for column in columns for column in columns
] ]
@ -2992,7 +3036,7 @@ class Table(Queryable):
) as bar: ) as bar:
for row in self.rows_where( for row in self.rows_where(
select=", ".join( select=", ".join(
"[{}]".format(column_name) for column_name in (pks + [column]) quote_identifier(column_name) for column_name in (pks + [column])
), ),
where=where, where=where,
where_args=where_args, where_args=where_args,
@ -3097,7 +3141,7 @@ class Table(Queryable):
record_values.append(value) record_values.append(value)
values.append(record_values) values.append(record_values)
columns_sql = ", ".join(f"[{c}]" for c in all_columns) columns_sql = ", ".join(quote_identifier(c) for c in all_columns)
placeholder_expr = ", ".join(conversions.get(c, "?") for c in all_columns) placeholder_expr = ", ".join(conversions.get(c, "?") for c in all_columns)
row_placeholders_sql = ", ".join(f"({placeholder_expr})" for _ in values) row_placeholders_sql = ", ".join(f"({placeholder_expr})" for _ in values)
flat_params = list(itertools.chain.from_iterable(values)) flat_params = list(itertools.chain.from_iterable(values))
@ -3105,7 +3149,7 @@ class Table(Queryable):
# replace=True mean INSERT OR REPLACE INTO # replace=True mean INSERT OR REPLACE INTO
if replace: if replace:
sql = ( sql = (
f"INSERT OR REPLACE INTO [{self.name}] " f"INSERT OR REPLACE INTO {quote_identifier(self.name)} "
f"({columns_sql}) VALUES {row_placeholders_sql}" f"({columns_sql}) VALUES {row_placeholders_sql}"
) )
return [(sql, flat_params)] return [(sql, flat_params)]
@ -3116,7 +3160,7 @@ class Table(Queryable):
if ignore: if ignore:
or_ignore = " OR IGNORE" or_ignore = " OR IGNORE"
sql = ( sql = (
f"INSERT{or_ignore} INTO [{self.name}] " f"INSERT{or_ignore} INTO {quote_identifier(self.name)} "
f"({columns_sql}) VALUES {row_placeholders_sql}" f"({columns_sql}) VALUES {row_placeholders_sql}"
) )
return [(sql, flat_params)] return [(sql, flat_params)]
@ -3124,26 +3168,27 @@ class Table(Queryable):
# Everything from here on is for upsert=True # Everything from here on is for upsert=True
pk_cols = [pk] if isinstance(pk, str) else list(pk) pk_cols = [pk] if isinstance(pk, str) else list(pk)
non_pk_cols = [c for c in all_columns if c not in pk_cols] non_pk_cols = [c for c in all_columns if c not in pk_cols]
conflict_sql = ", ".join(f"[{c}]" for c in pk_cols) conflict_sql = ", ".join(quote_identifier(c) for c in pk_cols)
if self.db.supports_on_conflict and not self.db.use_old_upsert: if self.db.supports_on_conflict and not self.db.use_old_upsert:
if non_pk_cols: if non_pk_cols:
# DO UPDATE # DO UPDATE
assignments = [] assignments = []
for c in non_pk_cols: for c in non_pk_cols:
c_quoted = quote_identifier(c)
if c in conversions: if c in conversions:
assignments.append( assignments.append(
f"[{c}] = {conversions[c].replace('?', f'excluded.[{c}]')}" f"{c_quoted} = {conversions[c].replace('?', f'excluded.{c_quoted}')}"
) )
else: else:
assignments.append(f"[{c}] = excluded.[{c}]") assignments.append(f"{c_quoted} = excluded.{c_quoted}")
do_clause = "DO UPDATE SET " + ", ".join(assignments) do_clause = "DO UPDATE SET " + ", ".join(assignments)
else: else:
# All columns are in the PK – nothing to update. # All columns are in the PK – nothing to update.
do_clause = "DO NOTHING" do_clause = "DO NOTHING"
sql = ( sql = (
f"INSERT INTO [{self.name}] ({columns_sql}) " f"INSERT INTO {quote_identifier(self.name)} ({columns_sql}) "
f"VALUES {row_placeholders_sql} " f"VALUES {row_placeholders_sql} "
f"ON CONFLICT({conflict_sql}) {do_clause}" f"ON CONFLICT({conflict_sql}) {do_clause}"
) )
@ -3164,24 +3209,30 @@ class Table(Queryable):
# them since it ignores the resulting integrity errors # them since it ignores the resulting integrity errors
if not_null: if not_null:
placeholders.extend(not_null) placeholders.extend(not_null)
sql = "INSERT OR IGNORE INTO [{table}]({cols}) VALUES({placeholders});".format( sql = (
table=self.name, "INSERT OR IGNORE INTO {table}({cols}) VALUES({placeholders});".format(
cols=", ".join(["[{}]".format(p) for p in placeholders]), table=quote_identifier(self.name),
placeholders=", ".join(["?" for p in placeholders]), cols=", ".join([quote_identifier(p) for p in placeholders]),
placeholders=", ".join(["?" for p in placeholders]),
)
) )
queries_and_params.append( queries_and_params.append(
(sql, [record[col] for col in pks] + ["" for _ in (not_null or [])]) (sql, [record[col] for col in pks] + ["" for _ in (not_null or [])])
) )
# UPDATE [book] SET [name] = 'Programming' WHERE [id] = 1001; # UPDATE "book" SET "name" = 'Programming' WHERE "id" = 1001;
set_cols = [col for col in all_columns if col not in pks] set_cols = [col for col in all_columns if col not in pks]
if set_cols: if set_cols:
sql2 = "UPDATE [{table}] SET {pairs} WHERE {wheres}".format( sql2 = "UPDATE {} SET {pairs} WHERE {wheres}".format(
table=self.name, quote_identifier(self.name),
pairs=", ".join( pairs=", ".join(
"[{}] = {}".format(col, conversions.get(col, "?")) "{} = {}".format(
quote_identifier(col), conversions.get(col, "?")
)
for col in set_cols for col in set_cols
), ),
wheres=" AND ".join("[{}] = ?".format(pk) for pk in pks), wheres=" AND ".join(
"{} = ?".format(quote_identifier(pk)) for pk in pks
),
) )
queries_and_params.append( queries_and_params.append(
( (
@ -3469,7 +3520,7 @@ class Table(Queryable):
self.last_rowid = None self.last_rowid = None
self.last_pk = None self.last_pk = None
if truncate and self.exists(): if truncate and self.exists():
self.db.execute("DELETE FROM [{}];".format(self.name)) self.db.execute("DELETE FROM {};".format(quote_identifier(self.name)))
result = None result = None
for chunk in chunks(itertools.chain([first_record], records_iter), batch_size): for chunk in chunks(itertools.chain([first_record], records_iter), batch_size):
chunk = list(chunk) chunk = list(chunk)
@ -3724,7 +3775,9 @@ class Table(Queryable):
unique_column_sets = [set(i.columns) for i in self.indexes] unique_column_sets = [set(i.columns) for i in self.indexes]
if set(lookup_values.keys()) not in unique_column_sets: if set(lookup_values.keys()) not in unique_column_sets:
self.create_index(lookup_values.keys(), unique=True) self.create_index(lookup_values.keys(), unique=True)
wheres = ["[{}] = ?".format(column) for column in lookup_values] wheres = [
"{} = ?".format(quote_identifier(column)) for column in lookup_values
]
rows = list( rows = list(
self.rows_where( self.rows_where(
" and ".join(wheres), [value for _, value in lookup_values.items()] " and ".join(wheres), [value for _, value in lookup_values.items()]
@ -3887,20 +3940,24 @@ class Table(Queryable):
value = value[:value_truncate] + "..." value = value[:value_truncate] + "..."
return value return value
table_quoted = quote_identifier(table)
column_quoted = quote_identifier(column)
num_null = db.execute( num_null = db.execute(
"select count(*) from [{}] where [{}] is null".format(table, column) "select count(*) from {} where {} is null".format(
table_quoted, column_quoted
)
).fetchone()[0] ).fetchone()[0]
num_blank = db.execute( num_blank = db.execute(
"select count(*) from [{}] where [{}] = ''".format(table, column) "select count(*) from {} where {} = ''".format(table_quoted, column_quoted)
).fetchone()[0] ).fetchone()[0]
num_distinct = db.execute( num_distinct = db.execute(
"select count(distinct [{}]) from [{}]".format(column, table) "select count(distinct {}) from {}".format(column_quoted, table_quoted)
).fetchone()[0] ).fetchone()[0]
most_common_results = None most_common_results = None
least_common_results = None least_common_results = None
if num_distinct == 1: if num_distinct == 1:
value = db.execute( value = db.execute(
"select [{}] from [{}] limit 1".format(column, table) "select {} from {} limit 1".format(column_quoted, table_quoted)
).fetchone()[0] ).fetchone()[0]
most_common_results = [(truncate(value), total_rows)] most_common_results = [(truncate(value), total_rows)]
elif num_distinct != total_rows: elif num_distinct != total_rows:
@ -3912,8 +3969,12 @@ class Table(Queryable):
most_common_results = [ most_common_results = [
(truncate(r[0]), r[1]) (truncate(r[0]), r[1])
for r in db.execute( for r in db.execute(
"select [{}], count(*) from [{}] group by [{}] order by count(*) desc, [{}] limit {}".format( "select {}, count(*) from {} group by {} order by count(*) desc, {} limit {}".format(
column, table, column, column, common_limit column_quoted,
table_quoted,
column_quoted,
column_quoted,
common_limit,
) )
).fetchall() ).fetchall()
] ]
@ -3926,8 +3987,12 @@ class Table(Queryable):
least_common_results = [ least_common_results = [
(truncate(r[0]), r[1]) (truncate(r[0]), r[1])
for r in db.execute( for r in db.execute(
"select [{}], count(*) from [{}] group by [{}] order by count(*), [{}] desc limit {}".format( "select {}, count(*) from {} group by {} order by count(*), {} desc limit {}".format(
column, table, column, column, common_limit column_quoted,
table_quoted,
column_quoted,
column_quoted,
common_limit,
) )
).fetchall() ).fetchall()
] ]
@ -4045,7 +4110,7 @@ class View(Queryable):
""" """
try: try:
self.db.execute("DROP VIEW [{}]".format(self.name)) self.db.execute("DROP VIEW {}".format(quote_identifier(self.name)))
except sqlite3.OperationalError: except sqlite3.OperationalError:
if not ignore: if not ignore:
raise raise
@ -4083,22 +4148,17 @@ def resolve_extracts(
def validate_column_names(columns): def validate_column_names(columns):
# Validate no columns contain '[' or ']' - #86 # Validate no columns contain problematic characters
for column in columns: # With double-quote identifier escaping, embedded quotes are handled
assert ( # by the quote_identifier function, so no validation is needed
"[" not in column and "]" not in column pass
), "'[' and ']' cannot be used in column names"
def fix_square_braces(records: Iterable[Dict[str, Any]]): def fix_square_braces(records: Iterable[Dict[str, Any]]):
for record in records: # Legacy function name kept for backward compatibility
if any("[" in key or "]" in key for key in record.keys()): # With double-quote identifier escaping, embedded quotes are handled
yield { # by the quote_identifier function, so records pass through unchanged
key.replace("[", "_").replace("]", "_"): value yield from records
for key, value in record.items()
}
else:
yield record
def _decode_default_value(value): def _decode_default_value(value):

266
tests/test_cli.py
View file

@ -132,7 +132,7 @@ def test_tables_schema(db_path):
assert ( assert (
'[{"table": "Gosh", "schema": "CREATE TABLE Gosh (c1 text, c2 text, c3 text)"},\n' '[{"table": "Gosh", "schema": "CREATE TABLE Gosh (c1 text, c2 text, c3 text)"},\n'
' {"table": "Gosh2", "schema": "CREATE TABLE Gosh2 (c1 text, c2 text, c3 text)"},\n' ' {"table": "Gosh2", "schema": "CREATE TABLE Gosh2 (c1 text, c2 text, c3 text)"},\n'
' {"table": "lots", "schema": "CREATE TABLE [lots] (\\n [id] INTEGER,\\n [age] INTEGER\\n)"}]' ' {"table": "lots", "schema": "CREATE TABLE \\"lots\\" (\\n \\"id\\" INTEGER,\\n \\"age\\" INTEGER\\n)"}]'
) == result.output.strip() ) == result.output.strip()
@ -264,38 +264,38 @@ def test_create_index_desc(db_path):
assert result.exit_code == 0 assert result.exit_code == 0
assert ( assert (
db.execute("select sql from sqlite_master where type='index'").fetchone()[0] db.execute("select sql from sqlite_master where type='index'").fetchone()[0]
== "CREATE INDEX [idx_Gosh_c1]\n ON [Gosh] ([c1] desc)" == 'CREATE INDEX "idx_Gosh_c1"\n ON "Gosh" ("c1" desc)'
) )
@pytest.mark.parametrize( @pytest.mark.parametrize(
"col_name,col_type,expected_schema", "col_name,col_type,expected_schema",
( (
("text", "TEXT", "CREATE TABLE [dogs] (\n [name] TEXT\n, [text] TEXT)"), ("text", "TEXT", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "text" TEXT)'),
("text", "str", "CREATE TABLE [dogs] (\n [name] TEXT\n, [text] TEXT)"), ("text", "str", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "text" TEXT)'),
("text", "STR", "CREATE TABLE [dogs] (\n [name] TEXT\n, [text] TEXT)"), ("text", "STR", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "text" TEXT)'),
( (
"integer", "integer",
"INTEGER", "INTEGER",
"CREATE TABLE [dogs] (\n [name] TEXT\n, [integer] INTEGER)", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "integer" INTEGER)',
), ),
( (
"integer", "integer",
"int", "int",
"CREATE TABLE [dogs] (\n [name] TEXT\n, [integer] INTEGER)", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "integer" INTEGER)',
), ),
("float", "FLOAT", "CREATE TABLE [dogs] (\n [name] TEXT\n, [float] FLOAT)"), ("float", "FLOAT", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "float" FLOAT)'),
("blob", "blob", "CREATE TABLE [dogs] (\n [name] TEXT\n, [blob] BLOB)"), ("blob", "blob", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'),
("blob", "BLOB", "CREATE TABLE [dogs] (\n [name] TEXT\n, [blob] BLOB)"), ("blob", "BLOB", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'),
("blob", "bytes", "CREATE TABLE [dogs] (\n [name] TEXT\n, [blob] BLOB)"), ("blob", "bytes", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'),
("blob", "BYTES", "CREATE TABLE [dogs] (\n [name] TEXT\n, [blob] BLOB)"), ("blob", "BYTES", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'),
("default", None, "CREATE TABLE [dogs] (\n [name] TEXT\n, [default] TEXT)"), ("default", None, 'CREATE TABLE "dogs" (\n "name" TEXT\n, "default" TEXT)'),
), ),
) )
def test_add_column(db_path, col_name, col_type, expected_schema): def test_add_column(db_path, col_name, col_type, expected_schema):
db = Database(db_path) db = Database(db_path)
db.create_table("dogs", {"name": str}) db.create_table("dogs", {"name": str})
assert db["dogs"].schema == "CREATE TABLE [dogs] (\n [name] TEXT\n)" assert db["dogs"].schema == 'CREATE TABLE "dogs" (\n "name" TEXT\n)'
args = ["add-column", db_path, "dogs", col_name] args = ["add-column", db_path, "dogs", col_name]
if col_type is not None: if col_type is not None:
args.append(col_type) args.append(col_type)
@ -319,7 +319,7 @@ def test_add_column_ignore(db_path, ignore):
def test_add_column_not_null_default(db_path): def test_add_column_not_null_default(db_path):
db = Database(db_path) db = Database(db_path)
db.create_table("dogs", {"name": str}) db.create_table("dogs", {"name": str})
assert db["dogs"].schema == "CREATE TABLE [dogs] (\n [name] TEXT\n)" assert db["dogs"].schema == 'CREATE TABLE "dogs" (\n "name" TEXT\n)'
args = [ args = [
"add-column", "add-column",
db_path, db_path,
@ -330,9 +330,9 @@ def test_add_column_not_null_default(db_path):
] ]
assert CliRunner().invoke(cli.cli, args).exit_code == 0 assert CliRunner().invoke(cli.cli, args).exit_code == 0
assert db["dogs"].schema == ( assert db["dogs"].schema == (
"CREATE TABLE [dogs] (\n" 'CREATE TABLE "dogs" (\n'
" [name] TEXT\n" ' "name" TEXT\n'
", [nickname] TEXT NOT NULL DEFAULT 'dogs''dawg')" ", \"nickname\" TEXT NOT NULL DEFAULT 'dogs''dawg')"
) )
@ -403,8 +403,8 @@ def test_add_column_foreign_key(db_path):
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
assert db["books"].schema == ( assert db["books"].schema == (
'CREATE TABLE "books" (\n' 'CREATE TABLE "books" (\n'
" [title] TEXT,\n" ' "title" TEXT,\n'
" [author_id] INTEGER REFERENCES [authors]([id])\n" ' "author_id" INTEGER REFERENCES "authors"("id")\n'
")" ")"
) )
# Try it again with a custom --fk-col # Try it again with a custom --fk-col
@ -424,9 +424,9 @@ def test_add_column_foreign_key(db_path):
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
assert db["books"].schema == ( assert db["books"].schema == (
'CREATE TABLE "books" (\n' 'CREATE TABLE "books" (\n'
" [title] TEXT,\n" ' "title" TEXT,\n'
" [author_id] INTEGER REFERENCES [authors]([id]),\n" ' "author_id" INTEGER REFERENCES "authors"("id"),\n'
" [author_name_ref] TEXT REFERENCES [authors]([name])\n" ' "author_name_ref" TEXT REFERENCES "authors"("name")\n'
")" ")"
) )
# Throw an error if the --fk table does not exist # Throw an error if the --fk table does not exist
@ -492,10 +492,10 @@ def test_enable_fts(db_path):
assert "http://example.com_fts" == db["http://example.com"].detect_fts() assert "http://example.com_fts" == db["http://example.com"].detect_fts()
# Check tokenize was set to porter # Check tokenize was set to porter
assert ( assert (
"CREATE VIRTUAL TABLE [http://example.com_fts] USING FTS4 (\n" 'CREATE VIRTUAL TABLE "http://example.com_fts" USING FTS4 (\n'
" [c1],\n" ' "c1",\n'
" tokenize='porter',\n" " tokenize='porter',\n"
" content=[http://example.com]" ' content="http://example.com"'
"\n)" "\n)"
) == db["http://example.com_fts"].schema ) == db["http://example.com_fts"].schema
db["http://example.com"].drop() db["http://example.com"].drop()
@ -516,7 +516,7 @@ def test_enable_fts_replace(db_path):
cli.cli, ["enable-fts", db_path, "Gosh", "c1", "--fts4"] cli.cli, ["enable-fts", db_path, "Gosh", "c1", "--fts4"]
) )
assert result2.exit_code == 1 assert result2.exit_code == 1
assert result2.output == "Error: table [Gosh_fts] already exists\n" assert result2.output == 'Error: table "Gosh_fts" already exists\n'
# This should work # This should work
result3 = CliRunner().invoke( result3 = CliRunner().invoke(
@ -1139,7 +1139,7 @@ def test_upsert_alter(db_path, tmpdir):
"age", "age",
"integer", "integer",
], ],
("CREATE TABLE [t] (\n [name] TEXT,\n [age] INTEGER\n)"), ('CREATE TABLE "t" (\n "name" TEXT,\n "age" INTEGER\n)'),
), ),
# All types: # All types:
( (
@ -1158,31 +1158,31 @@ def test_upsert_alter(db_path, tmpdir):
"id", "id",
], ],
( (
"CREATE TABLE [t] (\n" 'CREATE TABLE "t" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [age] INTEGER,\n" ' "age" INTEGER,\n'
" [weight] FLOAT,\n" ' "weight" FLOAT,\n'
" [thumbnail] BLOB\n" ' "thumbnail" BLOB\n'
")" ")"
), ),
), ),
# Not null: # Not null:
( (
["name", "text", "--not-null", "name"], ["name", "text", "--not-null", "name"],
("CREATE TABLE [t] (\n" " [name] TEXT NOT NULL\n" ")"), ('CREATE TABLE "t" (\n' ' "name" TEXT NOT NULL\n' ")"),
), ),
# Default: # Default:
( (
["age", "integer", "--default", "age", "3"], ["age", "integer", "--default", "age", "3"],
("CREATE TABLE [t] (\n" " [age] INTEGER DEFAULT '3'\n" ")"), ('CREATE TABLE "t" (\n' " \"age\" INTEGER DEFAULT '3'\n" ")"),
), ),
# Compound primary key # Compound primary key
( (
["category", "text", "name", "text", "--pk", "category", "--pk", "name"], ["category", "text", "name", "text", "--pk", "category", "--pk", "name"],
( (
"CREATE TABLE [t] (\n [category] TEXT,\n [name] TEXT,\n" 'CREATE TABLE "t" (\n "category" TEXT,\n "name" TEXT,\n'
" PRIMARY KEY ([category], [name])\n)" ' PRIMARY KEY ("category", "name")\n)'
), ),
), ),
], ],
@ -1233,16 +1233,16 @@ def test_create_table_foreign_key():
assert result.exit_code == 0 assert result.exit_code == 0
db = Database("books.db") db = Database("books.db")
assert ( assert (
"CREATE TABLE [authors] (\n" 'CREATE TABLE "authors" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT\n" ' "name" TEXT\n'
")" ")"
) == db["authors"].schema ) == db["authors"].schema
assert ( assert (
"CREATE TABLE [books] (\n" 'CREATE TABLE "books" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [title] TEXT,\n" ' "title" TEXT,\n'
" [author_id] INTEGER REFERENCES [authors]([id])\n" ' "author_id" INTEGER REFERENCES "authors"("id")\n'
")" ")"
) == db["books"].schema ) == db["books"].schema
@ -1271,7 +1271,7 @@ def test_create_table_ignore():
cli.cli, ["create-table", "test.db", "dogs", "id", "integer", "--ignore"] cli.cli, ["create-table", "test.db", "dogs", "id", "integer", "--ignore"]
) )
assert result.exit_code == 0 assert result.exit_code == 0
assert "CREATE TABLE [dogs] (\n [name] TEXT\n)" == db["dogs"].schema assert 'CREATE TABLE "dogs" (\n "name" TEXT\n)' == db["dogs"].schema
def test_create_table_replace(): def test_create_table_replace():
@ -1283,7 +1283,7 @@ def test_create_table_replace():
cli.cli, ["create-table", "test.db", "dogs", "id", "integer", "--replace"] cli.cli, ["create-table", "test.db", "dogs", "id", "integer", "--replace"]
) )
assert result.exit_code == 0 assert result.exit_code == 0
assert "CREATE TABLE [dogs] (\n [id] INTEGER\n)" == db["dogs"].schema assert 'CREATE TABLE "dogs" (\n "id" INTEGER\n)' == db["dogs"].schema
def test_create_view(): def test_create_view():
@ -1537,9 +1537,9 @@ def test_add_foreign_keys(db_path):
[], [],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [age] INTEGER NOT NULL DEFAULT '1',\n" " \"age\" INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT\n" ' "name" TEXT\n'
")" ")"
), ),
), ),
@ -1547,9 +1547,9 @@ def test_add_foreign_keys(db_path):
["--type", "age", "text"], ["--type", "age", "text"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [age] TEXT NOT NULL DEFAULT '1',\n" " \"age\" TEXT NOT NULL DEFAULT '1',\n"
" [name] TEXT\n" ' "name" TEXT\n'
")" ")"
), ),
), ),
@ -1557,8 +1557,8 @@ def test_add_foreign_keys(db_path):
["--drop", "age"], ["--drop", "age"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT\n" ' "name" TEXT\n'
")" ")"
), ),
), ),
@ -1566,9 +1566,9 @@ def test_add_foreign_keys(db_path):
["--rename", "age", "age2", "--rename", "id", "pk"], ["--rename", "age", "age2", "--rename", "id", "pk"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [pk] INTEGER PRIMARY KEY,\n" ' "pk" INTEGER PRIMARY KEY,\n'
" [age2] INTEGER NOT NULL DEFAULT '1',\n" " \"age2\" INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT\n" ' "name" TEXT\n'
")" ")"
), ),
), ),
@ -1576,9 +1576,9 @@ def test_add_foreign_keys(db_path):
["--not-null", "name"], ["--not-null", "name"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [age] INTEGER NOT NULL DEFAULT '1',\n" " \"age\" INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT NOT NULL\n" ' "name" TEXT NOT NULL\n'
")" ")"
), ),
), ),
@ -1586,9 +1586,9 @@ def test_add_foreign_keys(db_path):
["--not-null-false", "age"], ["--not-null-false", "age"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [age] INTEGER DEFAULT '1',\n" " \"age\" INTEGER DEFAULT '1',\n"
" [name] TEXT\n" ' "name" TEXT\n'
")" ")"
), ),
), ),
@ -1596,9 +1596,9 @@ def test_add_foreign_keys(db_path):
["--pk", "name"], ["--pk", "name"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [age] INTEGER NOT NULL DEFAULT '1',\n" " \"age\" INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT PRIMARY KEY\n" ' "name" TEXT PRIMARY KEY\n'
")" ")"
), ),
), ),
@ -1606,9 +1606,9 @@ def test_add_foreign_keys(db_path):
["--pk-none"], ["--pk-none"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [age] INTEGER NOT NULL DEFAULT '1',\n" " \"age\" INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT\n" ' "name" TEXT\n'
")" ")"
), ),
), ),
@ -1616,9 +1616,9 @@ def test_add_foreign_keys(db_path):
["--default", "name", "Turnip"], ["--default", "name", "Turnip"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [age] INTEGER NOT NULL DEFAULT '1',\n" " \"age\" INTEGER NOT NULL DEFAULT '1',\n"
" [name] TEXT DEFAULT 'Turnip'\n" " \"name\" TEXT DEFAULT 'Turnip'\n"
")" ")"
), ),
), ),
@ -1626,9 +1626,9 @@ def test_add_foreign_keys(db_path):
["--default-none", "age"], ["--default-none", "age"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [age] INTEGER NOT NULL,\n" ' "age" INTEGER NOT NULL,\n'
" [name] TEXT\n" ' "name" TEXT\n'
")" ")"
), ),
), ),
@ -1636,9 +1636,9 @@ def test_add_foreign_keys(db_path):
["-o", "name", "--column-order", "age", "-o", "id"], ["-o", "name", "--column-order", "age", "-o", "id"],
( (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [age] INTEGER NOT NULL DEFAULT '1',\n" " \"age\" INTEGER NOT NULL DEFAULT '1',\n"
" [id] INTEGER PRIMARY KEY\n" ' "id" INTEGER PRIMARY KEY\n'
")" ")"
), ),
), ),
@ -1667,11 +1667,11 @@ def test_transform(db_path, args, expected_schema):
["--drop-foreign-key", "country"], ["--drop-foreign-key", "country"],
( (
'CREATE TABLE "places" (\n' 'CREATE TABLE "places" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [country] INTEGER,\n" ' "country" INTEGER,\n'
" [city] INTEGER REFERENCES [city]([id]),\n" ' "city" INTEGER REFERENCES "city"("id"),\n'
" [continent] INTEGER\n" ' "continent" INTEGER\n'
")" ")"
), ),
), ),
@ -1679,11 +1679,11 @@ def test_transform(db_path, args, expected_schema):
["--drop-foreign-key", "country", "--drop-foreign-key", "city"], ["--drop-foreign-key", "country", "--drop-foreign-key", "city"],
( (
'CREATE TABLE "places" (\n' 'CREATE TABLE "places" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [country] INTEGER,\n" ' "country" INTEGER,\n'
" [city] INTEGER,\n" ' "city" INTEGER,\n'
" [continent] INTEGER\n" ' "continent" INTEGER\n'
")" ")"
), ),
), ),
@ -1691,11 +1691,11 @@ def test_transform(db_path, args, expected_schema):
["--add-foreign-key", "continent", "continent", "id"], ["--add-foreign-key", "continent", "continent", "id"],
( (
'CREATE TABLE "places" (\n' 'CREATE TABLE "places" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [country] INTEGER REFERENCES [country]([id]),\n" ' "country" INTEGER REFERENCES "country"("id"),\n'
" [city] INTEGER REFERENCES [city]([id]),\n" ' "city" INTEGER REFERENCES "city"("id"),\n'
" [continent] INTEGER REFERENCES [continent]([id])\n" ' "continent" INTEGER REFERENCES "continent"("id")\n'
")" ")"
), ),
), ),
@ -1734,7 +1734,7 @@ def test_transform_add_or_drop_foreign_key(db_path, extra_args, expected_schema)
_common_other_schema = ( _common_other_schema = (
"CREATE TABLE [species] (\n [id] INTEGER PRIMARY KEY,\n [species] TEXT\n)" 'CREATE TABLE "species" (\n "id" INTEGER PRIMARY KEY,\n "species" TEXT\n)'
) )
@ -1745,9 +1745,9 @@ _common_other_schema = (
[], [],
( (
'CREATE TABLE "trees" (\n' 'CREATE TABLE "trees" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [address] TEXT,\n" ' "address" TEXT,\n'
" [species_id] INTEGER REFERENCES [species]([id])\n" ' "species_id" INTEGER REFERENCES "species"("id")\n'
")" ")"
), ),
_common_other_schema, _common_other_schema,
@ -1756,20 +1756,20 @@ _common_other_schema = (
["--table", "custom_table"], ["--table", "custom_table"],
( (
'CREATE TABLE "trees" (\n' 'CREATE TABLE "trees" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [address] TEXT,\n" ' "address" TEXT,\n'
" [custom_table_id] INTEGER REFERENCES [custom_table]([id])\n" ' "custom_table_id" INTEGER REFERENCES "custom_table"("id")\n'
")" ")"
), ),
"CREATE TABLE [custom_table] (\n [id] INTEGER PRIMARY KEY,\n [species] TEXT\n)", 'CREATE TABLE "custom_table" (\n "id" INTEGER PRIMARY KEY,\n "species" TEXT\n)',
), ),
( (
["--fk-column", "custom_fk"], ["--fk-column", "custom_fk"],
( (
'CREATE TABLE "trees" (\n' 'CREATE TABLE "trees" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [address] TEXT,\n" ' "address" TEXT,\n'
" [custom_fk] INTEGER REFERENCES [species]([id])\n" ' "custom_fk" INTEGER REFERENCES "species"("id")\n'
")" ")"
), ),
_common_other_schema, _common_other_schema,
@ -1777,11 +1777,11 @@ _common_other_schema = (
( (
["--rename", "name", "name2"], ["--rename", "name", "name2"],
'CREATE TABLE "trees" (\n' 'CREATE TABLE "trees" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [address] TEXT,\n" ' "address" TEXT,\n'
" [species_id] INTEGER REFERENCES [species]([id])\n" ' "species_id" INTEGER REFERENCES "species"("id")\n'
")", ")",
"CREATE TABLE [species] (\n [id] INTEGER PRIMARY KEY,\n [species] TEXT\n)", 'CREATE TABLE "species" (\n "id" INTEGER PRIMARY KEY,\n "species" TEXT\n)',
), ),
], ],
) )
@ -2036,34 +2036,34 @@ def test_triggers(tmpdir, extra_args, expected):
( (
[], [],
( (
"CREATE TABLE [dogs] (\n" 'CREATE TABLE "dogs" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT\n" ' "name" TEXT\n'
");\n" ");\n"
"CREATE TABLE [chickens] (\n" 'CREATE TABLE "chickens" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [breed] TEXT\n" ' "breed" TEXT\n'
");\n" ");\n"
"CREATE INDEX [idx_chickens_breed]\n" 'CREATE INDEX "idx_chickens_breed"\n'
" ON [chickens] ([breed]);\n" ' ON "chickens" ("breed");\n'
), ),
), ),
( (
["dogs"], ["dogs"],
("CREATE TABLE [dogs] (\n" " [id] INTEGER,\n" " [name] TEXT\n" ")\n"), ('CREATE TABLE "dogs" (\n' ' "id" INTEGER,\n' ' "name" TEXT\n' ")\n"),
), ),
( (
["chickens", "dogs"], ["chickens", "dogs"],
( (
"CREATE TABLE [chickens] (\n" 'CREATE TABLE "chickens" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [breed] TEXT\n" ' "breed" TEXT\n'
")\n" ")\n"
"CREATE TABLE [dogs] (\n" 'CREATE TABLE "dogs" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT\n" ' "name" TEXT\n'
")\n" ")\n"
), ),
), ),
@ -2127,10 +2127,10 @@ def test_import_no_headers(tmpdir, args, tsv):
db = Database(db_path) db = Database(db_path)
schema = db["creatures"].schema schema = db["creatures"].schema
assert schema == ( assert schema == (
"CREATE TABLE [creatures] (\n" 'CREATE TABLE "creatures" (\n'
" [untitled_1] TEXT,\n" ' "untitled_1" TEXT,\n'
" [untitled_2] TEXT,\n" ' "untitled_2" TEXT,\n'
" [untitled_3] TEXT\n" ' "untitled_3" TEXT\n'
")" ")"
) )
rows = list(db["creatures"].rows) rows = list(db["creatures"].rows)
@ -2182,8 +2182,8 @@ def test_csv_insert_bom(tmpdir):
db = Database(db_path) db = Database(db_path)
tables = db.execute("select name, sql from sqlite_master").fetchall() tables = db.execute("select name, sql from sqlite_master").fetchall()
assert tables == [ assert tables == [
("broken", "CREATE TABLE [broken] (\n [\ufeffname] TEXT,\n [age] TEXT\n)"), ("broken", 'CREATE TABLE "broken" (\n "\ufeffname" TEXT,\n "age" TEXT\n)'),
("fixed", "CREATE TABLE [fixed] (\n [name] TEXT,\n [age] TEXT\n)"), ("fixed", 'CREATE TABLE "fixed" (\n "name" TEXT,\n "age" TEXT\n)'),
] ]
@ -2245,7 +2245,7 @@ def test_integer_overflow_error(tmpdir):
assert result.exit_code == 1 assert result.exit_code == 1
assert result.output == ( assert result.output == (
"Error: Python int too large to convert to SQLite INTEGER\n\n" "Error: Python int too large to convert to SQLite INTEGER\n\n"
"sql = INSERT INTO [items] ([bignumber]) VALUES (?)\n" 'sql = INSERT INTO "items" ("bignumber") VALUES (?)\n'
"parameters = [34223049823094832094802398430298048240]\n" "parameters = [34223049823094832094802398430298048240]\n"
) )

6
tests/test_cli_convert.py
View file

@ -406,9 +406,9 @@ def test_convert_multi_complex_column_types(fresh_db_and_path):
{"id": 4, "is_str": None, "is_float": None, "is_int": None, "is_bytes": None}, {"id": 4, "is_str": None, "is_float": None, "is_int": None, "is_bytes": None},
] ]
assert db["rows"].schema == ( assert db["rows"].schema == (
"CREATE TABLE [rows] (\n" 'CREATE TABLE "rows" (\n'
" [id] INTEGER PRIMARY KEY\n" ' "id" INTEGER PRIMARY KEY\n'
", [is_str] TEXT, [is_float] FLOAT, [is_int] INTEGER, [is_bytes] BLOB)" ', "is_str" TEXT, "is_float" FLOAT, "is_int" INTEGER, "is_bytes" BLOB)'
) )

28
tests/test_cli_insert.py
View file

@ -127,12 +127,12 @@ def test_insert_multiple_with_compound_primary_key(db_path, tmpdir):
assert dogs == list(db.query("select * from dogs order by breed, id")) assert dogs == list(db.query("select * from dogs order by breed, id"))
assert {"breed", "id"} == set(db["dogs"].pks) assert {"breed", "id"} == set(db["dogs"].pks)
assert ( assert (
"CREATE TABLE [dogs] (\n" 'CREATE TABLE "dogs" (\n'
" [breed] TEXT,\n" ' "breed" TEXT,\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [age] INTEGER,\n" ' "age" INTEGER,\n'
" PRIMARY KEY ([id], [breed])\n" ' PRIMARY KEY ("id", "breed")\n'
")" ")"
) == db["dogs"].schema ) == db["dogs"].schema
@ -154,11 +154,11 @@ def test_insert_not_null_default(db_path, tmpdir):
assert result.exit_code == 0 assert result.exit_code == 0
db = Database(db_path) db = Database(db_path)
assert ( assert (
"CREATE TABLE [dogs] (\n" 'CREATE TABLE "dogs" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT NOT NULL,\n" ' "name" TEXT NOT NULL,\n'
" [age] INTEGER NOT NULL DEFAULT '1',\n" " \"age\" INTEGER NOT NULL DEFAULT '1',\n"
" [score] INTEGER DEFAULT '5'\n)" " \"score\" INTEGER DEFAULT '5'\n)"
) == db["dogs"].schema ) == db["dogs"].schema
@ -466,7 +466,7 @@ def test_insert_convert_text(db_path):
) )
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
db = Database(db_path) db = Database(db_path)
rows = list(db.query("select [text] from [text]")) rows = list(db.query('select "text" from "text"'))
assert rows == [{"text": "THIS IS TEXT\nWILL BE UPPER NOW"}] assert rows == [{"text": "THIS IS TEXT\nWILL BE UPPER NOW"}]
@ -486,7 +486,7 @@ def test_insert_convert_text_returning_iterator(db_path):
) )
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
db = Database(db_path) db = Database(db_path)
rows = list(db.query("select [word] from [text]")) rows = list(db.query('select "word" from "text"'))
assert rows == [{"word": "A"}, {"word": "bunch"}, {"word": "of"}, {"word": "words"}] assert rows == [{"word": "A"}, {"word": "bunch"}, {"word": "of"}, {"word": "words"}]
@ -506,7 +506,7 @@ def test_insert_convert_lines(db_path):
) )
assert result.exit_code == 0, result.output assert result.exit_code == 0, result.output
db = Database(db_path) db = Database(db_path)
rows = list(db.query("select [line] from [all]")) rows = list(db.query('select "line" from "all"'))
assert rows == [{"line": "THIS IS TEXT"}, {"line": "WILL BE UPPER NOW"}] assert rows == [{"line": "THIS IS TEXT"}, {"line": "WILL BE UPPER NOW"}]

30
tests/test_cli_memory.py
View file

@ -167,13 +167,13 @@ def test_memory_dump(extra_args):
expected = ( expected = (
"BEGIN TRANSACTION;\n" "BEGIN TRANSACTION;\n"
'CREATE TABLE IF NOT EXISTS "stdin" (\n' 'CREATE TABLE IF NOT EXISTS "stdin" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT\n" ' "name" TEXT\n'
");\n" ");\n"
"INSERT INTO \"stdin\" VALUES(1,'Cleo');\n" "INSERT INTO \"stdin\" VALUES(1,'Cleo');\n"
"INSERT INTO \"stdin\" VALUES(2,'Bants');\n" "INSERT INTO \"stdin\" VALUES(2,'Bants');\n"
"CREATE VIEW t1 AS select * from [stdin];\n" 'CREATE VIEW t1 AS select * from "stdin";\n'
"CREATE VIEW t AS select * from [stdin];\n" 'CREATE VIEW t AS select * from "stdin";\n'
"COMMIT;" "COMMIT;"
) )
# Using sqlite-dump it won't have IF NOT EXISTS # Using sqlite-dump it won't have IF NOT EXISTS
@ -191,11 +191,11 @@ def test_memory_schema(extra_args):
assert result.exit_code == 0 assert result.exit_code == 0
assert result.output.strip() == ( assert result.output.strip() == (
'CREATE TABLE "stdin" (\n' 'CREATE TABLE "stdin" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT\n" ' "name" TEXT\n'
");\n" ");\n"
"CREATE VIEW t1 AS select * from [stdin];\n" 'CREATE VIEW t1 AS select * from "stdin";\n'
"CREATE VIEW t AS select * from [stdin];" 'CREATE VIEW t AS select * from "stdin";'
) )
@ -285,16 +285,16 @@ def test_memory_two_files_with_same_stem(tmpdir):
assert result.exit_code == 0 assert result.exit_code == 0
assert result.output == ( assert result.output == (
'CREATE TABLE "data" (\n' 'CREATE TABLE "data" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT\n" ' "name" TEXT\n'
");\n" ");\n"
"CREATE VIEW t1 AS select * from [data];\n" 'CREATE VIEW t1 AS select * from "data";\n'
"CREATE VIEW t AS select * from [data];\n" 'CREATE VIEW t AS select * from "data";\n'
'CREATE TABLE "data_2" (\n' 'CREATE TABLE "data_2" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT\n" ' "name" TEXT\n'
");\n" ");\n"
"CREATE VIEW t2 AS select * from [data_2];\n" 'CREATE VIEW t2 AS select * from "data_2";\n'
) )

134
tests/test_create.py
View file

@ -50,13 +50,13 @@ def test_create_table(fresh_db):
{"name": "datetime_col", "type": "TEXT"}, {"name": "datetime_col", "type": "TEXT"},
] == [{"name": col.name, "type": col.type} for col in table.columns] ] == [{"name": col.name, "type": col.type} for col in table.columns]
assert ( assert (
"CREATE TABLE [test_table] (\n" 'CREATE TABLE "test_table" (\n'
" [text_col] TEXT,\n" ' "text_col" TEXT,\n'
" [float_col] FLOAT,\n" ' "float_col" FLOAT,\n'
" [int_col] INTEGER,\n" ' "int_col" INTEGER,\n'
" [bool_col] INTEGER,\n" ' "bool_col" INTEGER,\n'
" [bytes_col] BLOB,\n" ' "bytes_col" BLOB,\n'
" [datetime_col] TEXT\n" ' "datetime_col" TEXT\n'
")" ")"
) == table.schema ) == table.schema
@ -66,11 +66,11 @@ def test_create_table_compound_primary_key(fresh_db):
"test_table", {"id1": str, "id2": str, "value": int}, pk=("id1", "id2") "test_table", {"id1": str, "id2": str, "value": int}, pk=("id1", "id2")
) )
assert ( assert (
"CREATE TABLE [test_table] (\n" 'CREATE TABLE "test_table" (\n'
" [id1] TEXT,\n" ' "id1" TEXT,\n'
" [id2] TEXT,\n" ' "id2" TEXT,\n'
" [value] INTEGER,\n" ' "value" INTEGER,\n'
" PRIMARY KEY ([id1], [id2])\n" ' PRIMARY KEY ("id1", "id2")\n'
")" ")"
) == table.schema ) == table.schema
assert ["id1", "id2"] == table.pks assert ["id1", "id2"] == table.pks
@ -80,13 +80,17 @@ def test_create_table_compound_primary_key(fresh_db):
def test_create_table_with_single_primary_key(fresh_db, pk): def test_create_table_with_single_primary_key(fresh_db, pk):
fresh_db["foo"].insert({"id": 1}, pk=pk) fresh_db["foo"].insert({"id": 1}, pk=pk)
assert ( assert (
fresh_db["foo"].schema == "CREATE TABLE [foo] (\n [id] INTEGER PRIMARY KEY\n)" fresh_db["foo"].schema == 'CREATE TABLE "foo" (\n "id" INTEGER PRIMARY KEY\n)'
) )
def test_create_table_with_invalid_column_characters(fresh_db): def test_create_table_with_special_column_characters(fresh_db):
with pytest.raises(AssertionError): # With double-quote escaping, columns with special characters are now valid
fresh_db.create_table("players", {"name[foo]": str}) table = fresh_db.create_table("players", {"name[foo]": str})
assert ["players"] == fresh_db.table_names()
assert [{"name": "name[foo]", "type": "TEXT"}] == [
{"name": col.name, "type": col.type} for col in table.columns
]
def test_create_table_with_defaults(fresh_db): def test_create_table_with_defaults(fresh_db):
@ -100,7 +104,7 @@ def test_create_table_with_defaults(fresh_db):
{"name": col.name, "type": col.type} for col in table.columns {"name": col.name, "type": col.type} for col in table.columns
] ]
assert ( assert (
"CREATE TABLE [players] (\n [name] TEXT DEFAULT 'bob''''bob',\n [score] INTEGER DEFAULT 1\n)" "CREATE TABLE \"players\" (\n \"name\" TEXT DEFAULT 'bob''''bob',\n \"score\" INTEGER DEFAULT 1\n)"
) == table.schema ) == table.schema
@ -123,7 +127,7 @@ def test_create_table_with_not_null(fresh_db):
{"name": col.name, "type": col.type} for col in table.columns {"name": col.name, "type": col.type} for col in table.columns
] ]
assert ( assert (
"CREATE TABLE [players] (\n [name] TEXT NOT NULL,\n [score] INTEGER NOT NULL DEFAULT 3\n)" 'CREATE TABLE "players" (\n "name" TEXT NOT NULL,\n "score" INTEGER NOT NULL DEFAULT 3\n)'
) == table.schema ) == table.schema
@ -145,7 +149,7 @@ def test_create_table_with_not_null(fresh_db):
[{"name": "memoryview", "type": "BLOB"}], [{"name": "memoryview", "type": "BLOB"}],
), ),
({"uuid": uuid.uuid4()}, [{"name": "uuid", "type": "TEXT"}]), ({"uuid": uuid.uuid4()}, [{"name": "uuid", "type": "TEXT"}]),
({"foo[bar]": 1}, [{"name": "foo_bar_", "type": "INTEGER"}]), ({"foo[bar]": 1}, [{"name": "foo[bar]", "type": "INTEGER"}]),
( (
{"timedelta": datetime.timedelta(hours=1)}, {"timedelta": datetime.timedelta(hours=1)},
[{"name": "timedelta", "type": "TEXT"}], [{"name": "timedelta", "type": "TEXT"}],
@ -307,9 +311,9 @@ def test_self_referential_foreign_key(fresh_db):
foreign_keys=(("ref", "test_table", "id"),), foreign_keys=(("ref", "test_table", "id"),),
) )
assert ( assert (
"CREATE TABLE [test_table] (\n" 'CREATE TABLE "test_table" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [ref] INTEGER REFERENCES [test_table]([id])\n" ' "ref" INTEGER REFERENCES "test_table"("id")\n'
")" ")"
) == table.schema ) == table.schema
@ -340,58 +344,58 @@ def test_create_error_if_invalid_self_referential_foreign_keys(fresh_db):
"nickname", "nickname",
str, str,
None, None,
"CREATE TABLE [dogs] (\n [name] TEXT\n, [nickname] TEXT)", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "nickname" TEXT)',
), ),
( (
"dob", "dob",
datetime.date, datetime.date,
None, None,
"CREATE TABLE [dogs] (\n [name] TEXT\n, [dob] TEXT)", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "dob" TEXT)',
), ),
("age", int, None, "CREATE TABLE [dogs] (\n [name] TEXT\n, [age] INTEGER)"), ("age", int, None, 'CREATE TABLE "dogs" (\n "name" TEXT\n, "age" INTEGER)'),
( (
"weight", "weight",
float, float,
None, None,
"CREATE TABLE [dogs] (\n [name] TEXT\n, [weight] FLOAT)", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "weight" FLOAT)',
), ),
("text", "TEXT", None, "CREATE TABLE [dogs] (\n [name] TEXT\n, [text] TEXT)"), ("text", "TEXT", None, 'CREATE TABLE "dogs" (\n "name" TEXT\n, "text" TEXT)'),
( (
"integer", "integer",
"INTEGER", "INTEGER",
None, None,
"CREATE TABLE [dogs] (\n [name] TEXT\n, [integer] INTEGER)", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "integer" INTEGER)',
), ),
( (
"float", "float",
"FLOAT", "FLOAT",
None, None,
"CREATE TABLE [dogs] (\n [name] TEXT\n, [float] FLOAT)", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "float" FLOAT)',
), ),
("blob", "blob", None, "CREATE TABLE [dogs] (\n [name] TEXT\n, [blob] BLOB)"), ("blob", "blob", None, 'CREATE TABLE "dogs" (\n "name" TEXT\n, "blob" BLOB)'),
( (
"default_str", "default_str",
None, None,
None, None,
"CREATE TABLE [dogs] (\n [name] TEXT\n, [default_str] TEXT)", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "default_str" TEXT)',
), ),
( (
"nickname", "nickname",
str, str,
"", "",
"CREATE TABLE [dogs] (\n [name] TEXT\n, [nickname] TEXT NOT NULL DEFAULT '')", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "nickname" TEXT NOT NULL DEFAULT \'\')',
), ),
( (
"nickname", "nickname",
str, str,
"dawg's dawg", "dawg's dawg",
"CREATE TABLE [dogs] (\n [name] TEXT\n, [nickname] TEXT NOT NULL DEFAULT 'dawg''s dawg')", 'CREATE TABLE "dogs" (\n "name" TEXT\n, "nickname" TEXT NOT NULL DEFAULT \'dawg\'\'s dawg\')',
), ),
), ),
) )
def test_add_column(fresh_db, col_name, col_type, not_null_default, expected_schema): def test_add_column(fresh_db, col_name, col_type, not_null_default, expected_schema):
fresh_db.create_table("dogs", {"name": str}) fresh_db.create_table("dogs", {"name": str})
assert fresh_db["dogs"].schema == "CREATE TABLE [dogs] (\n [name] TEXT\n)" assert fresh_db["dogs"].schema == 'CREATE TABLE "dogs" (\n "name" TEXT\n)'
fresh_db["dogs"].add_column(col_name, col_type, not_null_default=not_null_default) fresh_db["dogs"].add_column(col_name, col_type, not_null_default=not_null_default)
assert fresh_db["dogs"].schema == expected_schema assert fresh_db["dogs"].schema == expected_schema
@ -496,8 +500,8 @@ def test_add_column_foreign_key(fresh_db):
fresh_db["dogs"].add_column("breed_id", fk="breeds") fresh_db["dogs"].add_column("breed_id", fk="breeds")
assert fresh_db["dogs"].schema == ( assert fresh_db["dogs"].schema == (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [breed_id] INTEGER REFERENCES [breeds]([rowid])\n" ' "breed_id" INTEGER REFERENCES "breeds"("rowid")\n'
")" ")"
) )
# And again with an explicit primary key column # And again with an explicit primary key column
@ -505,9 +509,9 @@ def test_add_column_foreign_key(fresh_db):
fresh_db["dogs"].add_column("subbreed_id", fk="subbreeds") fresh_db["dogs"].add_column("subbreed_id", fk="subbreeds")
assert fresh_db["dogs"].schema == ( assert fresh_db["dogs"].schema == (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [breed_id] INTEGER REFERENCES [breeds]([rowid]),\n" ' "breed_id" INTEGER REFERENCES "breeds"("rowid"),\n'
" [subbreed_id] TEXT REFERENCES [subbreeds]([primkey])\n" ' "subbreed_id" TEXT REFERENCES "subbreeds"("primkey")\n'
")" ")"
) )
@ -519,8 +523,8 @@ def test_add_foreign_key_guess_table(fresh_db):
fresh_db["dogs"].add_foreign_key("breed_id") fresh_db["dogs"].add_foreign_key("breed_id")
assert fresh_db["dogs"].schema == ( assert fresh_db["dogs"].schema == (
'CREATE TABLE "dogs" (\n' 'CREATE TABLE "dogs" (\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [breed_id] INTEGER REFERENCES [breeds]([id])\n" ' "breed_id" INTEGER REFERENCES "breeds"("id")\n'
")" ")"
) )
@ -591,7 +595,7 @@ def test_add_missing_columns_case_insensitive(fresh_db):
table.add_missing_columns([{"Name": ".", "age": 4}]) table.add_missing_columns([{"Name": ".", "age": 4}])
assert ( assert (
table.schema table.schema
== "CREATE TABLE [foo] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n, [age] INTEGER)" == 'CREATE TABLE "foo" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT\n, "age" INTEGER)'
) )
@ -814,7 +818,7 @@ def test_create_index_desc(fresh_db):
"select sql from sqlite_master where name='idx_dogs_age_name'" "select sql from sqlite_master where name='idx_dogs_age_name'"
).fetchone()[0] ).fetchone()[0]
assert sql == ( assert sql == (
"CREATE INDEX [idx_dogs_age_name]\n" " ON [dogs] ([age] desc, [name])" 'CREATE INDEX "idx_dogs_age_name"\n' ' ON "dogs" ("age" desc, "name")'
) )
@ -1154,19 +1158,19 @@ def test_quote(fresh_db, input, expected):
( (
( (
{"id": int}, {"id": int},
"[id] INTEGER", '"id" INTEGER',
), ),
( (
{"col": dict}, {"col": dict},
"[col] TEXT", '"col" TEXT',
), ),
( (
{"col": tuple}, {"col": tuple},
"[col] TEXT", '"col" TEXT',
), ),
( (
{"col": list}, {"col": list},
"[col] TEXT", '"col" TEXT',
), ),
), ),
) )
@ -1191,12 +1195,12 @@ def test_create(fresh_db):
defaults={"integer": 0}, defaults={"integer": 0},
) )
assert fresh_db["t"].schema == ( assert fresh_db["t"].schema == (
"CREATE TABLE [t] (\n" 'CREATE TABLE "t" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [float] FLOAT NOT NULL,\n" ' "float" FLOAT NOT NULL,\n'
" [text] TEXT,\n" ' "text" TEXT,\n'
" [integer] INTEGER NOT NULL DEFAULT 0,\n" ' "integer" INTEGER NOT NULL DEFAULT 0,\n'
" [bytes] BLOB\n" ' "bytes" BLOB\n'
")" ")"
) )
@ -1232,7 +1236,7 @@ def test_create_replace(fresh_db):
fresh_db["t"].create({"id": int}) fresh_db["t"].create({"id": int})
# This should not # This should not
fresh_db["t"].create({"name": str}, replace=True) fresh_db["t"].create({"name": str}, replace=True)
assert fresh_db["t"].schema == ("CREATE TABLE [t] (\n" " [name] TEXT\n" ")") assert fresh_db["t"].schema == ('CREATE TABLE "t" (\n' ' "name" TEXT\n' ")")
@pytest.mark.parametrize( @pytest.mark.parametrize(
@ -1242,58 +1246,58 @@ def test_create_replace(fresh_db):
( (
{"id": int, "name": str}, {"id": int, "name": str},
{"pk": "id"}, {"pk": "id"},
"CREATE TABLE [demo] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n)", 'CREATE TABLE "demo" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT\n)',
False, False,
), ),
# Drop name column, remove primary key # Drop name column, remove primary key
({"id": int}, {}, 'CREATE TABLE "demo" (\n [id] INTEGER\n)', True), ({"id": int}, {}, 'CREATE TABLE "demo" (\n "id" INTEGER\n)', True),
# Add a new column # Add a new column
( (
{"id": int, "name": str, "age": int}, {"id": int, "name": str, "age": int},
{"pk": "id"}, {"pk": "id"},
'CREATE TABLE "demo" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] INTEGER\n)', 'CREATE TABLE "demo" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" INTEGER\n)',
True, True,
), ),
# Change a column type # Change a column type
( (
{"id": int, "name": bytes}, {"id": int, "name": bytes},
{"pk": "id"}, {"pk": "id"},
'CREATE TABLE "demo" (\n [id] INTEGER PRIMARY KEY,\n [name] BLOB\n)', 'CREATE TABLE "demo" (\n "id" INTEGER PRIMARY KEY,\n "name" BLOB\n)',
True, True,
), ),
# Change the primary key # Change the primary key
( (
{"id": int, "name": str}, {"id": int, "name": str},
{"pk": "name"}, {"pk": "name"},
'CREATE TABLE "demo" (\n [id] INTEGER,\n [name] TEXT PRIMARY KEY\n)', 'CREATE TABLE "demo" (\n "id" INTEGER,\n "name" TEXT PRIMARY KEY\n)',
True, True,
), ),
# Change in column order # Change in column order
( (
{"id": int, "name": str}, {"id": int, "name": str},
{"pk": "id", "column_order": ["name"]}, {"pk": "id", "column_order": ["name"]},
'CREATE TABLE "demo" (\n [name] TEXT,\n [id] INTEGER PRIMARY KEY\n)', 'CREATE TABLE "demo" (\n "name" TEXT,\n "id" INTEGER PRIMARY KEY\n)',
True, True,
), ),
# Same column order is ignored # Same column order is ignored
( (
{"id": int, "name": str}, {"id": int, "name": str},
{"pk": "id", "column_order": ["id", "name"]}, {"pk": "id", "column_order": ["id", "name"]},
"CREATE TABLE [demo] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n)", 'CREATE TABLE "demo" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT\n)',
False, False,
), ),
# Change not null # Change not null
( (
{"id": int, "name": str}, {"id": int, "name": str},
{"pk": "id", "not_null": {"name"}}, {"pk": "id", "not_null": {"name"}},
'CREATE TABLE "demo" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT NOT NULL\n)', 'CREATE TABLE "demo" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT NOT NULL\n)',
True, True,
), ),
# Change default values # Change default values
( (
{"id": int, "name": str}, {"id": int, "name": str},
{"pk": "id", "defaults": {"id": 0, "name": "Bob"}}, {"pk": "id", "defaults": {"id": 0, "name": "Bob"}},
"CREATE TABLE \"demo\" (\n [id] INTEGER PRIMARY KEY DEFAULT 0,\n [name] TEXT DEFAULT 'Bob'\n)", 'CREATE TABLE "demo" (\n "id" INTEGER PRIMARY KEY DEFAULT 0,\n "name" TEXT DEFAULT \'Bob\'\n)',
True, True,
), ),
), ),
@ -1356,11 +1360,11 @@ def test_insert_upsert_strict(fresh_db, method_name, strict):
def test_create_table_strict(fresh_db, strict): def test_create_table_strict(fresh_db, strict):
table = fresh_db.create_table("t", {"id": int, "f": float}, strict=strict) table = fresh_db.create_table("t", {"id": int, "f": float}, strict=strict)
assert table.strict == strict or not fresh_db.supports_strict assert table.strict == strict or not fresh_db.supports_strict
expected_schema = "CREATE TABLE [t] (\n" " [id] INTEGER,\n" " [f] FLOAT\n" ")" expected_schema = 'CREATE TABLE "t" (\n' ' "id" INTEGER,\n' ' "f" FLOAT\n' ")"
if strict and not fresh_db.supports_strict: if strict and not fresh_db.supports_strict:
return return
if strict: if strict:
expected_schema = "CREATE TABLE [t] (\n [id] INTEGER,\n [f] REAL\n) STRICT" expected_schema = 'CREATE TABLE "t" (\n "id" INTEGER,\n "f" REAL\n) STRICT'
assert table.schema == expected_schema assert table.schema == expected_schema

12
tests/test_duplicate.py
View file

@ -6,12 +6,12 @@ import pytest
def test_duplicate(fresh_db): def test_duplicate(fresh_db):
# Create table using native Sqlite statement: # Create table using native Sqlite statement:
fresh_db.execute( fresh_db.execute(
"""CREATE TABLE [table1] ( """CREATE TABLE "table1" (
[text_col] TEXT, "text_col" TEXT,
[real_col] REAL, "real_col" REAL,
[int_col] INTEGER, "int_col" INTEGER,
[bool_col] INTEGER, "bool_col" INTEGER,
[datetime_col] TEXT)""" "datetime_col" TEXT)"""
) )
# Insert one row of mock data: # Insert one row of mock data:
dt = datetime.datetime.now() dt = datetime.datetime.now()

18
tests/test_enable_counts.py
View file

@ -15,25 +15,25 @@ def test_enable_counts_specific_table(fresh_db):
foo.enable_counts() foo.enable_counts()
assert foo.triggers_dict == { assert foo.triggers_dict == {
"foo_counts_insert": ( "foo_counts_insert": (
"CREATE TRIGGER [foo_counts_insert] AFTER INSERT ON [foo]\n" 'CREATE TRIGGER "foo_counts_insert" AFTER INSERT ON "foo"\n'
"BEGIN\n" "BEGIN\n"
" INSERT OR REPLACE INTO [_counts]\n" ' INSERT OR REPLACE INTO "_counts"\n'
" VALUES (\n 'foo',\n" " VALUES (\n 'foo',\n"
" COALESCE(\n" " COALESCE(\n"
" (SELECT count FROM [_counts] WHERE [table] = 'foo'),\n" ' (SELECT count FROM "_counts" WHERE "table" = \'foo\'),\n'
" 0\n" " 0\n"
" ) + 1\n" " ) + 1\n"
" );\n" " );\n"
"END" "END"
), ),
"foo_counts_delete": ( "foo_counts_delete": (
"CREATE TRIGGER [foo_counts_delete] AFTER DELETE ON [foo]\n" 'CREATE TRIGGER "foo_counts_delete" AFTER DELETE ON "foo"\n'
"BEGIN\n" "BEGIN\n"
" INSERT OR REPLACE INTO [_counts]\n" ' INSERT OR REPLACE INTO "_counts"\n'
" VALUES (\n" " VALUES (\n"
" 'foo',\n" " 'foo',\n"
" COALESCE(\n" " COALESCE(\n"
" (SELECT count FROM [_counts] WHERE [table] = 'foo'),\n" ' (SELECT count FROM "_counts" WHERE "table" = \'foo\'),\n'
" 0\n" " 0\n"
" ) - 1\n" " ) - 1\n"
" );\n" " );\n"
@ -132,7 +132,7 @@ def test_uses_counts_after_enable_counts(counts_db_path):
assert db.table("foo").count == 1 assert db.table("foo").count == 1
assert logged == [ assert logged == [
("select name from sqlite_master where type = 'view'", None), ("select name from sqlite_master where type = 'view'", None),
("select count(*) from [foo]", []), ('select count(*) from "foo"', []),
] ]
logged.clear() logged.clear()
assert not db.use_counts_table assert not db.use_counts_table
@ -141,7 +141,7 @@ def test_uses_counts_after_enable_counts(counts_db_path):
assert db.table("foo").count == 1 assert db.table("foo").count == 1
assert logged == [ assert logged == [
( (
"CREATE TABLE IF NOT EXISTS [_counts](\n [table] TEXT PRIMARY KEY,\n count INTEGER DEFAULT 0\n);", 'CREATE TABLE IF NOT EXISTS "_counts"(\n "table" TEXT PRIMARY KEY,\n count INTEGER DEFAULT 0\n);',
None, None,
), ),
("select name from sqlite_master where type = 'table'", None), ("select name from sqlite_master where type = 'table'", None),
@ -157,7 +157,7 @@ def test_uses_counts_after_enable_counts(counts_db_path):
("SELECT quote(:value)", {"value": "baz"}), ("SELECT quote(:value)", {"value": "baz"}),
("select sql from sqlite_master where name = ?", ("_counts",)), ("select sql from sqlite_master where name = ?", ("_counts",)),
("select name from sqlite_master where type = 'view'", None), ("select name from sqlite_master where type = 'view'", None),
("select [table], count from _counts where [table] in (?)", ["foo"]), ('select "table", count from _counts where "table" in (?)', ["foo"]),
] ]

42
tests/test_extract.py
View file

@ -24,16 +24,16 @@ def test_extract_single_column(fresh_db, table, fk_column):
fresh_db["tree"].extract("species", table=table, fk_column=fk_column) fresh_db["tree"].extract("species", table=table, fk_column=fk_column)
assert fresh_db["tree"].schema == ( assert fresh_db["tree"].schema == (
'CREATE TABLE "tree" (\n' 'CREATE TABLE "tree" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [{}] INTEGER REFERENCES [{}]([id]),\n".format(expected_fk, expected_table) ' "{}" INTEGER REFERENCES "{}"("id"),\n'.format(expected_fk, expected_table)
+ " [end] INTEGER\n" + ' "end" INTEGER\n'
+ ")" + ")"
) )
assert fresh_db[expected_table].schema == ( assert fresh_db[expected_table].schema == (
"CREATE TABLE [{}] (\n".format(expected_table) 'CREATE TABLE "{}" (\n'.format(expected_table)
+ " [id] INTEGER PRIMARY KEY,\n" + ' "id" INTEGER PRIMARY KEY,\n'
" [species] TEXT\n" ' "species" TEXT\n'
")" ")"
) )
assert list(fresh_db[expected_table].rows) == [ assert list(fresh_db[expected_table].rows) == [
@ -71,16 +71,16 @@ def test_extract_multiple_columns_with_rename(fresh_db):
) )
assert fresh_db["tree"].schema == ( assert fresh_db["tree"].schema == (
'CREATE TABLE "tree" (\n' 'CREATE TABLE "tree" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [common_name_latin_name_id] INTEGER REFERENCES [common_name_latin_name]([id])\n" ' "common_name_latin_name_id" INTEGER REFERENCES "common_name_latin_name"("id")\n'
")" ")"
) )
assert fresh_db["common_name_latin_name"].schema == ( assert fresh_db["common_name_latin_name"].schema == (
"CREATE TABLE [common_name_latin_name] (\n" 'CREATE TABLE "common_name_latin_name" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [latin_name] TEXT\n" ' "latin_name" TEXT\n'
")" ")"
) )
assert list(fresh_db["common_name_latin_name"].rows) == [ assert list(fresh_db["common_name_latin_name"].rows) == [
@ -122,8 +122,8 @@ def test_extract_rowid_table(fresh_db):
fresh_db["tree"].extract(["common_name", "latin_name"]) fresh_db["tree"].extract(["common_name", "latin_name"])
assert fresh_db["tree"].schema == ( assert fresh_db["tree"].schema == (
'CREATE TABLE "tree" (\n' 'CREATE TABLE "tree" (\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [common_name_latin_name_id] INTEGER REFERENCES [common_name_latin_name]([id])\n" ' "common_name_latin_name_id" INTEGER REFERENCES "common_name_latin_name"("id")\n'
")" ")"
) )
assert ( assert (
@ -152,15 +152,15 @@ def test_reuse_lookup_table(fresh_db):
fresh_db["individuals"].extract("species", rename={"species": "name"}) fresh_db["individuals"].extract("species", rename={"species": "name"})
assert fresh_db["sightings"].schema == ( assert fresh_db["sightings"].schema == (
'CREATE TABLE "sightings" (\n' 'CREATE TABLE "sightings" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [species_id] INTEGER REFERENCES [species]([id])\n" ' "species_id" INTEGER REFERENCES "species"("id")\n'
")" ")"
) )
assert fresh_db["individuals"].schema == ( assert fresh_db["individuals"].schema == (
'CREATE TABLE "individuals" (\n' 'CREATE TABLE "individuals" (\n'
" [id] INTEGER PRIMARY KEY,\n" ' "id" INTEGER PRIMARY KEY,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [species_id] INTEGER REFERENCES [species]([id])\n" ' "species_id" INTEGER REFERENCES "species"("id")\n'
")" ")"
) )
assert list(fresh_db["species"].rows) == [ assert list(fresh_db["species"].rows) == [

4
tests/test_extracts.py
View file

@ -30,13 +30,13 @@ def test_extracts(fresh_db, kwargs, expected_table, use_table_factory):
# Should now have two tables: Trees and Species # Should now have two tables: Trees and Species
assert {expected_table, "Trees"} == set(fresh_db.table_names()) assert {expected_table, "Trees"} == set(fresh_db.table_names())
assert ( assert (
"CREATE TABLE [{}] (\n [id] INTEGER PRIMARY KEY,\n [value] TEXT\n)".format( 'CREATE TABLE "{}" (\n "id" INTEGER PRIMARY KEY,\n "value" TEXT\n)'.format(
expected_table expected_table
) )
== fresh_db[expected_table].schema == fresh_db[expected_table].schema
) )
assert ( assert (
"CREATE TABLE [Trees] (\n [id] INTEGER,\n [species_id] INTEGER REFERENCES [{}]([id])\n)".format( 'CREATE TABLE "Trees" (\n "id" INTEGER,\n "species_id" INTEGER REFERENCES "{}"("id")\n)'.format(
expected_table expected_table
) )
== fresh_db["Trees"].schema == fresh_db["Trees"].schema

132
tests/test_fts.py
View file

@ -435,40 +435,40 @@ def test_enable_fts_error_message_on_views():
{}, {},
"FTS5", "FTS5",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" *\n" " *\n"
" from [books]\n" ' from "books"\n'
")\n" ")\n"
"select\n" "select\n"
" [original].*\n" ' "original".*\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" [books_fts].rank" ' "books_fts".rank'
), ),
), ),
( (
{"columns": ["title"], "order_by": "rowid", "limit": 10}, {"columns": ["title"], "order_by": "rowid", "limit": 10},
"FTS5", "FTS5",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" [title]\n" ' "title"\n'
" from [books]\n" ' from "books"\n'
")\n" ")\n"
"select\n" "select\n"
" [original].[title]\n" ' "original"."title"\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" rowid\n" " rowid\n"
"limit 10" "limit 10"
@ -478,64 +478,64 @@ def test_enable_fts_error_message_on_views():
{"where": "author = :author"}, {"where": "author = :author"},
"FTS5", "FTS5",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" *\n" " *\n"
" from [books]\n" ' from "books"\n'
" where author = :author\n" " where author = :author\n"
")\n" ")\n"
"select\n" "select\n"
" [original].*\n" ' "original".*\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" [books_fts].rank" ' "books_fts".rank'
), ),
), ),
( (
{"columns": ["title"]}, {"columns": ["title"]},
"FTS4", "FTS4",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" [title]\n" ' "title"\n'
" from [books]\n" ' from "books"\n'
")\n" ")\n"
"select\n" "select\n"
" [original].[title]\n" ' "original"."title"\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" rank_bm25(matchinfo([books_fts], 'pcnalx'))" " rank_bm25(matchinfo(\"books_fts\", 'pcnalx'))"
), ),
), ),
( (
{"offset": 1, "limit": 1}, {"offset": 1, "limit": 1},
"FTS4", "FTS4",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" *\n" " *\n"
" from [books]\n" ' from "books"\n'
")\n" ")\n"
"select\n" "select\n"
" [original].*\n" ' "original".*\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" rank_bm25(matchinfo([books_fts], 'pcnalx'))\n" " rank_bm25(matchinfo(\"books_fts\", 'pcnalx'))\n"
"limit 1 offset 1" "limit 1 offset 1"
), ),
), ),
@ -543,21 +543,21 @@ def test_enable_fts_error_message_on_views():
{"limit": 2}, {"limit": 2},
"FTS4", "FTS4",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" *\n" " *\n"
" from [books]\n" ' from "books"\n'
")\n" ")\n"
"select\n" "select\n"
" [original].*\n" ' "original".*\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" rank_bm25(matchinfo([books_fts], 'pcnalx'))\n" " rank_bm25(matchinfo(\"books_fts\", 'pcnalx'))\n"
"limit 2" "limit 2"
), ),
), ),
@ -565,66 +565,66 @@ def test_enable_fts_error_message_on_views():
{"where": "author = :author"}, {"where": "author = :author"},
"FTS4", "FTS4",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" *\n" " *\n"
" from [books]\n" ' from "books"\n'
" where author = :author\n" " where author = :author\n"
")\n" ")\n"
"select\n" "select\n"
" [original].*\n" ' "original".*\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" rank_bm25(matchinfo([books_fts], 'pcnalx'))" " rank_bm25(matchinfo(\"books_fts\", 'pcnalx'))"
), ),
), ),
( (
{"include_rank": True}, {"include_rank": True},
"FTS5", "FTS5",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" *\n" " *\n"
" from [books]\n" ' from "books"\n'
")\n" ")\n"
"select\n" "select\n"
" [original].*,\n" ' "original".*,\n'
" [books_fts].rank rank\n" ' "books_fts".rank rank\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" [books_fts].rank" ' "books_fts".rank'
), ),
), ),
( (
{"include_rank": True}, {"include_rank": True},
"FTS4", "FTS4",
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" *\n" " *\n"
" from [books]\n" ' from "books"\n'
")\n" ")\n"
"select\n" "select\n"
" [original].*,\n" ' "original".*,\n'
" rank_bm25(matchinfo([books_fts], 'pcnalx')) rank\n" " rank_bm25(matchinfo(\"books_fts\", 'pcnalx')) rank\n"
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [books_fts] on [original].rowid = [books_fts].rowid\n" ' join "books_fts" on "original".rowid = "books_fts".rowid\n'
"where\n" "where\n"
" [books_fts] match :query\n" ' "books_fts" match :query\n'
"order by\n" "order by\n"
" rank_bm25(matchinfo([books_fts], 'pcnalx'))" " rank_bm25(matchinfo(\"books_fts\", 'pcnalx'))"
), ),
), ),
], ],

14
tests/test_introspect.py
View file

@ -200,19 +200,19 @@ def test_triggers_and_triggers_dict(fresh_db):
} }
expected_triggers = { expected_triggers = {
"authors_ai": ( "authors_ai": (
"CREATE TRIGGER [authors_ai] AFTER INSERT ON [authors] BEGIN\n" 'CREATE TRIGGER "authors_ai" AFTER INSERT ON "authors" BEGIN\n'
" INSERT INTO [authors_fts] (rowid, [name], [famous_works]) VALUES (new.rowid, new.[name], new.[famous_works]);\n" ' INSERT INTO "authors_fts" (rowid, "name", "famous_works") VALUES (new.rowid, new."name", new."famous_works");\n'
"END" "END"
), ),
"authors_ad": ( "authors_ad": (
"CREATE TRIGGER [authors_ad] AFTER DELETE ON [authors] BEGIN\n" 'CREATE TRIGGER "authors_ad" AFTER DELETE ON "authors" BEGIN\n'
" INSERT INTO [authors_fts] ([authors_fts], rowid, [name], [famous_works]) VALUES('delete', old.rowid, old.[name], old.[famous_works]);\n" ' INSERT INTO "authors_fts" ("authors_fts", rowid, "name", "famous_works") VALUES(\'delete\', old.rowid, old."name", old."famous_works");\n'
"END" "END"
), ),
"authors_au": ( "authors_au": (
"CREATE TRIGGER [authors_au] AFTER UPDATE ON [authors] BEGIN\n" 'CREATE TRIGGER "authors_au" AFTER UPDATE ON "authors" BEGIN\n'
" INSERT INTO [authors_fts] ([authors_fts], rowid, [name], [famous_works]) VALUES('delete', old.rowid, old.[name], old.[famous_works]);\n" ' INSERT INTO "authors_fts" ("authors_fts", rowid, "name", "famous_works") VALUES(\'delete\', old.rowid, old."name", old."famous_works");\n'
" INSERT INTO [authors_fts] (rowid, [name], [famous_works]) VALUES (new.rowid, new.[name], new.[famous_works]);\nEND" ' INSERT INTO "authors_fts" (rowid, "name", "famous_works") VALUES (new.rowid, new."name", new."famous_works");\nEND'
), ),
} }
assert authors.triggers_dict == expected_triggers assert authors.triggers_dict == expected_triggers

24
tests/test_lookup.py
View file

@ -114,18 +114,18 @@ def test_lookup_with_extra_insert_parameters(fresh_db):
columns={"make_this_integer": int}, columns={"make_this_integer": int},
) )
assert species.schema == ( assert species.schema == (
"CREATE TABLE [species] (\n" 'CREATE TABLE "species" (\n'
" [renamed_id] INTEGER PRIMARY KEY,\n" ' "renamed_id" INTEGER PRIMARY KEY,\n'
" [this_at_front] INTEGER,\n" ' "this_at_front" INTEGER,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [type] TEXT,\n" ' "type" TEXT,\n'
" [first_seen] TEXT,\n" ' "first_seen" TEXT,\n'
" [make_not_null] INTEGER NOT NULL,\n" ' "make_not_null" INTEGER NOT NULL,\n'
" [fk_to_other] INTEGER REFERENCES [other_table]([id]),\n" ' "fk_to_other" INTEGER REFERENCES "other_table"("id"),\n'
" [default_is_dog] TEXT DEFAULT 'dog',\n" " \"default_is_dog\" TEXT DEFAULT 'dog',\n"
" [extract_this] INTEGER REFERENCES [extract_this]([id]),\n" ' "extract_this" INTEGER REFERENCES "extract_this"("id"),\n'
" [convert_to_upper] TEXT,\n" ' "convert_to_upper" TEXT,\n'
" [make_this_integer] INTEGER\n" ' "make_this_integer" INTEGER\n'
")" ")"
) )
assert species.get(id) == { assert species.get(id) == {

24
tests/test_tracer.py
View file

@ -18,15 +18,15 @@ def test_tracer():
("select name from sqlite_master where type = 'view'", None), ("select name from sqlite_master where type = 'view'", None),
("select name from sqlite_master where type = 'table'", None), ("select name from sqlite_master where type = 'table'", None),
("select name from sqlite_master where type = 'view'", None), ("select name from sqlite_master where type = 'view'", None),
("CREATE TABLE [dogs] (\n [name] TEXT\n);\n ", None), ('CREATE TABLE "dogs" (\n "name" TEXT\n);\n ', None),
("select name from sqlite_master where type = 'view'", None), ("select name from sqlite_master where type = 'view'", None),
("INSERT INTO [dogs] ([name]) VALUES (?)", ["Cleopaws"]), ('INSERT INTO "dogs" ("name") VALUES (?)', ["Cleopaws"]),
( (
"CREATE VIRTUAL TABLE [dogs_fts] USING FTS5 (\n [name],\n content=[dogs]\n)", 'CREATE VIRTUAL TABLE "dogs_fts" USING FTS5 (\n "name",\n content="dogs"\n)',
None, None,
), ),
( (
"INSERT INTO [dogs_fts] (rowid, [name])\n SELECT rowid, [name] FROM [dogs];", 'INSERT INTO "dogs_fts" (rowid, "name")\n SELECT rowid, "name" FROM "dogs";',
None, None,
), ),
] ]
@ -64,7 +64,7 @@ def test_with_tracer():
" )\n" " )\n"
" )", " )",
{ {
"like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%", "like": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
"like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%', "like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
"table": "dogs", "table": "dogs",
}, },
@ -73,21 +73,21 @@ def test_with_tracer():
("select name from sqlite_master where type = 'view'", None), ("select name from sqlite_master where type = 'view'", None),
("select sql from sqlite_master where name = ?", ("dogs_fts",)), ("select sql from sqlite_master where name = ?", ("dogs_fts",)),
( (
"with original as (\n" 'with "original" as (\n'
" select\n" " select\n"
" rowid,\n" " rowid,\n"
" *\n" " *\n"
" from [dogs]\n" ' from "dogs"\n'
")\n" ")\n"
"select\n" "select\n"
" [original].*\n" ' "original".*\n'
"from\n" "from\n"
" [original]\n" ' "original"\n'
" join [dogs_fts] on [original].rowid = [dogs_fts].rowid\n" ' join "dogs_fts" on "original".rowid = "dogs_fts".rowid\n'
"where\n" "where\n"
" [dogs_fts] match :query\n" ' "dogs_fts" match :query\n'
"order by\n" "order by\n"
" [dogs_fts].rank", ' "dogs_fts".rank',
{"query": "Cleopaws"}, {"query": "Cleopaws"},
), ),
] ]

144
tests/test_transform.py
View file

@ -10,90 +10,90 @@ import pytest
( (
{}, {},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] TEXT\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" TEXT\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Change column type # Change column type
( (
{"types": {"age": int}}, {"types": {"age": int}},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] INTEGER\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" INTEGER\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Rename a column # Rename a column
( (
{"rename": {"age": "dog_age"}}, {"rename": {"age": "dog_age"}},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [dog_age] TEXT\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "dog_age" TEXT\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [dog_age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "dog_age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Drop a column # Drop a column
( (
{"drop": ["age"]}, {"drop": ["age"]},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name])\n SELECT [rowid], [id], [name] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name")\n SELECT "rowid", "id", "name" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Convert type AND rename column # Convert type AND rename column
( (
{"types": {"age": int}, "rename": {"age": "dog_age"}}, {"types": {"age": int}, "rename": {"age": "dog_age"}},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [dog_age] INTEGER\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "dog_age" INTEGER\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [dog_age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "dog_age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Change primary key # Change primary key
( (
{"pk": "age"}, {"pk": "age"},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT PRIMARY KEY\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER,\n "name" TEXT,\n "age" TEXT PRIMARY KEY\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Change primary key to a compound pk # Change primary key to a compound pk
( (
{"pk": ("age", "name")}, {"pk": ("age", "name")},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT,\n PRIMARY KEY ([age], [name])\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER,\n "name" TEXT,\n "age" TEXT,\n PRIMARY KEY ("age", "name")\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Remove primary key, creating a rowid table # Remove primary key, creating a rowid table
( (
{"pk": None}, {"pk": None},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER,\n "name" TEXT,\n "age" TEXT\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Keeping the table # Keeping the table
( (
{"drop": ["age"], "keep_table": "kept_table"}, {"drop": ["age"], "keep_table": "kept_table"},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name])\n SELECT [rowid], [id], [name] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name")\n SELECT "rowid", "id", "name" FROM "dogs";',
"ALTER TABLE [dogs] RENAME TO [kept_table];", 'ALTER TABLE "dogs" RENAME TO "kept_table";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
], ],
@ -133,40 +133,40 @@ def test_transform_sql_table_with_primary_key(
( (
{}, {},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER,\n "name" TEXT,\n "age" TEXT\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Change column type # Change column type
( (
{"types": {"age": int}}, {"types": {"age": int}},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [age] INTEGER\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER,\n "name" TEXT,\n "age" INTEGER\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Rename a column # Rename a column
( (
{"rename": {"age": "dog_age"}}, {"rename": {"age": "dog_age"}},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER,\n [name] TEXT,\n [dog_age] TEXT\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER,\n "name" TEXT,\n "dog_age" TEXT\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [dog_age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "dog_age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
# Make ID a primary key # Make ID a primary key
( (
{"pk": "id"}, {"pk": "id"},
[ [
"CREATE TABLE [dogs_new_suffix] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] TEXT\n);", 'CREATE TABLE "dogs_new_suffix" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" TEXT\n);',
"INSERT INTO [dogs_new_suffix] ([rowid], [id], [name], [age])\n SELECT [rowid], [id], [name], [age] FROM [dogs];", 'INSERT INTO "dogs_new_suffix" ("rowid", "id", "name", "age")\n SELECT "rowid", "id", "name", "age" FROM "dogs";',
"DROP TABLE [dogs];", 'DROP TABLE "dogs";',
"ALTER TABLE [dogs_new_suffix] RENAME TO [dogs];", 'ALTER TABLE "dogs_new_suffix" RENAME TO "dogs";',
], ],
), ),
], ],
@ -204,13 +204,13 @@ def test_transform_sql_with_no_primary_key_to_primary_key_of_id(fresh_db):
dogs.insert({"id": 1, "name": "Cleo", "age": "5"}) dogs.insert({"id": 1, "name": "Cleo", "age": "5"})
assert ( assert (
dogs.schema dogs.schema
== "CREATE TABLE [dogs] (\n [id] INTEGER,\n [name] TEXT,\n [age] TEXT\n)" == 'CREATE TABLE "dogs" (\n "id" INTEGER,\n "name" TEXT,\n "age" TEXT\n)'
) )
dogs.transform(pk="id") dogs.transform(pk="id")
# Slight oddity: [dogs] becomes "dogs" during the rename: # Slight oddity: [dogs] becomes "dogs" during the rename:
assert ( assert (
dogs.schema dogs.schema
== 'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] TEXT\n)' == 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" TEXT\n)'
) )
@ -220,7 +220,7 @@ def test_transform_rename_pk(fresh_db):
dogs.transform(rename={"id": "pk"}) dogs.transform(rename={"id": "pk"})
assert ( assert (
dogs.schema dogs.schema
== 'CREATE TABLE "dogs" (\n [pk] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] TEXT\n)' == 'CREATE TABLE "dogs" (\n "pk" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" TEXT\n)'
) )
@ -230,7 +230,7 @@ def test_transform_not_null(fresh_db):
dogs.transform(not_null={"name"}) dogs.transform(not_null={"name"})
assert ( assert (
dogs.schema dogs.schema
== 'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT NOT NULL,\n [age] TEXT\n)' == 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT NOT NULL,\n "age" TEXT\n)'
) )
@ -240,7 +240,7 @@ def test_transform_remove_a_not_null(fresh_db):
dogs.transform(not_null={"name": True, "age": False}) dogs.transform(not_null={"name": True, "age": False})
assert ( assert (
dogs.schema dogs.schema
== 'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT NOT NULL,\n [age] TEXT\n)' == 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT NOT NULL,\n "age" TEXT\n)'
) )
@ -251,7 +251,7 @@ def test_transform_add_not_null_with_rename(fresh_db, not_null):
dogs.transform(not_null=not_null, rename={"age": "dog_age"}) dogs.transform(not_null=not_null, rename={"age": "dog_age"})
assert ( assert (
dogs.schema dogs.schema
== 'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [dog_age] TEXT NOT NULL\n)' == 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "dog_age" TEXT NOT NULL\n)'
) )
@ -261,7 +261,7 @@ def test_transform_defaults(fresh_db):
dogs.transform(defaults={"age": 1}) dogs.transform(defaults={"age": 1})
assert ( assert (
dogs.schema dogs.schema
== 'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] INTEGER DEFAULT 1\n)' == 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" INTEGER DEFAULT 1\n)'
) )
@ -271,7 +271,7 @@ def test_transform_defaults_and_rename_column(fresh_db):
dogs.transform(rename={"age": "dog_age"}, defaults={"age": 1}) dogs.transform(rename={"age": "dog_age"}, defaults={"age": 1})
assert ( assert (
dogs.schema dogs.schema
== 'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [dog_age] INTEGER DEFAULT 1\n)' == 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "dog_age" INTEGER DEFAULT 1\n)'
) )
@ -281,7 +281,7 @@ def test_remove_defaults(fresh_db):
dogs.transform(defaults={"age": None}) dogs.transform(defaults={"age": None})
assert ( assert (
dogs.schema dogs.schema
== 'CREATE TABLE "dogs" (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT,\n [age] INTEGER\n)' == 'CREATE TABLE "dogs" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT,\n "age" INTEGER\n)'
) )
@ -391,7 +391,7 @@ def test_transform_verify_foreign_keys(fresh_db):
# This should have rolled us back # This should have rolled us back
assert ( assert (
fresh_db["authors"].schema fresh_db["authors"].schema
== "CREATE TABLE [authors] (\n [id] INTEGER PRIMARY KEY,\n [name] TEXT\n)" == 'CREATE TABLE "authors" (\n "id" INTEGER PRIMARY KEY,\n "name" TEXT\n)'
) )
assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0] assert fresh_db.conn.execute("PRAGMA foreign_keys").fetchone()[0]
@ -420,11 +420,11 @@ def test_transform_add_foreign_keys_from_scratch(fresh_db):
] ]
assert fresh_db["places"].schema == ( assert fresh_db["places"].schema == (
'CREATE TABLE "places" (\n' 'CREATE TABLE "places" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [country] INTEGER REFERENCES [country]([id]),\n" ' "country" INTEGER REFERENCES "country"("id"),\n'
" [continent] INTEGER REFERENCES [continent]([id]),\n" ' "continent" INTEGER REFERENCES "continent"("id"),\n'
" [city] INTEGER REFERENCES [city]([id])\n" ' "city" INTEGER REFERENCES "city"("id")\n'
")" ")"
) )
@ -491,11 +491,11 @@ def test_transform_replace_foreign_keys(fresh_db, foreign_keys):
fresh_db["places"].transform(foreign_keys=foreign_keys) fresh_db["places"].transform(foreign_keys=foreign_keys)
assert fresh_db["places"].schema == ( assert fresh_db["places"].schema == (
'CREATE TABLE "places" (\n' 'CREATE TABLE "places" (\n'
" [id] INTEGER,\n" ' "id" INTEGER,\n'
" [name] TEXT,\n" ' "name" TEXT,\n'
" [country] INTEGER REFERENCES [country]([id]),\n" ' "country" INTEGER REFERENCES "country"("id"),\n'
" [continent] INTEGER REFERENCES [continent]([id]),\n" ' "continent" INTEGER REFERENCES "continent"("id"),\n'
" [city] INTEGER\n" ' "city" INTEGER\n'
")" ")"
) )

7
tests/test_update.py
View file

@ -70,11 +70,12 @@ def test_update_alter(fresh_db):
] == list(table.rows) ] == list(table.rows)
def test_update_alter_with_invalid_column_characters(fresh_db): def test_update_alter_with_special_column_characters(fresh_db):
# With double-quote escaping, columns with special characters are now valid
table = fresh_db["table"] table = fresh_db["table"]
rowid = table.insert({"foo": "bar"}).last_pk rowid = table.insert({"foo": "bar"}).last_pk
with pytest.raises(AssertionError): table.update(rowid, {"new_col[abc]": 1.2}, alter=True)
table.update(rowid, {"new_col[abc]": 1.2}, alter=True) assert list(table.rows) == [{"foo": "bar", "new_col[abc]": 1.2}]
def test_update_with_no_values_sets_last_pk(fresh_db): def test_update_with_no_values_sets_last_pk(fresh_db):