Support foreign key option reflection for Oracle

The ON DELETE options for foreign keys are now part of
Oracle reflection.  Oracle does not support ON UPDATE
cascades.  Pull request courtesy Miroslav Shubernetskiy.

Change-Id: I135cd6cd3436354a86b2c1e1437c3785c38eed26
Pull-request: https://github.com/zzzeek/sqlalchemy/pull/418

diff --git a/doc/build/changelog/unreleased_12/ora_ondelete_reflect.rst b/doc/build/changelog/unreleased_12/ora_ondelete_reflect.rst
new file mode 100644 (file)
index 0000000..6784fe5
--- /dev/null
+++ b/doc/build/changelog/unreleased_12/ora_ondelete_reflect.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: feature, oracle
+
+    The ON DELETE options for foreign keys are now part of
+    Oracle reflection.  Oracle does not support ON UPDATE
+    cascades.  Pull request courtesy Miroslav Shubernetskiy.
+
+

diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index 96ba6c7cfe93e0f96528dcf7965257c84babb4c7..3970a181c3168d4d616a18436ea54cfde3ab391b 100644 (file)
--- 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 1990eb9d5b9f1d765c09e06160dc5a0999ebd056..b509c94d6100e85f45f9cbd7fa025d6538112de8 100644 (file)
--- 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 0c391fad01e3fea459bd6bb0c74434005d061539..1275cac032e379167b8149c95cfb065bed5e0562 100644 (file)
--- 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):

diff --git a/test/requirements.py b/test/requirements.py
index d7eaec5fdbf3926a23e932a092af0fc22dec10e9..ee1787323ec8e024eefda402cce8391b42f7d5cc 100644 (file)
--- a/test/requirements.py
+++ b/test/requirements.py
@@ -110,7 +110,11 @@ class DefaultRequirements(SuiteRequirements):
         return only_on(['oracle'])
 
     @property
-    def foreign_key_constraint_option_reflection(self):
+    def foreign_key_constraint_option_reflection_ondelete(self):
+        return only_on(['postgresql', 'mysql', 'sqlite', 'oracle'])
+
+    @property
+    def foreign_key_constraint_option_reflection_onupdate(self):
         return only_on(['postgresql', 'mysql', 'sqlite'])
 
     @property