--- /dev/null
+.. change::
+ :tags: bug, postgresql
+ :tickets: 5850
+
+ Fixed issue where using :meth:`_schema.Table.to_metadata` (called
+ :meth:`_schema.Table.tometadata` in 1.3) in conjunction with a PostgreSQL
+ :class:`_postgresql.ExcludeConstraint` that made use of ad-hoc column
+ expressions would fail to copy correctly.
\ No newline at end of file
from ...sql import elements
from ...sql import expression
from ...sql import functions
+from ...sql import schema
from ...sql.schema import ColumnCollectionConstraint
self.ops = kw.get("ops", {})
- def copy(self, **kw):
- elements = [(col, self.operators[col]) for col in self.columns.keys()]
+ def copy(self, target_table=None, **kw):
+ elements = [
+ (
+ schema._copy_expression(expr, self.parent, target_table),
+ self.operators[expr.name],
+ )
+ for expr in self.columns
+ ]
c = self.__class__(
*elements,
name=self.name,
from ...engine import reflection
from ...sql import ColumnElement
from ...sql import compiler
+from ...sql import schema
from ...types import BLOB # noqa
from ...types import BOOLEAN # noqa
from ...types import CHAR # noqa
"on_conflict"
]
if on_conflict_clause is None and len(constraint.columns) == 1:
- on_conflict_clause = list(constraint)[0].dialect_options["sqlite"][
- "on_conflict_unique"
- ]
+ col1 = list(constraint)[0]
+ if isinstance(col1, schema.SchemaItem):
+ on_conflict_clause = list(constraint)[0].dialect_options[
+ "sqlite"
+ ]["on_conflict_unique"]
if on_conflict_clause is not None:
text += " ON CONFLICT " + on_conflict_clause
# this should really be in sql/util.py but we'd have to
# break an import cycle
def _copy_expression(expression, source_table, target_table):
+ if source_table is None or target_table is None:
+ return expression
+
def replace(col):
if (
isinstance(col, Column)
def __contains__(self, x):
return x in self.columns
- def copy(self, **kw):
+ def copy(self, target_table=None, **kw):
# ticket #5276
constraint_kwargs = {}
for dialect_name in self.dialect_options:
name=self.name,
deferrable=self.deferrable,
initially=self.initially,
- *self.columns.keys(),
+ *[
+ _copy_expression(expr, self.parent, target_table)
+ for expr in self.columns
+ ],
**constraint_kwargs
)
return self._schema_item_copy(c)
def copy(self, target_table=None, **kw):
if target_table is not None:
+ # note that target_table is None for the copy process of
+ # a column-bound CheckConstraint, so this path is not reached
+ # in that case.
sqltext = _copy_expression(self.sqltext, self.table, target_table)
else:
sqltext = self.sqltext
return self
def copy(self, target_table=None, **kw):
- if target_table is not None:
- sqltext = _copy_expression(self.sqltext, self.table, target_table)
- else:
- sqltext = self.sqltext
+ sqltext = _copy_expression(
+ self.sqltext,
+ self.column.table if self.column is not None else None,
+ target_table,
+ )
g = Computed(sqltext, persisted=self.persisted)
return self._schema_item_copy(g)
from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import Computed
+from sqlalchemy import Date
from sqlalchemy import delete
from sqlalchemy import Enum
from sqlalchemy import exc
dialect=postgresql.dialect(),
)
+ @testing.combinations(
+ (True, "deferred"),
+ (False, "immediate"),
+ argnames="deferrable_value, initially_value",
+ )
+ def test_copy_exclude_constraint_adhoc_columns(
+ self, deferrable_value, initially_value
+ ):
+ meta = MetaData()
+ table = Table(
+ "mytable",
+ meta,
+ Column("myid", Integer, Sequence("foo_id_seq"), primary_key=True),
+ Column("valid_from_date", Date(), nullable=True),
+ Column("valid_thru_date", Date(), nullable=True),
+ )
+ cons = ExcludeConstraint(
+ (
+ literal_column(
+ "daterange(valid_from_date, valid_thru_date, '[]')"
+ ),
+ "&&",
+ ),
+ where=column("valid_from_date") <= column("valid_thru_date"),
+ name="ex_mytable_valid_date_range",
+ deferrable=deferrable_value,
+ initially=initially_value,
+ )
+
+ table.append_constraint(cons)
+ expected = (
+ "ALTER TABLE mytable ADD CONSTRAINT ex_mytable_valid_date_range "
+ "EXCLUDE USING gist "
+ "(daterange(valid_from_date, valid_thru_date, '[]') WITH &&) "
+ "WHERE (valid_from_date <= valid_thru_date) "
+ "%s %s"
+ % (
+ "NOT DEFERRABLE" if not deferrable_value else "DEFERRABLE",
+ "INITIALLY %s" % initially_value,
+ )
+ )
+ self.assert_compile(
+ schema.AddConstraint(cons),
+ expected,
+ dialect=postgresql.dialect(),
+ )
+
+ meta2 = MetaData()
+ table2 = table.tometadata(meta2)
+ cons2 = [
+ c for c in table2.constraints if isinstance(c, ExcludeConstraint)
+ ][0]
+ self.assert_compile(
+ schema.AddConstraint(cons2),
+ expected,
+ dialect=postgresql.dialect(),
+ )
+
def test_exclude_constraint_full(self):
m = MetaData()
room = Column("room", Integer, primary_key=True)
from sqlalchemy.sql import elements
from sqlalchemy.sql import naming
from sqlalchemy.sql.elements import _NONE_NAME
+from sqlalchemy.sql.elements import literal_column
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
eq_(repr(const), exp)
-class ToMetaDataTest(fixtures.TestBase, ComparesTables):
+class ToMetaDataTest(fixtures.TestBase, AssertsCompiledSQL, ComparesTables):
@testing.requires.check_constraints
def test_copy(self):
# TODO: modernize this test
a2 = a.tometadata(m2)
assert b2.c.y.references(a2.c.x)
+ def test_column_collection_constraint_w_ad_hoc_columns(self):
+ """Test ColumnCollectionConstraint that has columns that aren't
+ part of the Table.
+
+ """
+ meta = MetaData()
+
+ uq1 = UniqueConstraint(literal_column("some_name"))
+ cc1 = CheckConstraint(literal_column("some_name") > 5)
+ table = Table(
+ "mytable",
+ meta,
+ Column("myid", Integer, primary_key=True),
+ Column("name", String(40), nullable=True),
+ uq1,
+ cc1,
+ )
+
+ self.assert_compile(
+ schema.AddConstraint(uq1),
+ "ALTER TABLE mytable ADD UNIQUE (some_name)",
+ dialect="default",
+ )
+ self.assert_compile(
+ schema.AddConstraint(cc1),
+ "ALTER TABLE mytable ADD CHECK (some_name > 5)",
+ dialect="default",
+ )
+ meta2 = MetaData()
+ table2 = table.tometadata(meta2)
+ uq2 = [
+ c for c in table2.constraints if isinstance(c, UniqueConstraint)
+ ][0]
+ cc2 = [
+ c for c in table2.constraints if isinstance(c, CheckConstraint)
+ ][0]
+ self.assert_compile(
+ schema.AddConstraint(uq2),
+ "ALTER TABLE mytable ADD UNIQUE (some_name)",
+ dialect="default",
+ )
+ self.assert_compile(
+ schema.AddConstraint(cc2),
+ "ALTER TABLE mytable ADD CHECK (some_name > 5)",
+ dialect="default",
+ )
+
def test_change_schema(self):
meta = MetaData()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
