db.triggers and table.triggers introspection (#60)

Closes #59
This commit is contained in:
Simon Willison 2019-09-02 17:09:41 -07:00 committed by GitHub
commit 2ca63e3b2d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 61 additions and 0 deletions

View file

@ -868,6 +868,17 @@ The ``.indexes`` property shows you all indexes created for a table. It is not a
Index(seq=4, name='"Street_Tree_List_qCaretaker"', unique=0, origin='c', partial=0, columns=['qCaretaker']),
Index(seq=5, name='"Street_Tree_List_PlantType"', unique=0, origin='c', partial=0, columns=['PlantType'])]
The ``.triggers`` property lists database triggers. It can be used on both database and table objects.
::
>>> db["authors"].triggers
[Trigger(name='authors_ai', table='authors', sql='CREATE TRIGGER [authors_ai] AFTER INSERT...'),
Trigger(name='authors_ad', table='authors', sql="CREATE TRIGGER [authors_ad] AFTER DELETE..."),
Trigger(name='authors_au', table='authors', sql="CREATE TRIGGER [authors_au] AFTER UPDATE")]
>>> db.triggers
... similar output to db["authors"].triggers
Enabling full-text search
=========================

View file

@ -20,6 +20,9 @@ ForeignKey = namedtuple(
"ForeignKey", ("table", "column", "other_table", "other_column")
)
Index = namedtuple("Index", ("seq", "name", "unique", "origin", "partial", "columns"))
Trigger = namedtuple("Trigger", ("name", "table", "sql"))
DEFAULT = object()
COLUMN_TYPE_MAPPING = {
@ -146,6 +149,15 @@ class Database:
def views(self):
return [self[name] for name in self.view_names()]
@property
def triggers(self):
return [
Trigger(*r)
for r in self.conn.execute(
"select name, tbl_name, sql from sqlite_master where type = 'trigger'"
).fetchall()
]
def execute_returning_dicts(self, sql, params=None):
cursor = self.conn.execute(sql, params or tuple())
keys = [d[0] for d in cursor.description]
@ -561,6 +573,17 @@ class Table(Queryable):
indexes.append(Index(**row))
return indexes
@property
def triggers(self):
return [
Trigger(*r)
for r in self.db.conn.execute(
"select name, tbl_name, sql from sqlite_master where type = 'trigger'"
" and tbl_name = ?",
(self.name,),
).fetchall()
]
def create(
self,
columns,

View file

@ -121,3 +121,30 @@ def test_guess_foreign_table(fresh_db, column, expected_table_guess):
def test_pks(fresh_db, pk, expected):
fresh_db["foo"].insert_all([{"id": 1, "id2": 2}], pk=pk)
assert expected == fresh_db["foo"].pks
def test_triggers(fresh_db):
assert [] == fresh_db.triggers
authors = fresh_db["authors"]
authors.insert_all(
[
{"name": "Frank Herbert", "famous_works": "Dune"},
{"name": "Neal Stephenson", "famous_works": "Cryptonomicon"},
]
)
fresh_db["other"].insert({"foo": "bar"})
assert [] == authors.triggers
assert [] == fresh_db["other"].triggers
authors.enable_fts(
["name", "famous_works"], fts_version="FTS4", create_triggers=True
)
expected_triggers = {
("authors_ai", "authors"),
("authors_ad", "authors"),
("authors_au", "authors"),
}
assert expected_triggers == {(t.name, t.table) for t in fresh_db.triggers}
assert expected_triggers == {
(t.name, t.table) for t in fresh_db["authors"].triggers
}
assert [] == fresh_db["other"].triggers