Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Add support for constraint triggers.

 * docs/trigger.rst: Indicate that the Trigger class now covers both
   regular and constraint triggers.
 * pyrseas/dbobject/trigger.py (Trigger.create): Add support for
   CONSTRAINT, DEFERRABLE and INITIALLY DEFERRED keywords.
   (TriggerDict.query): Change to select constraint triggers and their
   features.
 * tests/dbobject/test_trigger.py: New test classes for constraint
   triggers.
  • Loading branch information...
commit 38319e62587893b6fb28657b72f469138c041143 1 parent bb430bb
@jmafc jmafc authored
View
7 docs/trigger.rst
@@ -11,7 +11,12 @@ Trigger
-------
:class:`Trigger` is derived from
-:class:`~pyrseas.dbobject.DbSchemaObject` and represents a trigger.
+:class:`~pyrseas.dbobject.DbSchemaObject` and represents a PostgreSQL
+regular `trigger
+<http://www.postgresql.org/docs/current/static/sql-createtrigger.html>`_
+or `constraint trigger
+<http://www.postgresql.org/docs/current/static/sql-createconstraint.html>`_.
+
.. autoclass:: Trigger
View
27 pyrseas/dbobject/trigger.py
@@ -46,6 +46,15 @@ def create(self):
:return: SQL statements
"""
stmts = []
+ constr = defer = ''
+ if hasattr(self, 'constraint') and self.constraint:
+ constr = "CONSTRAINT "
+ if hasattr(self, 'deferrable') and self.deferrable:
+ defer = "DEFERRABLE "
+ if hasattr(self, 'initially_deferred') and self.initially_deferred:
+ defer += "INITIALLY DEFERRED"
+ if defer:
+ defer = '\n ' + defer
evts = " OR ".join(self.events).upper()
if hasattr(self, 'columns') and 'update' in self.events:
evts = evts.replace("UPDATE", "UPDATE OF %s" % (
@@ -53,9 +62,10 @@ def create(self):
cond = ''
if hasattr(self, 'condition'):
cond = "\n WHEN (%s)" % self.condition
- stmts.append("CREATE TRIGGER %s\n %s %s ON %s\n FOR EACH %s"
+ stmts.append("CREATE %sTRIGGER %s\n %s %s ON %s%s\n FOR EACH %s"
"%s\n EXECUTE PROCEDURE %s" % (
- self.name, self.timing.upper(), evts, self._table.qualname(),
+ constr, self.name, self.timing.upper(), evts,
+ self._table.qualname(), defer,
self.level.upper(), cond, self.procedure))
if hasattr(self, 'description'):
stmts.append(self.comment())
@@ -78,15 +88,19 @@ def diff_map(self, intrigger):
QUERY_PRE90 = \
"""SELECT nspname AS schema, relname AS table,
- tgname AS name, pg_get_triggerdef(t.oid) AS definition,
+ tgname AS name, tgisconstraint AS constraint,
+ tgdeferrable AS deferrable,
+ tginitdeferred AS initially_deferred,
+ pg_get_triggerdef(t.oid) AS definition,
NULL AS columns, description
FROM pg_trigger t
JOIN pg_class c ON (t.tgrelid = c.oid)
JOIN pg_namespace n ON (c.relnamespace = n.oid)
JOIN pg_roles ON (n.nspowner = pg_roles.oid)
+ LEFT JOIN pg_constraint cn ON (tgconstraint = cn.oid)
LEFT JOIN pg_description d
ON (t.oid = d.objoid AND d.objsubid = 0)
- WHERE NOT tgisconstraint
+ WHERE contype != 'f' OR contype IS NULL
AND (nspname != 'pg_catalog' AND nspname != 'information_schema')
ORDER BY 1, 2, 3"""
@@ -98,11 +112,16 @@ class TriggerDict(DbObjectDict):
query = \
"""SELECT nspname AS schema, relname AS table,
tgname AS name, pg_get_triggerdef(t.oid) AS definition,
+ CASE WHEN contype = 't' THEN true ELSE false END AS
+ constraint,
+ tgdeferrable AS deferrable,
+ tginitdeferred AS initially_deferred,
tgattr AS columns, description
FROM pg_trigger t
JOIN pg_class c ON (t.tgrelid = c.oid)
JOIN pg_namespace n ON (c.relnamespace = n.oid)
JOIN pg_roles ON (n.nspowner = pg_roles.oid)
+ LEFT JOIN pg_constraint cn ON (tgconstraint = cn.oid)
LEFT JOIN pg_description d
ON (t.oid = d.objoid AND d.objsubid = 0)
WHERE NOT tgisinternal
View
97 tests/dbobject/test_trigger.py
@@ -88,6 +88,44 @@ def test_map_trigger_comment(self):
['tr1']['description'], 'Test trigger tr1')
+class ConstraintTriggerToMapTestCase(PyrseasTestCase):
+ """Test mapping of existing constraint triggers"""
+
+ def setUp(self):
+ super(self.__class__, self).setUp()
+ if self.db.version < 90000:
+ if not self.db.is_plpgsql_installed():
+ self.db.execute_commit("CREATE LANGUAGE plpgsql")
+
+ def test_map_trigger(self):
+ "Map a simple constraint trigger"
+ self.db.execute(CREATE_TABLE_STMT)
+ self.db.execute(CREATE_FUNC_STMT)
+ expmap = {'tr1': {'constraint': True, 'timing': 'after',
+ 'events': ['insert', 'update'],
+ 'level': 'row', 'procedure': 'f1()'}}
+ dbmap = self.db.execute_and_map(
+ "CREATE CONSTRAINT TRIGGER tr1 AFTER INSERT OR UPDATE ON t1 "
+ "FOR EACH ROW EXECUTE PROCEDURE f1()")
+ self.assertEqual(dbmap['schema public']['table t1']['triggers'],
+ expmap)
+
+ def test_map_trigger_deferrable(self):
+ "Map a deferrable, initially deferred constraint trigger"
+ self.db.execute(CREATE_TABLE_STMT)
+ self.db.execute(CREATE_FUNC_STMT)
+ expmap = {'tr1': {'constraint': True, 'deferrable': True,
+ 'initially_deferred': True, 'timing': 'after',
+ 'events': ['insert', 'update'],
+ 'level': 'row', 'procedure': 'f1()'}}
+ dbmap = self.db.execute_and_map(
+ "CREATE CONSTRAINT TRIGGER tr1 AFTER INSERT OR UPDATE ON t1 "
+ "DEFERRABLE INITIALLY DEFERRED "
+ "FOR EACH ROW EXECUTE PROCEDURE f1()")
+ self.assertEqual(dbmap['schema public']['table t1']['triggers'],
+ expmap)
+
+
class TriggerToSqlTestCase(PyrseasTestCase):
"""Test SQL generation from input triggers"""
@@ -329,10 +367,69 @@ def test_change_trigger_comment(self):
"COMMENT ON TRIGGER tr1 ON t1 IS 'Changed trigger tr1'"])
+class ConstraintTriggerToSqlTestCase(PyrseasTestCase):
+ """Test SQL generation from input triggers"""
+
+ def setUp(self):
+ super(self.__class__, self).setUp()
+ if self.db.version < 90000:
+ if not self.db.is_plpgsql_installed():
+ self.db.execute_commit("CREATE LANGUAGE plpgsql")
+
+ def test_create_trigger(self):
+ "Create a constraint trigger"
+ inmap = new_std_map()
+ inmap.update({'language plpgsql': {'trusted': True}})
+ inmap['schema public'].update({'function f1()': {
+ 'language': 'plpgsql', 'returns': 'trigger',
+ 'source': FUNC_SRC}})
+ inmap['schema public'].update({'table t1': {
+ 'columns': [{'c1': {'type': 'integer'}},
+ {'c2': {'type': 'text'}},
+ {'c3': {'type': 'timestamp with time zone'}}],
+ 'triggers': {'tr1': {
+ 'constraint': True, 'timing': 'after',
+ 'events': ['insert', 'update'],
+ 'level': 'row', 'procedure': 'f1()'}}}})
+ dbsql = self.db.process_map(inmap)
+ self.assertEqual(fix_indent(dbsql[1]), CREATE_FUNC_STMT)
+ self.assertEqual(fix_indent(dbsql[2]), CREATE_TABLE_STMT)
+ self.assertEqual(fix_indent(dbsql[3]), "CREATE CONSTRAINT TRIGGER tr1 "
+ "AFTER INSERT OR UPDATE ON t1 "
+ "FOR EACH ROW EXECUTE PROCEDURE f1()")
+
+ def test_create_trigger_deferrable(self):
+ "Create a deferrable constraint trigger"
+ inmap = new_std_map()
+ inmap.update({'language plpgsql': {'trusted': True}})
+ inmap['schema public'].update({'function f1()': {
+ 'language': 'plpgsql', 'returns': 'trigger',
+ 'source': FUNC_SRC}})
+ inmap['schema public'].update({'table t1': {
+ 'columns': [{'c1': {'type': 'integer'}},
+ {'c2': {'type': 'text'}},
+ {'c3': {'type': 'timestamp with time zone'}}],
+ 'triggers': {'tr1': {
+ 'constraint': True, 'deferrable': True,
+ 'initially_deferred': True, 'timing': 'after',
+ 'events': ['insert', 'update'],
+ 'level': 'row', 'procedure': 'f1()'}}}})
+ dbsql = self.db.process_map(inmap)
+ self.assertEqual(fix_indent(dbsql[1]), CREATE_FUNC_STMT)
+ self.assertEqual(fix_indent(dbsql[2]), CREATE_TABLE_STMT)
+ self.assertEqual(fix_indent(dbsql[3]), "CREATE CONSTRAINT TRIGGER tr1 "
+ "AFTER INSERT OR UPDATE ON t1 DEFERRABLE INITIALLY "
+ "DEFERRED FOR EACH ROW EXECUTE PROCEDURE f1()")
+
+
def suite():
tests = unittest.TestLoader().loadTestsFromTestCase(TriggerToMapTestCase)
tests.addTest(unittest.TestLoader().loadTestsFromTestCase(
TriggerToSqlTestCase))
+ tests.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ ConstraintTriggerToMapTestCase))
+ tests.addTest(unittest.TestLoader().loadTestsFromTestCase(
+ ConstraintTriggerToSqlTestCase))
return tests
if __name__ == '__main__':
Please sign in to comment.
Something went wrong with that request. Please try again.