remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-09-28 21:04:13 +02:00
Code Issues Projects Releases Packages Wiki Activity

rework to store schema name explicitly

Browse source
This commit is contained in:
Peter Gaultney 2024-01-09 09:27:59 -06:00
commit 7b2488c1a2
No known key found for this signature in database
1 changed files with 102 additions and 99 deletions
Download patch file Download diff file Expand all files Collapse all files

199
sqlite_utils/db.py
View file

@ -532,7 +532,9 @@ class Database:
self._tracer(sql, None) self._tracer(sql, None)
return self.conn.executescript(sql) return self.conn.executescript(sql)
def table(self, table_name: str, **kwargs) -> Union["Table", "View"]: def table(
self, table_name: str, schema: str = "", **kwargs
) -> Union["Table", "View"]:
""" """
Return a table object, optionally configured with default options. Return a table object, optionally configured with default options.
@ -541,10 +543,10 @@ class Database:
:param table_name: Name of the table :param table_name: Name of the table
""" """
if table_name in self.view_names(): if table_name in self.view_names():
return View(self, table_name, **kwargs) return View(self, table_name, schema_name=schema, **kwargs)
else: else:
kwargs.setdefault("strict", self.strict) kwargs.setdefault("strict", self.strict)
return Table(self, table_name, **kwargs) return Table(self, table_name, schema_name=schema, **kwargs)
def quote(self, value: str) -> str: def quote(self, value: str) -> str:
""" """
@ -601,44 +603,46 @@ class Database:
return self.quote(value) return self.quote(value)
def database_names(self) -> List[str]: def schema_names(self) -> List[str]:
"List of string database names available in this connection." """List of string database schemas available in this connection.
Unless other databases are ATTACHed using `attach`, this will only return
`['main']` or `['main', 'temp']`. See https://www.sqlite.org/lang_attach.html
"""
return [r[1] for r in self.execute("PRAGMA database_list").fetchall()] return [r[1] for r in self.execute("PRAGMA database_list").fetchall()]
def table_names(self, fts4: bool = False, fts5: bool = False) -> List[str]: def _from_schema(self, schema: str) -> str:
""" if schema and schema != "main":
List of string table names in this database. return f"{schema}.sqlite_master"
return "sqlite_master" # keep SQL simple for the standard case.
def table_names(
self, fts4: bool = False, fts5: bool = False, schema: str = ""
) -> List[str]:
"""List of string table names in the specified database schema.
:param fts4: Only return tables that are part of FTS4 indexes :param fts4: Only return tables that are part of FTS4 indexes
:param fts5: Only return tables that are part of FTS5 indexes :param fts5: Only return tables that are part of FTS5 indexes
:param schema: By default, the `main` schema is queried, but a different,
attached database can be queried instead.
""" """
where = ["type = 'table'"] where = ["type = 'table'"]
if fts4: if fts4:
where.append("sql like '%USING FTS4%'") where.append("sql like '%USING FTS4%'")
if fts5: if fts5:
where.append("sql like '%USING FTS5%'") where.append("sql like '%USING FTS5%'")
sql = "select name from sqlite_master where {}".format(" AND ".join(where))
def _exec_in_db(db_name: str, sql: str) -> List[str]: sql = "select name from {} where {}".format(
if db_name == "main": self._from_schema(schema), " AND ".join(where)
db_name = ""
if db_name:
sql = sql.replace("sqlite_master", f"{db_name}.sqlite_master")
table_names = [r[0] for r in self.execute(sql).fetchall()]
if db_name:
return [f"{db_name}.{tbl_name}" for tbl_name in table_names]
return table_names
return list(
itertools.chain(*[_exec_in_db(db_name, sql) for db_name in self.database_names()])
) )
return [r[0] for r in self.execute(sql).fetchall()]
def view_names(self) -> List[str]: def view_names(self, schema: str = "") -> List[str]:
"List of string view names in this database." "List of string view names in this database."
return [ return [
r[0] r[0]
for r in self.execute( for r in self.execute(
"select name from sqlite_master where type = 'view'" f"select name from {self._from_schema(schema)} where type = 'view'"
).fetchall() ).fetchall()
] ]
@ -1288,26 +1292,10 @@ class Database:
return result and bool(result[0]) return result and bool(result[0])
def _split_names(fullname: str) -> Tuple[str, str]: def _fullname(schema_name: str, table_name: str) -> str:
if '.' not in fullname: if schema_name:
return '', fullname return f"{schema_name}.[{table_name}]"
return fullname.split('.') return "[" + table_name + "]"
def dbname(fullname: str) -> str:
return _split_names(fullname)[0]
def tablename(fullname: str) -> str:
return _split_names(fullname)[1]
def escaped_name(fullname: str) -> str:
"""This is how SQLite expects a database name joined to a table name to use the square-bracket escapes."""
db, tbl = _split_names(fullname)
if not db:
return f'[{tbl}]'
return f'{db}.[{tbl}]'
class Queryable: class Queryable:
@ -1315,9 +1303,14 @@ class Queryable:
"Does this table or view exist yet?" "Does this table or view exist yet?"
return False return False
def __init__(self, db, name: str): def __init__(self, db, name: str, schema_name: str = ""):
self.db = db self.db = db
self.name = name self.name = name
self.schema_name = schema_name # default is empty string, a.k.a. 'main'
@property
def _fullname(self) -> str:
return _fullname(self.schema_name, self.name)
def count_where( def count_where(
self, self,
@ -1331,7 +1324,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(escaped_name(self.name)) sql = "select count(*) from {}".format(self._fullname)
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]
@ -1374,7 +1367,7 @@ class Queryable:
""" """
if not self.exists(): if not self.exists():
return return
sql = "select {} from {}".format(select, escaped_name(self.name)) sql = "select {} from {}".format(select, self._fullname)
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:
@ -1428,13 +1421,12 @@ class Queryable:
@property @property
def is_attached(self) -> bool: def is_attached(self) -> bool:
return dbname(self.name) not in {'', 'main'} return self.schema_name not in {"", "main"}
@property @property
def _pragma_name(self) -> Tuple[str, str]: def _pragma_name(self) -> Tuple[str, str]:
if "." in self.name: if self.schema_name:
db, name = self.name.split(".") return self.schema_name + ".", self.name
return db + ".", name
return "", self.name return "", self.name
@property @property
@ -1442,7 +1434,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._pragma_name)).fetchall() rows = self.db.execute(
"PRAGMA {}table_info([{}])".format(*self._pragma_name)
).fetchall()
return [Column(*row) for row in rows] return [Column(*row) for row in rows]
@property @property
@ -1454,9 +1448,9 @@ class Queryable:
def schema(self) -> str: def schema(self) -> str:
"SQL schema for this table or view." "SQL schema for this table or view."
db, name = self._pragma_name db, name = self._pragma_name
return self.db.execute(f"select sql from {db}sqlite_master where name = ?", (name,)).fetchone()[ return self.db.execute(
0 f"select sql from {db}sqlite_master where name = ?", (name,)
] ).fetchone()[0]
class Table(Queryable): class Table(Queryable):
@ -1509,8 +1503,9 @@ class Table(Queryable):
conversions: Optional[dict] = None, conversions: Optional[dict] = None,
columns: Optional[Dict[str, Any]] = None, columns: Optional[Dict[str, Any]] = None,
strict: bool = False, strict: bool = False,
schema_name: str = "",
): ):
super().__init__(db, name) super().__init__(db, name, schema_name=schema_name)
self._defaults = dict( self._defaults = dict(
pk=pk, pk=pk,
foreign_keys=foreign_keys, foreign_keys=foreign_keys,
@ -1547,7 +1542,7 @@ class Table(Queryable):
return self.count_where() return self.count_where()
def exists(self) -> bool: def exists(self) -> bool:
return self.name in self.db.table_names() return self.name in self.db.table_names(schema=self.schema_name)
@property @property
def pks(self) -> List[str]: def pks(self) -> List[str]:
@ -1764,8 +1759,8 @@ class Table(Queryable):
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 {new_table} AS SELECT * FROM {table};".format(
new_table=escaped_name(new_name), new_table=new_name,
table=escaped_name(self.name), table=self._fullname,
) )
self.db.execute(sql) self.db.execute(sql)
return self.db[new_name] return self.db[new_name]
@ -1820,9 +1815,9 @@ class Table(Queryable):
keep_table=keep_table, keep_table=keep_table,
) )
db, _ = self._pragma_name db, _ = self._pragma_name
pragma_foreign_keys_was_on = self.db.execute(f"PRAGMA {db}foreign_keys").fetchone()[ pragma_foreign_keys_was_on = self.db.execute(
0 f"PRAGMA {db}foreign_keys"
] ).fetchone()[0]
try: try:
if pragma_foreign_keys_was_on: if pragma_foreign_keys_was_on:
self.db.execute(f"PRAGMA {db}foreign_keys=0;") self.db.execute(f"PRAGMA {db}foreign_keys=0;")
@ -1999,9 +1994,12 @@ 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")
old_fullname = _fullname(self.schema_name, self.name)
new_fullname = _fullname(self.schema_name, new_table_name)
copy_sql = "INSERT INTO {new_table} ({new_cols})\n SELECT {old_cols} FROM {old_table};".format( copy_sql = "INSERT INTO {new_table} ({new_cols})\n SELECT {old_cols} FROM {old_table};".format(
new_table=escaped_name(new_table_name), new_table=new_fullname,
old_table=escaped_name(self.name), old_table=old_fullname,
old_cols=", ".join("[{}]".format(col) for col in old_cols), old_cols=", ".join("[{}]".format(col) for col in old_cols),
new_cols=", ".join("[{}]".format(col) for col in new_cols), new_cols=", ".join("[{}]".format(col) for col in new_cols),
) )
@ -2009,14 +2007,14 @@ class Table(Queryable):
# 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(escaped_name(self.name), escaped_name(keep_table)) "ALTER TABLE {} RENAME TO {};".format(
old_fullname, _fullname(self.schema_name, keep_table)
)
) )
else: else:
sqls.append("DROP TABLE {};".format(escaped_name(self.name))) sqls.append("DROP TABLE {};".format(old_fullname))
# Rename the new one # Rename the new one
sqls.append( sqls.append("ALTER TABLE {} RENAME TO {};".format(new_fullname, old_fullname))
"ALTER TABLE {} RENAME TO {};".format(escaped_name(new_table_name), escaped_name(self.name))
)
return sqls return sqls
def extract( def extract(
@ -2079,10 +2077,10 @@ class Table(Queryable):
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_table} ({lookup_columns}) SELECT DISTINCT {table_cols} FROM {table}".format(
lookup_table=escaped_name(table), lookup_table=_fullname(self.schema_name, table),
lookup_columns=", ".join("[{}]".format(c) for c in lookup_columns), lookup_columns=", ".join("[{}]".format(c) for c in lookup_columns),
table_cols=", ".join("[{}]".format(c) for c in columns), table_cols=", ".join("[{}]".format(c) for c in columns),
table=escaped_name(self.name), table=self._fullname,
) )
) )
@ -2090,15 +2088,16 @@ class Table(Queryable):
self.add_column(magic_lookup_column, int) self.add_column(magic_lookup_column, int)
# And populate it # And populate it
lookup_table_full = _fullname(self.schema_name, table)
self.db.execute( self.db.execute(
"UPDATE {table} SET [{magic_lookup_column}] = (SELECT id FROM {lookup_table} WHERE {where})".format( "UPDATE {table} SET [{magic_lookup_column}] = (SELECT id FROM {lookup_table} WHERE {where})".format(
table=escaped_name(self.name), table=self._fullname,
magic_lookup_column=magic_lookup_column, magic_lookup_column=magic_lookup_column,
lookup_table=escaped_name(table), lookup_table=lookup_table_full,
where=" AND ".join( where=" AND ".join(
"{table}.[{column}] IS {lookup_table}.[{lookup_column}]".format( "{table}.[{column}] IS {lookup_table}.[{lookup_column}]".format(
table=escaped_name(self.name), table=self._fullname,
lookup_table=escaped_name(table), lookup_table=lookup_table_full,
column=column, column=column,
lookup_column=rename.get(column) or column, lookup_column=rename.get(column) or column,
) )
@ -2178,7 +2177,7 @@ class Table(Queryable):
.strip() .strip()
.format( .format(
index_name=created_index_name, index_name=created_index_name,
table_name=escaped_name(self.name), table_name=self._fullname,
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 "",
@ -2226,7 +2225,7 @@ class Table(Queryable):
fk_col_type = None fk_col_type = None
if fk is not None: if fk is not None:
# fk must be a valid table # fk must be a valid table
if fk not in self.db.table_names(): if fk not in self.db.table_names(schema=self.schema_name):
raise AlterError("table '{}' does not exist".format(fk)) raise AlterError("table '{}' does not exist".format(fk))
# if fk_col specified, must be a valid column # if fk_col specified, must be a valid column
if fk_col is not None: if fk_col is not None:
@ -2249,7 +2248,7 @@ class Table(Queryable):
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 {table} ADD COLUMN [{col_name}] {col_type}{not_null_default};".format(
table=escaped_name(self.name), table=self._fullname,
col_name=col_name, col_name=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 "",
@ -2266,7 +2265,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(escaped_name(self.name))) self.db.execute("DROP TABLE {}".format(self._fullname))
except sqlite3.OperationalError: except sqlite3.OperationalError:
if not ignore: if not ignore:
raise raise
@ -2292,7 +2291,9 @@ class Table(Queryable):
possibilities.append(column_without_id + "s") possibilities.append(column_without_id + "s")
elif not column.endswith("s"): elif not column.endswith("s"):
possibilities.append(column + "s") possibilities.append(column + "s")
existing_tables = {t.lower(): t for t in self.db.table_names()} existing_tables = {
t.lower(): t for t in self.db.table_names(schema=self.schema_name)
}
for table in possibilities: for table in possibilities:
if table in existing_tables: if table in existing_tables:
return existing_tables[table] return existing_tables[table]
@ -2444,7 +2445,7 @@ class Table(Queryable):
:param tokenize: Custom SQLite tokenizer to use, for example ``"porter"`` to enable Porter stemming. :param tokenize: Custom SQLite tokenizer to use, for example ``"porter"`` to enable Porter stemming.
:param replace: Should any existing FTS index for this table be replaced by the new one? :param replace: Should any existing FTS index for this table be replaced by the new one?
""" """
table_name = tablename(self.name) table_name = self.name
create_fts_sql = ( create_fts_sql = (
textwrap.dedent( textwrap.dedent(
""" """
@ -2468,7 +2469,9 @@ class Table(Queryable):
fts_schema = self.db["{}_fts".format(table_name)].schema fts_schema = self.db["{}_fts".format(table_name)].schema
if fts_schema != create_fts_sql: if fts_schema != create_fts_sql:
should_recreate = True should_recreate = True
expected_triggers = {table_name + suffix for suffix in ("_ai", "_ad", "_au")} expected_triggers = {
table_name + suffix for suffix in ("_ai", "_ad", "_au")
}
existing_triggers = {t.name for t in self.triggers} existing_triggers = {t.name for t in self.triggers}
has_triggers = existing_triggers.issuperset(expected_triggers) has_triggers = existing_triggers.issuperset(expected_triggers)
if has_triggers != create_triggers: if has_triggers != create_triggers:
@ -2528,7 +2531,7 @@ class Table(Queryable):
) )
.strip() .strip()
.format( .format(
table=tablename(self.name), columns=", ".join("[{}]".format(c) for c in columns) table=self.name, columns=", ".join("[{}]".format(c) for c in columns)
) )
) )
self.db.executescript(sql) self.db.executescript(sql)
@ -2564,11 +2567,9 @@ class Table(Queryable):
fts_table = self.detect_fts() fts_table = self.detect_fts()
if fts_table is None: if fts_table is None:
# Assume this is itself an FTS table # Assume this is itself an FTS table
fts_table = escaped_name(self.name) fts_table = self._fullname
self.db.execute( self.db.execute(
"INSERT INTO {table}({table}) VALUES('rebuild');".format( "INSERT INTO {table}({table}) VALUES('rebuild');".format(table=fts_table)
table=fts_table
)
) )
return self return self
@ -2588,7 +2589,7 @@ class Table(Queryable):
) )
""" """
).strip() ).strip()
table_name = tablename(self.name) table_name = self.name
args = { args = {
"like": "%VIRTUAL TABLE%USING FTS%content=[{}]%".format(table_name), "like": "%VIRTUAL TABLE%USING FTS%content=[{}]%".format(table_name),
"like2": '%VIRTUAL TABLE%USING FTS%content="{}"%'.format(table_name), "like2": '%VIRTUAL TABLE%USING FTS%content="{}"%'.format(table_name),
@ -2681,7 +2682,7 @@ 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=escaped_name(self.name), dbtable=self._fullname,
where_clause="\n where {}".format(where) if where else "", where_clause="\n where {}".format(where) if where else "",
original=original, original=original,
columns=columns_sql, columns=columns_sql,
@ -2753,7 +2754,7 @@ class Table(Queryable):
self.get(pk_values) self.get(pk_values)
wheres = ["[{}] = ?".format(pk_name) for pk_name in self.pks] wheres = ["[{}] = ?".format(pk_name) for pk_name in self.pks]
sql = "delete from {table} where {wheres}".format( sql = "delete from {table} where {wheres}".format(
table=escaped_name(self.name), wheres=" and ".join(wheres) table=self._fullname, wheres=" and ".join(wheres)
) )
with self.db.conn: with self.db.conn:
self.db.execute(sql, pk_values) self.db.execute(sql, pk_values)
@ -2777,7 +2778,7 @@ class Table(Queryable):
""" """
if not self.exists(): if not self.exists():
return self return self
sql = f"delete from {escaped_name(self.name)}" sql = f"delete from {self._fullname}"
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 [])
@ -2823,7 +2824,7 @@ class Table(Queryable):
wheres = ["[{}] = ?".format(pk_name) for pk_name in pks] wheres = ["[{}] = ?".format(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 {table} set {sets} where {wheres}".format(
table=escaped_name(self.name), sets=", ".join(sets), wheres=" and ".join(wheres) table=self._fullname, sets=", ".join(sets), wheres=" and ".join(wheres)
) )
with self.db.conn: with self.db.conn:
try: try:
@ -2904,7 +2905,7 @@ class Table(Queryable):
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 {table} set {sets}{where};".format(
table=escaped_name(self.name), table=self._fullname,
sets=", ".join( sets=", ".join(
[ [
"[{output_column}] = {fn_name}([{column}])".format( "[{output_column}] = {fn_name}([{column}])".format(
@ -3026,7 +3027,7 @@ class Table(Queryable):
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 = "INSERT OR IGNORE INTO {table}({cols}) VALUES({placeholders});".format(
table=escaped_name(self.name), table=self._fullname,
cols=", ".join(["[{}]".format(p) for p in placeholders]), cols=", ".join(["[{}]".format(p) for p in placeholders]),
placeholders=", ".join(["?" for p in placeholders]), placeholders=", ".join(["?" for p in placeholders]),
) )
@ -3037,7 +3038,7 @@ class Table(Queryable):
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 {table} SET {pairs} WHERE {wheres}".format(
table=escaped_name(self.name), table=self._fullname,
pairs=", ".join( pairs=", ".join(
"[{}] = {}".format(col, conversions.get(col, "?")) "[{}] = {}".format(col, conversions.get(col, "?"))
for col in set_cols for col in set_cols
@ -3067,7 +3068,7 @@ class Table(Queryable):
INSERT {or_what}INTO {table} ({columns}) VALUES {rows}; INSERT {or_what}INTO {table} ({columns}) VALUES {rows};
""".strip().format( """.strip().format(
or_what=or_what, or_what=or_what,
table=escaped_name(self.name), table=self._fullname,
columns=", ".join("[{}]".format(c) for c in all_columns), columns=", ".join("[{}]".format(c) for c in all_columns),
rows=", ".join( rows=", ".join(
"({placeholders})".format( "({placeholders})".format(
@ -3325,7 +3326,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(escaped_name(self.name))) self.db.execute("DELETE FROM {};".format(self._fullname))
for chunk in chunks(itertools.chain([first_record], records), batch_size): for chunk in chunks(itertools.chain([first_record], records), batch_size):
chunk = list(chunk) chunk = list(chunk)
num_records_processed += len(chunk) num_records_processed += len(chunk)
@ -3809,7 +3810,9 @@ class Table(Queryable):
:param column_name: Geometry column to create the spatial index against :param column_name: Geometry column to create the spatial index against
""" """
if f"idx_{self.name}_{column_name}" in self.db.table_names(): if f"idx_{self.name}_{column_name}" in self.db.table_names(
schema=self.schema_name
):
return False return False
cursor = self.db.execute( cursor = self.db.execute(
@ -3836,7 +3839,7 @@ class View(Queryable):
""" """
try: try:
self.db.execute("DROP VIEW {}".format(escaped_name(self.name))) self.db.execute("DROP VIEW {}".format(self._fullname))
except sqlite3.OperationalError: except sqlite3.OperationalError:
if not ignore: if not ignore:
raise raise