repair schema_translate_map for schema type use cases

Fixed issue where the PostgreSQL ``ENUM`` datatype as embedded in the
``ARRAY`` datatype would fail to emit correctly in create/drop when the
``schema_translate_map`` feature were also in use. Additionally repairs a
related issue where the same ``schema_translate_map`` feature would not
work for the ``ENUM`` datatype in combination with a ``CAST``, that's also
intrinsic to how the ``ARRAY(ENUM)`` combination works on the PostgreSQL
dialect.

Fixes: #6739
Change-Id: I44b1ad4db4af3acbf639aa422c46c22dd3b0d3a6
This commit is contained in:
Mike Bayer
2021-07-11 19:23:40 -04:00
parent ca52e87268
commit 520b808db2
4 changed files with 62 additions and 3 deletions
+12
View File
@@ -0,0 +1,12 @@
.. change::
:tags: bug, postgresql
:tickets: 6739
Fixed issue where the PostgreSQL ``ENUM`` datatype as embedded in the
``ARRAY`` datatype would fail to emit correctly in create/drop when the
``schema_translate_map`` feature were also in use. Additionally repairs a
related issue where the same ``schema_translate_map`` feature would not
work for the ``ENUM`` datatype in combination with a ``CAST``, that's also
intrinsic to how the ``ARRAY(ENUM)`` combination works on the PostgreSQL
dialect.
+1 -3
View File
@@ -2820,7 +2820,6 @@ class PGTypeCompiler(compiler.GenericTypeCompiler):
def visit_ENUM(self, type_, identifier_preparer=None, **kw):
if identifier_preparer is None:
identifier_preparer = self.dialect.identifier_preparer
return identifier_preparer.format_type(type_)
def visit_TIMESTAMP(self, type_, **kw):
@@ -2867,8 +2866,7 @@ class PGTypeCompiler(compiler.GenericTypeCompiler):
def visit_ARRAY(self, type_, **kw):
# TODO: pass **kw?
inner = self.process(type_.item_type)
inner = self.process(type_.item_type, **kw)
return re.sub(
r"((?: COLLATE.*)?)$",
(
+1
View File
@@ -1497,6 +1497,7 @@ class SQLCompiler(Compiled):
def visit_typeclause(self, typeclause, **kw):
kw["type_expression"] = typeclause
kw["identifier_preparer"] = self.preparer
return self.dialect.type_compiler.process(typeclause.type, **kw)
def post_process_text(self, text):
+48
View File
@@ -249,6 +249,33 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
"'y', 'z')))",
)
def test_cast_enum_schema(self):
"""test #6739"""
e1 = Enum("x", "y", "z", name="somename")
e2 = Enum("x", "y", "z", name="somename", schema="someschema")
stmt = select(cast(column("foo"), e1), cast(column("bar"), e2))
self.assert_compile(
stmt,
"SELECT CAST(foo AS somename) AS foo, "
"CAST(bar AS someschema.somename) AS bar",
)
def test_cast_enum_schema_translate(self):
"""test #6739"""
e1 = Enum("x", "y", "z", name="somename")
e2 = Enum("x", "y", "z", name="somename", schema="someschema")
schema_translate_map = {None: "bat", "someschema": "hoho"}
stmt = select(cast(column("foo"), e1), cast(column("bar"), e2))
self.assert_compile(
stmt,
"SELECT CAST(foo AS bat.somename) AS foo, "
"CAST(bar AS hoho.somename) AS bar",
schema_translate_map=schema_translate_map,
render_schema_translate=True,
)
def test_create_type_schema_translate(self):
e1 = Enum("x", "y", "z", name="somename")
e2 = Enum("x", "y", "z", name="somename", schema="someschema")
@@ -285,6 +312,27 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
render_schema_translate=True,
)
def test_create_table_array_embedded_schema_type_schema_translate(self):
"""test #6739"""
e1 = Enum("x", "y", "z", name="somename")
e2 = Enum("x", "y", "z", name="somename", schema="someschema")
schema_translate_map = {None: "foo", "someschema": "bar"}
table = Table(
"some_table",
MetaData(),
Column("q", PG_ARRAY(e1)),
Column("p", PG_ARRAY(e2)),
)
from sqlalchemy.schema import CreateTable
self.assert_compile(
CreateTable(table),
"CREATE TABLE foo.some_table (q foo.somename[], p bar.somename[])",
schema_translate_map=schema_translate_map,
render_schema_translate=True,
)
def test_create_table_with_tablespace(self):
m = MetaData()
tbl = Table(