Add has_unique_index utility function

CHANGES.rst

@@ -8,6 +8,7 @@ Here you can see the full list of changes between each SQLAlchemy-Utils release.
 ^^^^^^^^^^^^^^^^^^^
 
 - Added support for more SQLAlchemy based objects and classes in get_tables function
+- Added has_unique_index utility function
 
 
 0.27.0 (2014-10-14)

docs/database_helpers.rst

@@ -35,6 +35,12 @@ has_index
 .. autofunction:: has_index
 
 
+has_unique_index
+^^^^^^^^^^^^^^^^
+
+.. autofunction:: has_unique_index
+
+
 render_expression
 ^^^^^^^^^^^^^^^^^
 

sqlalchemy_utils/__init__.py

@@ -26,6 +26,7 @@ from .functions import (
     has_any_changes,
     has_changes,
     has_index,
+    has_unique_index,
     identity,
     merge_references,
     mock_engine,

sqlalchemy_utils/functions/__init__.py

@@ -9,6 +9,7 @@ from .database import (
     drop_database,
     escape_like,
     has_index,
+    has_unique_index,
     is_auto_assigned_date_column,
 )
 from .foreign_keys import (

sqlalchemy_utils/functions/database.py

@@ -169,6 +169,49 @@ def has_index(column):
     )
 
 
+def has_unique_index(column):
+    """
+    Return whether or not given column has a unique index. A column has a
+    unique index if it has a single column unique index or it is a part of
+    single column UniqueConstraint.
+
+    :param column: SQLAlchemy Column object
+
+    .. versionadded: 0.27.1
+
+    ::
+
+        from sqlalchemy_utils import has_unique_index
+
+
+        class Article(Base):
+            __tablename__ = 'article'
+            id = sa.Column(sa.Integer, primary_key=True)
+            title = sa.Column(sa.String(100))
+            is_published = sa.Column(sa.Boolean, unique=True)
+            is_deleted = sa.Column(sa.Boolean)
+            is_archived = sa.Column(sa.Boolean)
+
+
+        table = Article.__table__
+
+        has_unique_index(table.c.is_published) # True
+        has_unique_index(table.c.is_deleted)   # False
+        has_unique_index(table.c.id)           # True
+    """
+    pks = column.table.primary_key.columns
+    return (
+        (column is pks.values()[0] and len(pks) == 1)
+        or
+        any(
+            constraint.columns.values()[0] is column and
+            len(constraint.columns) == 1
+            for constraint in column.table.constraints
+            if isinstance(constraint, sa.sql.schema.UniqueConstraint)
+        )
+    )
+
+
 def is_auto_assigned_date_column(column):
     """
     Returns whether or not given SQLAlchemy Column object's is auto assigned

tests/__init__.py

@@ -111,3 +111,10 @@ class TestCase(object):
         self.User = User
         self.Category = Category
         self.Article = Article
+
+
+def assert_contains(clause, query):
+    # Test that query executes
+    query.all()
+    assert clause in str(query)
+

tests/functions/test_get_primary_keys.py

@@ -6,7 +6,6 @@ import sqlalchemy as sa
 from sqlalchemy.ext.declarative import declarative_base
 
 from sqlalchemy_utils import get_primary_keys
-from tests import TestCase
 
 
 class TestGetPrimaryKeys(object):

tests/functions/test_has_unique_index.py

@@ -0,0 +1,46 @@
+import sqlalchemy as sa
+from sqlalchemy.ext.declarative import declarative_base
+
+from sqlalchemy_utils import has_unique_index
+
+
+class TestHasIndex(object):
+    def setup_method(self, method):
+        Base = declarative_base()
+
+        class Article(Base):
+            __tablename__ = 'article'
+            id = sa.Column(sa.Integer, primary_key=True)
+
+        class ArticleTranslation(Base):
+            __tablename__ = 'article_translation'
+            id = sa.Column(sa.Integer, primary_key=True)
+            locale = sa.Column(sa.String(10), primary_key=True)
+            title = sa.Column(sa.String(100))
+            is_published = sa.Column(sa.Boolean, index=True)
+            is_deleted = sa.Column(sa.Boolean, unique=True)
+            is_archived = sa.Column(sa.Boolean)
+
+            __table_args__ = (
+                sa.Index('my_index', is_archived, is_published, unique=True),
+            )
+
+        self.articles = Article.__table__
+        self.article_translations = ArticleTranslation.__table__
+
+    def test_primary_key(self):
+        assert has_unique_index(self.articles.c.id)
+
+    def test_unique_index(self):
+        assert has_unique_index(self.article_translations.c.is_deleted)
+
+    def test_compound_primary_key(self):
+        assert not has_unique_index(self.article_translations.c.id)
+        assert not has_unique_index(self.article_translations.c.locale)
+
+    def test_single_column_index(self):
+        assert not has_unique_index(self.article_translations.c.is_published)
+
+    def test_compound_column_unique_index(self):
+        assert not has_unique_index(self.article_translations.c.is_published)
+        assert not has_unique_index(self.article_translations.c.is_archived)