From 2ca63e3b2de5408a860c6c7c1852deb9a138279e Mon Sep 17 00:00:00 2001
From: Simon Willison <swillison@gmail.com>
Date: Mon, 2 Sep 2019 17:09:41 -0700
Subject: [PATCH] db.triggers and table.triggers introspection (#60)

Closes #59
---
 docs/python-api.rst      | 11 +++++++++++
 sqlite_utils/db.py       | 23 +++++++++++++++++++++++
 tests/test_introspect.py | 27 +++++++++++++++++++++++++++
 3 files changed, 61 insertions(+)

diff --git a/docs/python-api.rst b/docs/python-api.rst
index 9d272c3..997bd97 100644
--- a/docs/python-api.rst
+++ b/docs/python-api.rst
@@ -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
 =========================
 
diff --git a/sqlite_utils/db.py b/sqlite_utils/db.py
index b7837c7..e36e191 100644
--- a/sqlite_utils/db.py
+++ b/sqlite_utils/db.py
@@ -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,
diff --git a/tests/test_introspect.py b/tests/test_introspect.py
index b04af6d..7c01a0b 100644
--- a/tests/test_introspect.py
+++ b/tests/test_introspect.py
@@ -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