from sqlalchemy.sql.expression import _BindParamClause
from sqlalchemy.ext.compiler import compiles
-from sqlalchemy import schema, text
+from sqlalchemy import schema, text, sql
+from sqlalchemy.sql import expression
from sqlalchemy import types as sqltypes
from ..compat import string_types, text_type, with_metaclass
return compiler.render_literal_bindparam(element, **kw)
+def _textual_index_column(table, text_):
+ """a workaround for the Index construct's severe lack of flexibility"""
+ if isinstance(text_, string_types):
+ c = schema.Column(text_, sqltypes.NULLTYPE)
+ table.append_column(c)
+ return c
+ elif isinstance(text_, expression.TextClause):
+ return _textual_index_element(table, text_)
+ else:
+ raise ValueError("String or text() construct expected")
+
+
+class _textual_index_element(sql.ColumnElement):
+ """Wrap around a sqlalchemy text() construct in such a way that
+ we appear like a column-oriented SQL expression to an Index
+ construct.
+
+ The issue here is that currently the Postgresql dialect, the biggest
+ recipient of functional indexes, keys all the index expressions to
+ the corresponding column expressions when rendering CREATE INDEX,
+ so the Index we create here needs to have a .columns collection that
+ is the same length as the .expressions collection. Ultimately
+ SQLAlchemy should support text() expressions in indexes.
+
+ See https://bitbucket.org/zzzeek/sqlalchemy/issue/3174/support-text-sent-to-indexes
+
+ """
+ __visit_name__ = '_textual_idx_element'
+
+ def __init__(self, table, text):
+ self.table = table
+ self.text = text
+ self.key = text.text
+ self.fake_column = schema.Column(self.text.text, sqltypes.NULLTYPE)
+ table.append_column(self.fake_column)
+
+ def get_children(self):
+ return [self.fake_column]
+
+
+@compiles(_textual_index_element)
+def _render_textual_index_column(element, compiler, **kw):
+ return compiler.process(element.text, **kw)
+
+
def _string_compare(t1, t2):
return \
t1.length is not None and \
return sa_schema.Column(name, type_, **kw)
def _index(self, name, tablename, columns, schema=None, **kw):
- t = sa_schema.Table(tablename or 'no_table', self._metadata(),
- *[sa_schema.Column(n, NULLTYPE) for n in columns],
+ t = sa_schema.Table(
+ tablename or 'no_table', self._metadata(),
schema=schema
)
- return sa_schema.Index(name, *[t.c[n] for n in columns], **kw)
+ idx = sa_schema.Index(
+ name,
+ *[impl._textual_index_column(t, n) for n in columns],
+ **kw)
+ return idx
def _parse_table_key(self, table_key):
if '.' in table_key:
from alembic import op
op.create_index('ik_test', 't1', ['foo', 'bar'])
+ Functional indexes can be produced by using the
+ :func:`sqlalchemy.sql.expression.text` construct::
+
+ from alembic import op
+ from sqlalchemy import text
+ op.create_index('ik_test', 't1', [text('lower(foo)')])
+
+ .. versionadded:: 0.6.7 support for making use of the
+ :func:`~sqlalchemy.sql.expression.text` construct in
+ conjunction with
+ :meth:`.Operations.create_index` in
+ order to produce functional expressions within CREATE INDEX.
+
:param name: name of the index.
:param table_name: name of the owning table.
As this is a positional argument, the old name is no
longer present.
- :param columns: a list of string column names in the
- table.
+ :param columns: a list consisting of string column names and/or
+ :func:`~sqlalchemy.sql.expression.text` constructs.
:param schema: Optional schema name to operate within.
.. versionadded:: 0.4.0
==========
Changelog
==========
+.. changelog::
+ :version: 0.6.7
+
+ .. change::
+ :tags: enhancement
+ :tickets: 222
+
+ Added support for functional indexes when using the
+ :meth:`.Operations.create_index` directive. Within the list of columns,
+ the SQLAlchemy ``text()`` construct can be sent, embedding a literal
+ SQL expression; the :meth:`.Operations.create_index` will perform some hackery
+ behind the scenes to get the :class:`.Index` construct to cooperate.
+ This works around some current limitations in :class:`.Index`
+ which should be resolved on the SQLAlchemy side at some point.
+
.. changelog::
:version: 0.6.6
:released: August 7, 2014
op.rename_table('t1', 't2', schema="foo")
context.assert_("ALTER TABLE foo.t1 RENAME TO t2")
+
+def test_create_index_no_expr_allowed():
+ op_fixture()
+ assert_raises_message(
+ ValueError,
+ "String or text\(\) construct expected",
+ op.create_index, 'name', 'tname', [func.foo(column('x'))]
+ )
+
+
+def test_create_index_quoting():
+ context = op_fixture("postgresql")
+ op.create_index(
+ 'geocoded',
+ 'locations',
+ ["IShouldBeQuoted"])
+ context.assert_(
+ 'CREATE INDEX geocoded ON locations ("IShouldBeQuoted")')
+
+
+def test_create_index_expressions():
+ context = op_fixture()
+ op.create_index(
+ 'geocoded',
+ 'locations',
+ [text('lower(coordinates)')])
+ context.assert_(
+ "CREATE INDEX geocoded ON locations (lower(coordinates))")
+
+
+def test_create_index_postgresql_expressions():
+ context = op_fixture("postgresql")
+ op.create_index(
+ 'geocoded',
+ 'locations',
+ [text('lower(coordinates)')],
+ postgresql_where=text("locations.coordinates != Null"))
+ context.assert_(
+ "CREATE INDEX geocoded ON locations (lower(coordinates)) "
+ "WHERE locations.coordinates != Null")
+
def test_create_index_postgresql_where():
context = op_fixture("postgresql")
op.create_index(
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
