remote/sqlite-utils
1
0
Fork 0
mirror of https://github.com/simonw/sqlite-utils.git synced 2026-07-25 18:34:32 +02:00
Code Issues Projects Releases Packages Wiki Activity

db.table() only returns tables, added db.view()

Browse source 
* db.table() only returns tables, added db.view(), closes #657
* Massive documentation update for db.table()

Refs #656
This commit is contained in:
Simon Willison 2025-05-08 23:19:36 -07:00 • committed by GitHub
commit 094b010fd8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 276 additions and 229 deletions
Download patch file Download diff file Expand all files Collapse all files

353
docs/python-api.rst
View file

File diff suppressed because it is too large Load diff

10
sqlite_utils/cli.py
View file

@ -186,6 +186,8 @@ def tables(
if schema:
headers.append("schema")
method = db.view if views else db.table
def _iter():
if views:
items = db.view_names()
@ -194,15 +196,15 @@ def tables(
for name in items:
row = [name]
if counts:
row.append(db[name].count)
row.append(method(name).count)
if columns:
cols = [c.name for c in db[name].columns]
cols = [c.name for c in method(name).columns]
if csv:
row.append("\n".join(cols))
else:
row.append(cols)
if schema:
row.append(db[name].schema)
row.append(method(name).schema)
yield row
if table or fmt:
@ -1693,7 +1695,7 @@ def create_view(path, view, select, ignore, replace, load_extension):
if ignore:
return
elif replace:
db[view].drop()
db.view(view).drop()
else:
raise click.ClickException(
'View "{}" already exists. Use --replace to delete and replace it.'.format(

47
sqlite_utils/db.py
View file

@ -247,6 +247,10 @@ class NoTable(Exception):
"Specified table does not exist"
class NoView(Exception):
"Specified view does not exist"
class BadPrimaryKey(Exception):
"Table does not have a single obvious primary key"
@ -419,6 +423,8 @@ class Database:
:param table_name: The name of the table
"""
if table_name in self.view_names():
return self.view(table_name)
return self.table(table_name)
def __repr__(self) -> str:
@ -541,7 +547,7 @@ class Database:
self._tracer(sql, None)
return self.conn.executescript(sql)
def table(self, table_name: str, **kwargs) -> Union["Table", "View"]:
def table(self, table_name: str, **kwargs) -> "Table":
"""
Return a table object, optionally configured with default options.
@ -550,10 +556,19 @@ class Database:
:param table_name: Name of the table
"""
if table_name in self.view_names():
return View(self, table_name, **kwargs)
else:
kwargs.setdefault("strict", self.strict)
return Table(self, table_name, **kwargs)
raise NoTable("Table {} is actually a view".format(table_name))
kwargs.setdefault("strict", self.strict)
return Table(self, table_name, **kwargs)
def view(self, view_name: str) -> "View":
"""
Return a view object.
:param view_name: Name of the view
"""
if view_name not in self.view_names():
raise NoView("View {} does not exist".format(view_name))
return View(self, view_name)
def quote(self, value: str) -> str:
"""
@ -637,12 +652,12 @@ class Database:
@property
def tables(self) -> List["Table"]:
"List of Table objects in this database."
return cast(List["Table"], [self[name] for name in self.table_names()])
return [self.table(name) for name in self.table_names()]
@property
def views(self) -> List["View"]:
"List of View objects in this database."
return cast(List["View"], [self[name] for name in self.view_names()])
return [self.view(name) for name in self.view_names()]
@property
def triggers(self) -> List[Trigger]:
@ -808,7 +823,7 @@ class Database:
or a tuple of (column, other_table, other_column), or a tuple of
(table, column, other_table, other_column)
"""
table = cast(Table, self[name])
table = self.table(name)
if all(isinstance(fk, ForeignKey) for fk in foreign_keys):
return cast(List[ForeignKey], foreign_keys)
if all(isinstance(fk, str) for fk in foreign_keys):
@ -1039,11 +1054,11 @@ class Database:
# Transform table to match the new definition if table already exists:
if self[name].exists():
if ignore:
return cast(Table, self[name])
return self.table(name)
elif replace:
self[name].drop()
if transform and self[name].exists():
table = cast(Table, self[name])
table = self.table(name)
should_transform = False
# First add missing columns and figure out columns to drop
existing_columns = table.columns_dict
@ -1109,7 +1124,7 @@ class Database:
strict=strict,
)
self.execute(sql)
created_table = self.table(
return self.table(
name,
pk=pk,
foreign_keys=foreign_keys,
@ -1119,7 +1134,6 @@ class Database:
hash_id=hash_id,
hash_id_columns=hash_id_columns,
)
return cast(Table, created_table)
def rename_table(self, name: str, new_name: str):
"""
@ -1196,12 +1210,9 @@ class Database:
# Verify that all tables and columns exist
for table, column, other_table, other_column in foreign_keys:
if not self[table].exists():
if not self.table(table).exists():
raise AlterError("No such table: {}".format(table))
table_obj = self[table]
if not isinstance(table_obj, Table):
raise AlterError("Must be a table, not a view: {}".format(table))
table_obj = cast(Table, table_obj)
table_obj = self.table(table)
if column not in table_obj.columns_dict:
raise AlterError("No such column: {} in {}".format(column, table))
if not self[other_table].exists():
@ -1231,7 +1242,7 @@ class Database:
by_table.setdefault(fk[0], []).append(fk)
for table, fks in by_table.items():
cast(Table, self[table]).transform(add_foreign_keys=fks)
self.table(table).transform(add_foreign_keys=fks)
self.vacuum()

13
tests/test_create.py
View file

@ -8,6 +8,8 @@ from sqlite_utils.db import (
ForeignKey,
Table,
View,
NoTable,
NoView,
)
from sqlite_utils.utils import hash_record, sqlite3
import collections
@ -1367,3 +1369,14 @@ def test_create_strict(fresh_db, strict):
table = fresh_db["t"]
table.create({"id": int}, strict=strict)
assert table.strict == strict or not fresh_db.supports_strict
def test_bad_table_and_view_exceptions(fresh_db):
fresh_db.table("t").insert({"id": 1}, pk="id")
fresh_db.create_view("v", "select * from t")
with pytest.raises(NoTable) as ex:
fresh_db.table("v")
assert ex.value.args[0] == "Table v is actually a view"
with pytest.raises(NoView) as ex2:
fresh_db.view("t")
assert ex2.value.args[0] == "View t does not exist"

4
tests/test_enable_counts.py
View file

@ -129,7 +129,7 @@ def test_uses_counts_after_enable_counts(counts_db_path):
db = Database(counts_db_path)
logged = []
with db.tracer(lambda sql, parameters: logged.append((sql, parameters))):
assert db["foo"].count == 1
assert db.table("foo").count == 1
assert logged == [
("select name from sqlite_master where type = 'view'", None),
("select count(*) from [foo]", []),
@ -138,7 +138,7 @@ def test_uses_counts_after_enable_counts(counts_db_path):
assert not db.use_counts_table
db.enable_counts()
assert db.use_counts_table
assert db["foo"].count == 1
assert db.table("foo").count == 1
assert logged == [
(
"CREATE TABLE IF NOT EXISTS [_counts](\n [table] TEXT PRIMARY KEY,\n count INTEGER DEFAULT 0\n);",

86
tests/test_tracer.py
View file

@ -6,20 +6,21 @@ def test_tracer():
db = Database(
memory=True, tracer=lambda sql, params: collected.append((sql, params))
)
db["dogs"].insert({"name": "Cleopaws"})
db["dogs"].enable_fts(["name"])
db["dogs"].search("Cleopaws")
dogs = db.table("dogs")
dogs.insert({"name": "Cleopaws"})
dogs.enable_fts(["name"])
dogs.search("Cleopaws")
assert collected == [
("PRAGMA recursive_triggers=on;", 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 = '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 = 'view'", None),
("CREATE TABLE [dogs] (\n [name] TEXT\n);\n ", None),
("select name from sqlite_master where type = 'view'", None),
("INSERT INTO [dogs] ([name]) VALUES (?)", ["Cleopaws"]),
("select name from sqlite_master where type = 'view'", None),
(
"CREATE VIRTUAL TABLE [dogs_fts] USING FTS5 (\n [name],\n content=[dogs]\n)",
None,
@ -28,7 +29,6 @@ def test_tracer():
"INSERT INTO [dogs_fts] (rowid, [name])\n SELECT rowid, [name] FROM [dogs];",
None,
),
("select name from sqlite_master where type = 'view'", None),
]
@ -40,60 +40,58 @@ def test_with_tracer():
db = Database(memory=True)
db["dogs"].insert({"name": "Cleopaws"})
db["dogs"].enable_fts(["name"])
dogs = db.table("dogs")
dogs.insert({"name": "Cleopaws"})
dogs.enable_fts(["name"])
assert len(collected) == 0
with db.tracer(tracer):
list(db["dogs"].search("Cleopaws"))
list(dogs.search("Cleopaws"))
assert len(collected) == 5
assert collected == [
("select name from sqlite_master where type = 'view'", None),
(
(
"SELECT name FROM sqlite_master\n"
" WHERE rootpage = 0\n"
" AND (\n"
" sql LIKE :like\n"
" OR sql LIKE :like2\n"
" OR (\n"
" tbl_name = :table\n"
" AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
" )\n"
" )",
{
"like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%",
"like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
"table": "dogs",
},
)
"SELECT name FROM sqlite_master\n"
" WHERE rootpage = 0\n"
" AND (\n"
" sql LIKE :like\n"
" OR sql LIKE :like2\n"
" OR (\n"
" tbl_name = :table\n"
" AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
" )\n"
" )",
{
"like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%",
"like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
"table": "dogs",
},
),
("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",)),
(
(
"with original as (\n"
" select\n"
" rowid,\n"
" *\n"
" from [dogs]\n"
")\n"
"select\n"
" [original].*\n"
"from\n"
" [original]\n"
" join [dogs_fts] on [original].rowid = [dogs_fts].rowid\n"
"where\n"
" [dogs_fts] match :query\n"
"order by\n"
" [dogs_fts].rank"
),
"with original as (\n"
" select\n"
" rowid,\n"
" *\n"
" from [dogs]\n"
")\n"
"select\n"
" [original].*\n"
"from\n"
" [original]\n"
" join [dogs_fts] on [original].rowid = [dogs_fts].rowid\n"
"where\n"
" [dogs_fts] match :query\n"
"order by\n"
" [dogs_fts].rank",
{"query": "Cleopaws"},
),
]
# Outside the with block collected should not be appended to
db["dogs"].insert({"name": "Cleopaws"})
dogs.insert({"name": "Cleopaws"})
assert len(collected) == 5