sqlite: introspection of ON DELETE and ON UPDATEpr/244

2 files changed, 53 insertions, 4 deletions

diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index b50170759..312188f96 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -1366,14 +1366,19 @@ class SQLiteDialect(default.DefaultDialect):
             FK_PATTERN = (
                 '(?:CONSTRAINT (\w+) +)?'
                 'FOREIGN KEY *\( *(.+?) *\) +'
-                'REFERENCES +(?:(?:"(.+?)")|([a-z0-9_]+)) *\((.+?)\)'
+                'REFERENCES +(?:(?:"(.+?)")|([a-z0-9_]+)) *\((.+?)\) *'
+                '(?:ON DELETE '
+                '(SET NULL|SET DEFAULT|CASCADE|RESTRICT|NO ACTION) *)?'
+                '(?:ON UPDATE '
+                '(SET NULL|SET DEFAULT|CASCADE|RESTRICT|NO ACTION))?'
             )
 
             for match in re.finditer(FK_PATTERN, table_data, re.I):
                 (
                     constraint_name, constrained_columns,
                     referred_quoted_name, referred_name,
-                    referred_columns) = match.group(1, 2, 3, 4, 5)
+                    referred_columns, ondelete,
+                    onupdate) = match.group(1, 2, 3, 4, 5, 6, 7)
                 constrained_columns = list(
                     self._find_cols_in_sig(constrained_columns))
                 if not referred_columns:
@@ -1382,14 +1387,19 @@ class SQLiteDialect(default.DefaultDialect):
                     referred_columns = list(
                         self._find_cols_in_sig(referred_columns))
                 referred_name = referred_quoted_name or referred_name
+                options = {}
+                if ondelete:
+                    options['ondelete'] = ondelete
+                if onupdate:
+                    options['onupdate'] = onupdate
                 yield (
                     constraint_name, constrained_columns,
-                    referred_name, referred_columns)
+                    referred_name, referred_columns, options)
         fkeys = []
 
         for (
             constraint_name, constrained_columns,
-                referred_name, referred_columns) in parse_fks():
+                referred_name, referred_columns, options) in parse_fks():
             sig = fk_sig(
                 constrained_columns, referred_name, referred_columns)
             if sig not in keys_by_signature:
@@ -1403,6 +1413,8 @@ class SQLiteDialect(default.DefaultDialect):
                 continue
             key = keys_by_signature.pop(sig)
             key['name'] = constraint_name
+            if options:
+                key['options'] = options
             fkeys.append(key)
         # assume the remainders are the unnamed, inline constraints, just
         # use them as is as it's extremely difficult to parse inline
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index 33903ff89..314825a19 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -1134,6 +1134,16 @@ class ConstraintReflectionTest(fixtures.TestBase):
 
             # will contain an "autoindex"
             conn.execute("create table o (foo varchar(20) primary key)")
+            conn.execute(
+                "CREATE TABLE p (id INTEGER PRIMARY KEY, "
+                "c1 INTEGER, c2 INTEGER, c3 INTEGER, "
+                "CONSTRAINT fk1 FOREIGN KEY (c1) REFERENCES a1(id) "
+                "ON DELETE SET NULL, "
+                "CONSTRAINT fk2 FOREIGN KEY (c2) REFERENCES a1(id) "
+                "ON UPDATE CASCADE, "
+                "CONSTRAINT fk3 FOREIGN KEY (c3) REFERENCES a2(id) "
+                "ON DELETE CASCADE ON UPDATE SET NULL)"
+            )
 
     @classmethod
     def teardown_class(cls):
@@ -1266,6 +1276,33 @@ class ConstraintReflectionTest(fixtures.TestBase):
              'constrained_columns': ['q', 'p']}]
         )
 
+    def test_foreign_key_ondelete_onupdate(self):
+        inspector = Inspector(testing.db)
+        fks = inspector.get_foreign_keys('p')
+        eq_(
+            fks,
+            [
+                {
+                    'referred_table': 'a1', 'referred_columns': ['id'],
+                    'referred_schema': None, 'name': 'fk1',
+                    'constrained_columns': ['c1'],
+                    'options': {'ondelete': 'SET NULL'}
+                },
+                {
+                    'referred_table': 'a1', 'referred_columns': ['id'],
+                    'referred_schema': None, 'name': 'fk2',
+                    'constrained_columns': ['c2'],
+                    'options': {'onupdate': 'CASCADE'}
+                },
+                {
+                    'referred_table': 'a2', 'referred_columns': ['id'],
+                    'referred_schema': None, 'name': 'fk3',
+                    'constrained_columns': ['c3'],
+                    'options': {'ondelete': 'CASCADE', 'onupdate': 'SET NULL'}
+                },
+            ]
+        )
+
     def test_dont_reflect_autoindex(self):
         inspector = Inspector(testing.db)
         eq_(inspector.get_indexes('o'), [])