Apply dialect_options copy fix

Fixes: #5276
Change-Id: Ic608310d4a85934fc9fa4d72daef66323c6e2525
(cherry picked from commit 5d7d96b53e)
This commit is contained in:
Gord Thompson
2020-06-06 10:04:34 -06:00
parent 7a0f3d7cd1
commit 519b2b6367
3 changed files with 115 additions and 3 deletions
+7
View File
@@ -0,0 +1,7 @@
.. change::
:tags: bug, schema
:tickets: 5276
Fixed issue where ``dialect_options`` were omitted when a
database object (e.g., :class:`.Table`) was copied using
:func:`.tometadata`.
+29 -3
View File
@@ -1624,6 +1624,18 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
c.copy(**kw) for c in self.constraints if not c._type_bound
] + [c.copy(**kw) for c in self.foreign_keys if not c.constraint]
# ticket #5276
column_kwargs = {}
for dialect_name in self.dialect_options:
dialect_options = self.dialect_options[dialect_name]._non_defaults
for (
dialect_option_key,
dialect_option_value,
) in dialect_options.items():
column_kwargs[
dialect_name + "_" + dialect_option_key
] = dialect_option_value
server_default = self.server_default
server_onupdate = self.server_onupdate
if isinstance(server_default, Computed):
@@ -1642,7 +1654,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
nullable=self.nullable,
unique=self.unique,
system=self.system,
# quote=self.quote,
# quote=self.quote, # disabled 2013-08-27 (commit 031ef080)
index=self.index,
autoincrement=self.autoincrement,
default=self.default,
@@ -1651,7 +1663,8 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause):
server_onupdate=server_onupdate,
doc=self.doc,
comment=self.comment,
*args
*args,
**column_kwargs
)
return self._schema_item_copy(c)
@@ -3055,11 +3068,24 @@ class ColumnCollectionConstraint(ColumnCollectionMixin, Constraint):
return x in self.columns
def copy(self, **kw):
# ticket #5276
constraint_kwargs = {}
for dialect_name in self.dialect_options:
dialect_options = self.dialect_options[dialect_name]._non_defaults
for (
dialect_option_key,
dialect_option_value,
) in dialect_options.items():
constraint_kwargs[
dialect_name + "_" + dialect_option_key
] = dialect_option_value
c = self.__class__(
name=self.name,
deferrable=self.deferrable,
initially=self.initially,
*self.columns.keys()
*self.columns.keys(),
**constraint_kwargs
)
return self._schema_item_copy(c)
+79
View File
@@ -5127,3 +5127,82 @@ class NamingConventionTest(fixtures.TestBase, AssertsCompiledSQL):
self.assert_compile(
CreateIndex(ix), "CREATE INDEX ix_t_q ON t (q + 5)"
)
class CopyDialectOptionsTest(fixtures.TestBase):
@contextmanager
def _fixture(self):
from sqlalchemy.engine.default import DefaultDialect
class CopyDialectOptionsTestDialect(DefaultDialect):
construct_arguments = [
(Table, {"some_table_arg": None}),
(Column, {"some_column_arg": None}),
(Index, {"some_index_arg": None}),
(PrimaryKeyConstraint, {"some_pk_arg": None}),
(UniqueConstraint, {"some_uq_arg": None}),
]
def load(dialect_name):
if dialect_name == "copydialectoptionstest":
return CopyDialectOptionsTestDialect
else:
raise exc.NoSuchModuleError("no dialect %r" % dialect_name)
with mock.patch("sqlalchemy.dialects.registry.load", load):
yield
@classmethod
def check_dialect_options_(cls, t):
eq_(
t.dialect_kwargs["copydialectoptionstest_some_table_arg"], "a1",
)
eq_(
t.c.foo.dialect_kwargs["copydialectoptionstest_some_column_arg"],
"a2",
)
eq_(
t.primary_key.dialect_kwargs["copydialectoptionstest_some_pk_arg"],
"a3",
)
eq_(
list(t.indexes)[0].dialect_kwargs[
"copydialectoptionstest_some_index_arg"
],
"a4",
)
eq_(
list(c for c in t.constraints if isinstance(c, UniqueConstraint))[
0
].dialect_kwargs["copydialectoptionstest_some_uq_arg"],
"a5",
)
def test_dialect_options_are_copied(self):
with self._fixture():
t1 = Table(
"t",
MetaData(),
Column(
"foo",
Integer,
copydialectoptionstest_some_column_arg="a2",
),
Column("bar", Integer),
PrimaryKeyConstraint(
"foo", copydialectoptionstest_some_pk_arg="a3"
),
UniqueConstraint(
"bar", copydialectoptionstest_some_uq_arg="a5"
),
copydialectoptionstest_some_table_arg="a1",
)
Index(
"idx", t1.c.foo, copydialectoptionstest_some_index_arg="a4",
)
self.check_dialect_options_(t1)
m2 = MetaData()
t2 = t1.tometadata(m2) # make a copy
self.check_dialect_options_(t2)