lowercase collation names such as ``"utf8"`` are no longer unconditionally
quoted in generated DDL, as this quoting was never necessary for such names.
+Additionally, reflection of columns and :class:`_postgresql.DOMAIN` objects
+now populates :paramref:`.String.collation_schema` /
+:paramref:`_postgresql.DOMAIN.collation_schema` when the underlying
+collation is schema-qualified, so that reflected DDL round-trips exactly.
+The schema is omitted from the reflected value when the collation is
+visible on the current ``search_path`` without qualification.
+
.. seealso::
:ref:`postgresql_collation`
:ticket:`9693`
+:ticket:`6511`
+
Microsoft SQL Server
====================
--- /dev/null
+.. change::
+ :tags: usecase, postgresql
+ :tickets: 6511
+
+ The PostgreSQL dialect now reflects the schema of a schema-qualified
+ column or :class:`_postgresql.DOMAIN` collation, populating the new
+ :paramref:`.String.collation_schema` /
+ :paramref:`_postgresql.DOMAIN.collation_schema` parameters so that
+ reflected DDL round-trips exactly. The schema is omitted from the
+ reflected value when the collation is visible on the current
+ ``search_path`` without qualification.
+
+ .. seealso::
+
+ :ref:`postgresql_collation`
"""
collation: Optional[str]
"""The collation for the domain."""
+ collation_schema: Optional[str]
+ """The schema of the collation for the domain, if the collation is not
+ visible on the current search_path."""
class ReflectedEnum(ReflectedNamedType):
else_=sql.null(),
).label("default")
- # get the name of the collate when it's different from the default one
+ # get the name and schema of the collate when it's different from
+ # the default one; the schema is only included when the collation
+ # is not visible on the current search_path
collate = sql.case(
(
sql.and_(
.scalar_subquery()
!= pg_catalog.pg_attribute.c.attcollation,
),
- select(pg_catalog.pg_collation.c.collname)
+ select(
+ sql.func.json_build_object(
+ "name",
+ pg_catalog.pg_collation.c.collname,
+ "schema",
+ sql.case(
+ (
+ pg_catalog.pg_collation_is_visible(
+ pg_catalog.pg_collation.c.oid
+ ),
+ sql.null(),
+ ),
+ else_=pg_catalog.pg_namespace.c.nspname,
+ ),
+ type_=sqltypes.JSON(),
+ )
+ )
+ .select_from(pg_catalog.pg_collation)
+ .join(
+ pg_catalog.pg_namespace,
+ pg_catalog.pg_namespace.c.oid
+ == pg_catalog.pg_collation.c.collnamespace,
+ )
.where(
pg_catalog.pg_collation.c.oid
== pg_catalog.pg_attribute.c.attcollation
named_type_loader: _NamedTypeLoader,
type_description: str,
collation: Optional[str],
+ collation_schema: Optional[str] = None,
) -> sqltypes.TypeEngine[Any]:
"""
Attempts to reconstruct a column type defined in ischema_names based
named_type_loader,
type_description="DOMAIN '%s'" % domain["name"],
collation=domain["collation"],
+ collation_schema=domain["collation_schema"],
)
args = (domain["name"], data_type)
kwargs["collation"] = domain["collation"]
+ kwargs["collation_schema"] = domain["collation_schema"]
kwargs["default"] = domain["default"]
kwargs["not_null"] = not domain["nullable"]
kwargs["create_type"] = False
if collation is not None:
kwargs["collation"] = collation
+ if collation_schema is not None:
+ kwargs["collation_schema"] = collation_schema
data_type = schema_type(*args, **kwargs)
if array_dim >= 1:
continue
table_cols = columns[(schema, row_dict["table_name"])]
- collation = row_dict["collation"]
+ collation_info = row_dict["collation"]
+ if collation_info is not None:
+ collation = collation_info["name"]
+ collation_schema = collation_info["schema"]
+ else:
+ collation = collation_schema = None
coltype = self._reflect_type(
row_dict["format_type"],
named_type_loader,
type_description="column '%s'" % row_dict["name"],
collation=collation,
+ collation_schema=collation_schema,
)
default = row_dict["default"]
.subquery("domain_constraints")
)
+ collation_namespace = pg_catalog.pg_namespace.alias(
+ "collation_namespace"
+ )
+
query = (
select(
pg_catalog.pg_type.c.typname.label("name"),
con_sq.c.condefs,
con_sq.c.connames,
pg_catalog.pg_collation.c.collname,
+ sql.case(
+ (
+ pg_catalog.pg_collation.c.oid.is_(None),
+ sql.null(),
+ ),
+ (
+ pg_catalog.pg_collation_is_visible(
+ pg_catalog.pg_collation.c.oid
+ ),
+ sql.null(),
+ ),
+ else_=collation_namespace.c.nspname,
+ ).label("collation_schema"),
)
.join(
pg_catalog.pg_namespace,
pg_catalog.pg_type.c.typcollation
== pg_catalog.pg_collation.c.oid,
)
+ .outerjoin(
+ collation_namespace,
+ collation_namespace.c.oid
+ == pg_catalog.pg_collation.c.collnamespace,
+ )
.outerjoin(
con_sq,
pg_catalog.pg_type.c.oid == con_sq.c.contypid,
"default": domain["default"],
"constraints": constraints,
"collation": domain["collname"],
+ "collation_schema": domain["collation_schema"],
}
domains.append(domain_rec)
quote_ident = _pg_cat.quote_ident
pg_table_is_visible = _pg_cat.pg_table_is_visible
pg_type_is_visible = _pg_cat.pg_type_is_visible
+pg_collation_is_visible = _pg_cat.pg_collation_is_visible
pg_get_viewdef = _pg_cat.pg_get_viewdef
pg_get_serial_sequence = _pg_cat.pg_get_serial_sequence
format_type = _pg_cat.format_type
yield
connection.exec_driver_sql('DROP DOMAIN "SomeSchema"."Quoted.Domain"')
+ @testing.fixture
+ def some_collation(self, connection, some_schema):
+ connection.exec_driver_sql(
+ 'CREATE COLLATION "SomeSchema"."SomeCollation" '
+ "(LOCALE = 'C.utf8')"
+ )
+ yield
+ connection.exec_driver_sql(
+ 'DROP COLLATION "SomeSchema"."SomeCollation"'
+ )
+
+ @testing.fixture
+ def schema_collation_domain(self, connection, some_collation):
+ connection.exec_driver_sql(
+ 'CREATE DOMAIN "SomeSchema".domain_with_collation AS TEXT '
+ 'COLLATE "SomeSchema"."SomeCollation"'
+ )
+ yield
+ connection.exec_driver_sql(
+ 'DROP DOMAIN "SomeSchema".domain_with_collation'
+ )
+
@testing.fixture
def int_domain(self, connection):
connection.exec_driver_sql(
int_domain,
testdomain,
testdomain_schema,
+ schema_collation_domain,
):
return {
"public": [
"default": None,
"constraints": [],
"collation": None,
+ "collation_schema": None,
},
{
"visible": True,
"default": None,
"constraints": [],
"collation": None,
+ "collation_schema": None,
},
{
"visible": True,
"default": None,
"constraints": [],
"collation": None,
+ "collation_schema": None,
},
{
"visible": True,
"default": None,
"constraints": [],
"collation": None,
+ "collation_schema": None,
},
{
"visible": True,
{"check": "VALUE <> 22", "name": "my_int_check"},
],
"collation": None,
+ "collation_schema": None,
},
{
"visible": True,
"default": None,
"constraints": [],
"collation": "default",
+ "collation_schema": None,
},
{
"visible": True,
}
],
"collation": "C",
+ "collation_schema": None,
},
{
"visible": True,
"default": "42",
"constraints": [],
"collation": None,
+ "collation_schema": None,
},
],
"test_schema": [
"default": "0",
"constraints": [],
"collation": None,
+ "collation_schema": None,
}
],
"SomeSchema": [
"default": "0",
"constraints": [],
"collation": None,
- }
+ "collation_schema": None,
+ },
+ {
+ "visible": False,
+ "name": "domain_with_collation",
+ "schema": "SomeSchema",
+ "nullable": True,
+ "type": "text",
+ "default": None,
+ "constraints": [],
+ "collation": "SomeCollation",
+ "collation_schema": "SomeSchema",
+ },
],
}
)
eq_(fkey_set_default.ondelete, "SET DEFAULT (fk_id_del_set_default)")
+ def test_column_collation_reflection_with_schema(
+ self, connection, metadata
+ ):
+ """test #6511
+
+ schema-qualified collations are a PostgreSQL-only concept; this
+ test lives here rather than in the cross-dialect reflection suite
+ since no other backend supports the feature.
+
+ """
+ connection.exec_driver_sql('CREATE SCHEMA IF NOT EXISTS "SomeSchema"')
+ connection.exec_driver_sql(
+ 'CREATE COLLATION IF NOT EXISTS "SomeSchema"."SomeCollation" '
+ "(LOCALE = 'C.utf8')"
+ )
+ Table(
+ "t",
+ metadata,
+ Column(
+ "collated",
+ String(
+ collation="SomeCollation", collation_schema="SomeSchema"
+ ),
+ ),
+ Column("not_collated", String()),
+ )
+ metadata.create_all(connection)
+
+ m2 = MetaData()
+ t2 = Table("t", m2, autoload_with=connection)
+
+ eq_(
+ (
+ t2.c.collated.type.collation,
+ t2.c.collated.type.collation_schema,
+ ),
+ ("SomeCollation", "SomeSchema"),
+ )
+ is_(t2.c.not_collated.type.collation, None)
+
+ insp = inspect(connection)
+ collated, not_collated = insp.get_columns("t")
+ eq_(
+ (
+ collated["type"].collation,
+ collated["type"].collation_schema,
+ ),
+ ("SomeCollation", "SomeSchema"),
+ )
+ is_(not_collated["type"].collation, None)
+
def test_pg_weirdchar_reflection(self, metadata, connection):
meta1 = metadata
subject = Table(
self.domains = domains
class CustomType:
- def __init__(self, arg1=None, arg2=None, collation=None):
+ def __init__(
+ self, arg1=None, arg2=None, collation=None, collation_schema=None
+ ):
self.arg1 = arg1
self.arg2 = arg2
self.collation = collation
+ self.collation_schema = collation_schema
ischema_names = None
("my_custom_type(ARG1)", ("ARG1", None)),
("my_custom_type(ARG1, ARG2)", ("ARG1", "ARG2")),
]:
+ if sch == "my_custom_type()":
+ collation = {"name": "cc", "schema": "myschema"}
+ else:
+ collation = None
row_dict = {
"name": "colname",
"table_name": "tblname",
"format_type": sch,
"default": None,
"not_null": False,
- "collation": "cc" if sch == "my_custom_type()" else None,
+ "collation": collation,
"comment": None,
"generated": "",
"identity_options": None,
eq_(column_info["type"].arg2, args[1])
if sch == "my_custom_type()":
eq_(column_info["type"].collation, "cc")
+ eq_(column_info["type"].collation_schema, "myschema")
else:
eq_(column_info["type"].collation, None)
+ eq_(column_info["type"].collation_schema, None)
def test_clslevel(self):
postgresql.PGDialect.ischema_names["my_custom_type"] = self.CustomType
}
],
"collation": "default",
+ "collation_schema": None,
}
],
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
