.. changelog::
:version: 1.0.6
+ .. change::
+ :tags: bug, postgresql
+ :tickets: 3454
+
+ Repaired the :class:`.ExcludeConstraint` construct to support common
+ features that other objects like :class:`.Index` now do, that
+ the column expression may be specified as an arbitrary SQL
+ expression such as :obj:`.cast` or :obj:`.text`.
+
.. change::
:tags: feature, postgresql
:pullreq: github:182
text += " WHERE " + where_compiled
return text
- def visit_exclude_constraint(self, constraint):
+ def visit_exclude_constraint(self, constraint, **kw):
text = ""
if constraint.name is not None:
text += "CONSTRAINT %s " % \
self.preparer.format_constraint(constraint)
elements = []
- for c in constraint.columns:
- op = constraint.operators[c.name]
- elements.append(self.preparer.quote(c.name) + ' WITH ' + op)
+ for expr, name, op in constraint._render_exprs:
+ kw['include_table'] = False
+ elements.append(
+ "%s WITH %s" % (self.sql_compiler.process(expr, **kw), op)
+ )
text += "EXCLUDE USING %s (%s)" % (constraint.using,
', '.join(elements))
if constraint.where is not None:
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php
-from sqlalchemy.schema import ColumnCollectionConstraint
-from sqlalchemy.sql import expression
+from ...sql.schema import ColumnCollectionConstraint
+from ...sql import expression
+from ... import util
class ExcludeConstraint(ColumnCollectionConstraint):
for this constraint.
"""
+ columns = []
+ render_exprs = []
+ self.operators = {}
+
+ expressions, operators = zip(*elements)
+
+ for (expr, column, strname, add_element), operator in zip(
+ self._extract_col_expression_collection(expressions),
+ operators
+ ):
+ if add_element is not None:
+ columns.append(add_element)
+
+ name = column.name if column is not None else strname
+
+ if name is not None:
+ # backwards compat
+ self.operators[name] = operator
+
+ expr = expression._literal_as_text(expr)
+
+ render_exprs.append(
+ (expr, name, operator)
+ )
+
+ self._render_exprs = render_exprs
ColumnCollectionConstraint.__init__(
self,
- *[col for col, op in elements],
+ *columns,
name=kw.get('name'),
deferrable=kw.get('deferrable'),
initially=kw.get('initially')
)
- self.operators = {}
- for col_or_string, op in elements:
- name = getattr(col_or_string, 'name', col_or_string)
- self.operators[name] = op
self.using = kw.get('using', 'gist')
where = kw.get('where')
if where:
if _autoattach and self._pending_colargs:
self._check_attach()
+ @classmethod
+ def _extract_col_expression_collection(cls, expressions):
+ for expr in expressions:
+ strname = None
+ column = None
+ if not isinstance(expr, ClauseElement):
+ # this assumes a string
+ strname = expr
+ else:
+ cols = []
+ visitors.traverse(expr, {}, {'column': cols.append})
+ if cols:
+ column = cols[0]
+ add_element = column if column is not None else strname
+ yield expr, column, strname, add_element
+
def _check_attach(self, evt=False):
col_objs = [
c for c in self._pending_colargs
self.table = None
columns = []
- for expr in expressions:
- if not isinstance(expr, ClauseElement):
- columns.append(expr)
- else:
- cols = []
- visitors.traverse(expr, {}, {'column': cols.append})
- if cols:
- columns.append(cols[0])
+ for expr, column, strname, add_element in self.\
+ _extract_col_expression_collection(expressions):
+ if add_element is not None:
+ columns.append(add_element)
self.expressions = expressions
self.name = quoted_name(name, kw.pop("quote", None))
from sqlalchemy.testing import engines, fixtures
from sqlalchemy import testing
from sqlalchemy import Sequence, Table, Column, Integer, update, String,\
- insert, func, MetaData, Enum, Index, and_, delete, select, cast, text
+ insert, func, MetaData, Enum, Index, and_, delete, select, cast, text, \
+ Text
from sqlalchemy.dialects.postgresql import ExcludeConstraint, array
from sqlalchemy import exc, schema
from sqlalchemy.dialects.postgresql import base as postgresql
tbl.append_constraint(cons_copy)
self.assert_compile(schema.AddConstraint(cons_copy),
'ALTER TABLE testtbl ADD EXCLUDE USING gist '
- '(room WITH =)',
- dialect=postgresql.dialect())
+ '(room WITH =)')
+
+ def test_exclude_constraint_text(self):
+ m = MetaData()
+ cons = ExcludeConstraint((text('room::TEXT'), '='))
+ Table(
+ 'testtbl', m,
+ Column('room', String),
+ cons)
+ self.assert_compile(
+ schema.AddConstraint(cons),
+ 'ALTER TABLE testtbl ADD EXCLUDE USING gist '
+ '(room::TEXT WITH =)')
+
+ def test_exclude_constraint_cast(self):
+ m = MetaData()
+ tbl = Table(
+ 'testtbl', m,
+ Column('room', String)
+ )
+ cons = ExcludeConstraint((cast(tbl.c.room, Text), '='))
+ tbl.append_constraint(cons)
+ self.assert_compile(
+ schema.AddConstraint(cons),
+ 'ALTER TABLE testtbl ADD EXCLUDE USING gist '
+ '(CAST(room AS TEXT) WITH =)'
+ )
+
+ def test_exclude_constraint_cast_quote(self):
+ m = MetaData()
+ tbl = Table(
+ 'testtbl', m,
+ Column('Room', String)
+ )
+ cons = ExcludeConstraint((cast(tbl.c.Room, Text), '='))
+ tbl.append_constraint(cons)
+ self.assert_compile(
+ schema.AddConstraint(cons),
+ 'ALTER TABLE testtbl ADD EXCLUDE USING gist '
+ '(CAST("Room" AS TEXT) WITH =)'
+ )
def test_substring(self):
self.assert_compile(func.substring('abc', 1, 2),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
