diff --git a/sqlalchemy_utils/__init__.py b/sqlalchemy_utils/__init__.py index 63b20ee..9474e38 100644 --- a/sqlalchemy_utils/__init__.py +++ b/sqlalchemy_utils/__init__.py @@ -19,6 +19,7 @@ from .functions import ( get_referencing_foreign_keys, get_tables, group_foreign_keys, + has_index, identity, merge_references, mock_engine, @@ -93,6 +94,7 @@ __all__ = ( get_referencing_foreign_keys, get_tables, group_foreign_keys, + has_index, identity, instrumented_list, merge_references, diff --git a/sqlalchemy_utils/functions/__init__.py b/sqlalchemy_utils/functions/__init__.py index d8e2ef9..93533d5 100644 --- a/sqlalchemy_utils/functions/__init__.py +++ b/sqlalchemy_utils/functions/__init__.py @@ -3,10 +3,11 @@ from .mock import create_mock_engine, mock_engine from .render import render_expression, render_statement from .sort_query import sort_query, QuerySorterException from .database import ( - database_exists, create_database, + database_exists, drop_database, escape_like, + has_index, is_auto_assigned_date_column, ) from .foreign_keys import ( diff --git a/sqlalchemy_utils/functions/database.py b/sqlalchemy_utils/functions/database.py index 5986cc3..780f607 100644 --- a/sqlalchemy_utils/functions/database.py +++ b/sqlalchemy_utils/functions/database.py @@ -1,6 +1,5 @@ from sqlalchemy.engine.url import make_url import sqlalchemy as sa -from sqlalchemy.schema import MetaData, Table, ForeignKeyConstraint from sqlalchemy.exc import ProgrammingError, OperationalError import os from copy import copy @@ -31,6 +30,69 @@ def escape_like(string, escape_char='*'): ) +def has_index(column): + """ + Return whether or not given column has an index. A column has an index if + it has a single column index or it is the first column in compound column + index. + + :param column: SQLAlchemy Column object + + .. versionadded: 0.26.2 + + :: + + from sqlalchemy_utils import has_index + + + class Article(Base): + __tablename__ = 'article_translation' + id = sa.Column(sa.Integer, primary_key=True) + title = sa.Column(sa.String(100)) + is_published = sa.Column(sa.Boolean) + is_deleted = sa.Column(sa.Boolean) + is_archived = sa.Column(sa.Boolean) + + __table_args__ = ( + ) + + + table = Article.__table__ + + has_index(table.c.is_published) # True + has_index(table.c.is_deleted) # True + has_index(table.c.is_archived) # False + + + Also supports primary key indexes + + :: + + from sqlalchemy_utils import has_index + + + 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)) + + + table = ArticleTranslation.__table__ + + has_index(table.c.locale) # False + has_index(table.c.id) # True + """ + return ( + column is column.table.primary_key.columns.values()[0] + or + any( + index.columns.values()[0] is column + for index in column.table.indexes + ) + ) + + def is_auto_assigned_date_column(column): """ Returns whether or not given SQLAlchemy Column object's is auto assigned diff --git a/sqlalchemy_utils/functions/orm.py b/sqlalchemy_utils/functions/orm.py index d16e008..99ef536 100644 --- a/sqlalchemy_utils/functions/orm.py +++ b/sqlalchemy_utils/functions/orm.py @@ -3,7 +3,6 @@ try: except ImportError: from ordereddict import OrderedDict from functools import partial -from itertools import groupby from inspect import isclass from operator import attrgetter import sqlalchemy as sa @@ -16,7 +15,6 @@ from sqlalchemy.orm.mapper import Mapper from sqlalchemy.orm.query import _ColumnEntity from sqlalchemy.orm.session import object_session from sqlalchemy.orm.util import AliasedInsp -from ..query_chain import QueryChain def get_mapper(mixed): diff --git a/tests/functions/test_has_index.py b/tests/functions/test_has_index.py new file mode 100644 index 0000000..43f0474 --- /dev/null +++ b/tests/functions/test_has_index.py @@ -0,0 +1,35 @@ +import sqlalchemy as sa +from sqlalchemy.ext.declarative import declarative_base + +from sqlalchemy_utils import has_index + + +class TestHasIndex(object): + def setup_method(self, method): + Base = declarative_base() + + 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) + is_archived = sa.Column(sa.Boolean) + + __table_args__ = ( + sa.Index('my_index', is_deleted, is_archived), + ) + + self.table = ArticleTranslation.__table__ + + def test_compound_primary_key(self): + assert has_index(self.table.c.id) + assert not has_index(self.table.c.locale) + + def test_single_column_index(self): + assert has_index(self.table.c.is_published) + + def test_compound_column_index(self): + assert has_index(self.table.c.is_deleted) + assert not has_index(self.table.c.is_archived)