Merge "Support foreign key option reflection for Oracle"

3 files changed, 45 insertions, 28 deletions

diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 96ba6c7cf..3970a181c 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -1543,34 +1543,37 @@ class OracleDialect(default.DefaultDialect):
 
         params = {'table_name': table_name}
 
-        text = \
-            "SELECT"\
-            "\nac.constraint_name,"\
-            "\nac.constraint_type,"\
-            "\nloc.column_name AS local_column,"\
-            "\nrem.table_name AS remote_table,"\
-            "\nrem.column_name AS remote_column,"\
-            "\nrem.owner AS remote_owner,"\
-            "\nloc.position as loc_pos,"\
-            "\nrem.position as rem_pos,"\
-            "\nac.search_condition"\
-            "\nFROM all_constraints%(dblink)s ac,"\
-            "\nall_cons_columns%(dblink)s loc,"\
-            "\nall_cons_columns%(dblink)s rem"\
-            "\nWHERE ac.table_name = :table_name"\
+        text = (
+            "SELECT"
+            "\nac.constraint_name,"  # 0
+            "\nac.constraint_type,"  # 1
+            "\nloc.column_name AS local_column,"  # 2
+            "\nrem.table_name AS remote_table,"  # 3
+            "\nrem.column_name AS remote_column,"  # 4
+            "\nrem.owner AS remote_owner,"  # 5
+            "\nloc.position as loc_pos,"  # 6
+            "\nrem.position as rem_pos,"  # 7
+            "\nac.search_condition,"  # 8
+            "\nac.delete_rule"  # 9
+            "\nFROM all_constraints%(dblink)s ac,"
+            "\nall_cons_columns%(dblink)s loc,"
+            "\nall_cons_columns%(dblink)s rem"
+            "\nWHERE ac.table_name = :table_name"
             "\nAND ac.constraint_type IN ('R','P', 'U', 'C')"
+        )
 
         if schema is not None:
             params['owner'] = schema
             text += "\nAND ac.owner = :owner"
 
-        text += \
-            "\nAND ac.owner = loc.owner"\
-            "\nAND ac.constraint_name = loc.constraint_name"\
-            "\nAND ac.r_owner = rem.owner(+)"\
-            "\nAND ac.r_constraint_name = rem.constraint_name(+)"\
-            "\nAND (rem.position IS NULL or loc.position=rem.position)"\
+        text += (
+            "\nAND ac.owner = loc.owner"
+            "\nAND ac.constraint_name = loc.constraint_name"
+            "\nAND ac.r_owner = rem.owner(+)"
+            "\nAND ac.r_constraint_name = rem.constraint_name(+)"
+            "\nAND (rem.position IS NULL or loc.position=rem.position)"
             "\nORDER BY ac.constraint_name, loc.position"
+        )
 
         text = text % {'dblink': dblink}
         rp = connection.execute(sql.text(text), **params)
@@ -1613,7 +1616,6 @@ class OracleDialect(default.DefaultDialect):
             dblink
 
         """
-
         requested_schema = schema  # to check later on
         resolve_synonyms = kw.get('oracle_resolve_synonyms', False)
         dblink = kw.get('dblink', '')
@@ -1634,7 +1636,8 @@ class OracleDialect(default.DefaultDialect):
                 'constrained_columns': [],
                 'referred_schema': None,
                 'referred_table': None,
-                'referred_columns': []
+                'referred_columns': [],
+                'options': {},
             }
 
         fkeys = util.defaultdict(fkey_rec)
@@ -1680,6 +1683,9 @@ class OracleDialect(default.DefaultDialect):
                        self.denormalize_name(remote_owner) != schema:
                         rec['referred_schema'] = remote_owner
 
+                    if row[9] != 'NO ACTION':
+                        rec['options']['ondelete'] = row[9]
+
                 local_cols.append(local_column)
                 remote_cols.append(remote_column)
 
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index 1990eb9d5..b509c94d6 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -409,7 +409,11 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
-    def foreign_key_constraint_option_reflection(self):
+    def foreign_key_constraint_option_reflection_ondelete(self):
+        return exclusions.closed()
+
+    @property
+    def foreign_key_constraint_option_reflection_onupdate(self):
         return exclusions.closed()
 
     @property
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 0c391fad0..1275cac03 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -540,9 +540,16 @@ class ComponentReflectionTest(fixtures.TablesTest):
     def test_get_foreign_keys_with_schema(self):
         self._test_get_foreign_keys(schema=testing.config.test_schema)
 
-    @testing.requires.foreign_key_constraint_option_reflection
+    @testing.requires.foreign_key_constraint_option_reflection_ondelete
+    def test_get_foreign_key_options_ondelete(self):
+        self._test_get_foreign_key_options(ondelete="CASCADE")
+
+    @testing.requires.foreign_key_constraint_option_reflection_onupdate
+    def test_get_foreign_key_options_onupdate(self):
+        self._test_get_foreign_key_options(onupdate="SET NULL")
+
     @testing.provide_metadata
-    def test_get_foreign_key_options(self):
+    def _test_get_foreign_key_options(self, **options):
         meta = self.metadata
 
         Table(
@@ -564,7 +571,7 @@ class ComponentReflectionTest(fixtures.TablesTest):
               sa.ForeignKeyConstraint(
                   ['tid'], ['table.id'],
                   name='myfk',
-                  onupdate="SET NULL", ondelete="CASCADE"),
+                  **options),
               test_needs_fk=True)
 
         meta.create_all()
@@ -589,7 +596,7 @@ class ComponentReflectionTest(fixtures.TablesTest):
                 (k, opts[k])
                 for k in opts if opts[k]
             ),
-            {'onupdate': 'SET NULL', 'ondelete': 'CASCADE'}
+            options
         )
 
     def _assert_insp_indexes(self, indexes, expected_indexes):