db.supports_strict and table.strict properties, refs #344

This commit is contained in:
Simon Willison 2021-11-29 14:19:30 -08:00
commit e3f108e0f3
3 changed files with 65 additions and 14 deletions

View file

@ -1609,8 +1609,8 @@ A more useful example: if you are working with `SpatiaLite <https://www.gaia-gis
.. _python_api_introspection: .. _python_api_introspection:
Introspection Introspecting tables and views
============= ==============================
If you have loaded an existing table or view, you can use introspection to find out more about it:: If you have loaded an existing table or view, you can use introspection to find out more about it::
@ -1741,6 +1741,18 @@ The ``.schema`` property outputs the table's schema as a SQL string::
FOREIGN KEY ("qCareAssistant") REFERENCES [qCareAssistant](id), FOREIGN KEY ("qCareAssistant") REFERENCES [qCareAssistant](id),
FOREIGN KEY ("qLegalStatus") REFERENCES [qLegalStatus](id)) FOREIGN KEY ("qLegalStatus") REFERENCES [qLegalStatus](id))
.. _python_api_introspection_strict:
.strict
-------
The ``.strict`` property identifies if the table is a `SQLite STRICT table <https://www.sqlite.org/stricttables.html>`__.
::
>>> db["ny_times_us_counties"].strict
False
.. _python_api_introspection_indexes: .. _python_api_introspection_indexes:
.indexes .indexes

View file

@ -18,6 +18,7 @@ import json
import os import os
import pathlib import pathlib
import re import re
import secrets
from sqlite_fts4 import rank_bm25 # type: ignore from sqlite_fts4 import rank_bm25 # type: ignore
import sys import sys
import textwrap import textwrap
@ -521,6 +522,19 @@ class Database:
sqls.append(sql) sqls.append(sql)
return "\n".join(sqls) return "\n".join(sqls)
@property
def supports_strict(self):
try:
table_name = "t{}".format(secrets.token_hex(16))
with self.conn:
self.conn.execute(
"create table {} (name text) strict".format(table_name)
)
self.conn.execute("drop table {}".format(table_name))
return True
except Exception as e:
return False
@property @property
def journal_mode(self) -> str: def journal_mode(self) -> str:
"Current ``journal_mode`` of this database." "Current ``journal_mode`` of this database."
@ -1219,6 +1233,13 @@ class Table(Queryable):
"``{trigger_name: sql}`` dictionary of triggers defined on this table." "``{trigger_name: sql}`` dictionary of triggers defined on this table."
return {trigger.name: trigger.sql for trigger in self.triggers} return {trigger.name: trigger.sql for trigger in self.triggers}
@property
def strict(self) -> bool:
"Is this a STRICT table?"
table_suffix = self.schema.split(")")[-1].strip().upper()
table_options = [bit.strip() for bit in table_suffix.split(",")]
return "STRICT" in table_options
def create( def create(
self, self,
columns: Dict[str, Any], columns: Dict[str, Any],

View file

@ -252,15 +252,33 @@ def test_has_counts_triggers(fresh_db):
), ),
], ],
) )
def test_virtual_table_using(sql, expected_name, expected_using): def test_virtual_table_using(fresh_db, sql, expected_name, expected_using):
db = Database(memory=True) fresh_db.execute(sql)
db.execute(sql) assert fresh_db[expected_name].virtual_table_using == expected_using
assert db[expected_name].virtual_table_using == expected_using
def test_use_rowid(): def test_use_rowid(fresh_db):
db = Database(memory=True) fresh_db["rowid_table"].insert({"name": "Cleo"})
db["rowid_table"].insert({"name": "Cleo"}) fresh_db["regular_table"].insert({"id": 1, "name": "Cleo"}, pk="id")
db["regular_table"].insert({"id": 1, "name": "Cleo"}, pk="id") assert fresh_db["rowid_table"].use_rowid
assert db["rowid_table"].use_rowid assert not fresh_db["regular_table"].use_rowid
assert not db["regular_table"].use_rowid
@pytest.mark.skipif(
not Database(memory=True).supports_strict,
reason="Needs SQLite version that supports strict",
)
@pytest.mark.parametrize(
"create_table,expected_strict",
(
("create table t (id integer) strict", True),
("create table t (id integer) STRICT", True),
("create table t (id integer primary key) StriCt, WITHOUT ROWID", True),
("create table t (id integer primary key) WITHOUT ROWID", False),
("create table t (id integer)", False),
),
)
def test_table_strict(fresh_db, create_table, expected_strict):
fresh_db.execute(create_table)
table = fresh_db["t"]
assert table.strict == expected_strict