]> git.ipfire.org Git - thirdparty/sqlalchemy/sqlalchemy.git/commitdiff
- get_unique_constraints() pull request
authorMike Bayer <mike_mp@zzzcomputing.com>
Sun, 9 Jun 2013 21:54:11 +0000 (17:54 -0400)
committerMike Bayer <mike_mp@zzzcomputing.com>
Sun, 9 Jun 2013 21:54:11 +0000 (17:54 -0400)
- version (0.9 for now)
- changelog
- move the test into the test suite so that all dialects can take advantage of it

doc/build/changelog/changelog_09.rst
lib/sqlalchemy/engine/interfaces.py
lib/sqlalchemy/engine/reflection.py
lib/sqlalchemy/testing/requirements.py
lib/sqlalchemy/testing/suite/test_reflection.py
test/engine/test_reflection.py
test/requirements.py

index f1d0bcc926558ea4a1f3dd96aa2ad4146efbe718..91990d646fbc74951fd731abf2db8fa54dcd5a61 100644 (file)
@@ -6,6 +6,13 @@
 .. changelog::
     :version: 0.9.0
 
+    .. change::
+        :tags: feature, sql
+
+        Added support for "unique constraint" reflection, via the
+        :meth:`.Inspector.get_unique_constraints` method.
+        Thanks for Roman Podolyaka for the patch.
+
     .. change::
         :tags: feature, pool
         :tickets: 2752
index d5fe5c5e230e5f6740455c470047c3f14c563794..750aa2fcdaad1b2af9d6c66cfd91d0714d49f0a5 100644 (file)
@@ -353,6 +353,8 @@ class Dialect(object):
         \**kw
           other options passed to the dialect's get_unique_constraints() method.
 
+        .. versionadded:: 0.9.0
+
         """
 
         raise NotImplementedError()
index 1926e693a8ca4000573c01a66968eb746054fc35..e290c4429eadadf71c51f982373bb34f1289c9f9 100644 (file)
@@ -362,6 +362,8 @@ class Inspector(object):
         \**kw
           other options passed to the dialect's get_unique_constraints() method.
 
+        .. versionadded:: 0.9.0
+
         """
 
         return self.dialect.get_unique_constraints(
index e032cdd4a8e409458eedc585e843c625a4dc849c..0350cd06b3714615e5ea6b68f2de461f0670ad10 100644 (file)
@@ -264,6 +264,11 @@ class SuiteRequirements(Requirements):
     def index_reflection(self):
         return exclusions.open()
 
+    @property
+    def unique_constraint_reflection(self):
+        """target dialect supports reflection of unique constraints"""
+        return exclusions.open()
+
     @property
     def unbounded_varchar(self):
         """Target database must support VARCHAR with no length"""
index ff76bd2cd4980665722d5ecfaec05355a4d545e8..858ca8a77eb2285780aa30ef8787fc77762a503a 100644 (file)
@@ -11,7 +11,7 @@ from sqlalchemy.testing.schema import Table, Column
 from sqlalchemy.testing import eq_, assert_raises_message
 from sqlalchemy import testing
 from .. import config
-
+import operator
 from sqlalchemy.schema import DDL, Index
 from sqlalchemy import event
 
@@ -356,6 +356,49 @@ class ComponentReflectionTest(fixtures.TablesTest):
     def test_get_indexes_with_schema(self):
         self._test_get_indexes(schema='test_schema')
 
+
+    @testing.requires.unique_constraint_reflection
+    def test_get_unique_constraints(self):
+        self._test_get_unique_constraints()
+
+    @testing.requires.unique_constraint_reflection
+    @testing.requires.schemas
+    def test_get_unique_constraints_with_schema(self):
+        self._test_get_unique_constraints(schema='test_schema')
+
+    @testing.provide_metadata
+    def _test_get_unique_constraints(self, schema=None):
+        uniques = sorted(
+            [
+                {'name': 'unique_a_b_c', 'column_names': ['a', 'b', 'c']},
+                {'name': 'unique_a_c', 'column_names': ['a', 'c']},
+                {'name': 'unique_b_c', 'column_names': ['b', 'c']},
+            ],
+            key=operator.itemgetter('name')
+        )
+        orig_meta = self.metadata
+        table = Table(
+            'testtbl', orig_meta,
+            Column('a', sa.String(20)),
+            Column('b', sa.String(30)),
+            Column('c', sa.Integer),
+            schema=schema
+        )
+        for uc in uniques:
+            table.append_constraint(
+                sa.UniqueConstraint(*uc['column_names'], name=uc['name'])
+            )
+        orig_meta.create_all()
+
+        inspector = inspect(orig_meta.bind)
+        reflected = sorted(
+            inspector.get_unique_constraints('testtbl', schema=schema),
+            key=operator.itemgetter('name')
+        )
+
+        eq_(uniques, reflected)
+
+
     @testing.provide_metadata
     def _test_get_view_definition(self, schema=None):
         meta = self.metadata
index fd941187466fa3fbe460fc738dbd7245c59a2992..52cbc15e6229d375604bb423248ed4f485d53483 100644 (file)
@@ -880,40 +880,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert set([t2.c.name, t2.c.id]) == set(r2.columns)
         assert set([t2.c.name]) == set(r3.columns)
 
-    @testing.provide_metadata
-    def test_unique_constraints_reflection(self):
-        uniques = sorted(
-            [
-                {'name': 'unique_a_b_c', 'column_names': ['a', 'b', 'c']},
-                {'name': 'unique_a_c', 'column_names': ['a', 'c']},
-                {'name': 'unique_b_c', 'column_names': ['b', 'c']},
-            ],
-            key=operator.itemgetter('name')
-        )
-
-        try:
-            orig_meta = sa.MetaData(bind=testing.db)
-            table = Table(
-                'testtbl', orig_meta,
-                Column('a', sa.String(20)),
-                Column('b', sa.String(30)),
-                Column('c', sa.Integer),
-            )
-            for uc in uniques:
-                table.append_constraint(
-                    sa.UniqueConstraint(*uc['column_names'], name=uc['name'])
-                )
-            orig_meta.create_all()
-
-            inspector = inspect(testing.db)
-            reflected = sorted(
-                inspector.get_unique_constraints('testtbl'),
-                key=operator.itemgetter('name')
-            )
-
-            assert uniques == reflected
-        finally:
-            testing.db.execute('drop table if exists testtbl;')
 
     @testing.requires.views
     @testing.provide_metadata
index 78544a3968077a9da492bd599a008e1c8e141431..973ad9a10edb77bcaa0e8c7c5cfd5d4bd9c8b166 100644 (file)
@@ -248,6 +248,14 @@ class DefaultRequirements(SuiteRequirements):
                     "postgresql"
                 ])
 
+    @property
+    def unique_constraint_reflection(self):
+        return fails_on_everything_except(
+                    "postgresql",
+                    "mysql",
+                    "sqlite"
+                )
+
     @property
     def update_nowait(self):
         """Target database must support SELECT...FOR UPDATE NOWAIT"""