:ticket:`13014`
+.. _change_9693_postgresql:
+
+Schema-Qualified Collation Names
+---------------------------------
+
+PostgreSQL collations are schema-qualified catalog objects (e.g.
+``CREATE COLLATION my_schema.my_collation (...)``). Previously, there was
+no way to correctly indicate a schema-qualified collation using the
+:paramref:`.String.collation` parameter; a value such as
+``collation="my_schema.my_collation"`` would render as a single,
+incorrectly-quoted identifier.
+
+A new parameter :paramref:`.String.collation_schema` is added, used
+in conjunction with :paramref:`.String.collation`, to indicate the schema
+in which the collation is defined. The same parameter is also added to
+:class:`_postgresql.DOMAIN` as :paramref:`_postgresql.DOMAIN.collation_schema`,
+as well as to :func:`_sql.collate` and :meth:`.ColumnOperators.collate` as
+``collation_schema``::
+
+ Column(
+ "data",
+ String(collation="my_collation", collation_schema="my_schema"),
+ )
+
+The above renders DDL similar to:
+
+.. sourcecode:: sql
+
+ data VARCHAR COLLATE "my_schema"."my_collation"
+
+As part of this change, all collation-name rendering across DDL and the
+:func:`_sql.collate` construct now consistently uses the dialect's
+identifier preparer for quoting, rather than several separate, inconsistent
+hand-quoting code paths that existed previously. As a side effect, simple
+lowercase collation names such as ``"utf8"`` are no longer unconditionally
+quoted in generated DDL, as this quoting was never necessary for such names.
+
+.. seealso::
+
+ :ref:`postgresql_collation`
+
+:ticket:`9693`
+
Microsoft SQL Server
====================
--- /dev/null
+.. change::
+ :tags: usecase, postgresql
+ :tickets: 9693
+
+ Added a new parameter :paramref:`.String.collation_schema`, as well as
+ :paramref:`_postgresql.DOMAIN.collation_schema` and
+ :paramref:`.ColumnOperators.collate.collation_schema`, allowing a
+ PostgreSQL schema-qualified collation name to be specified explicitly,
+ rather than embedding the schema name within the ``collation`` string
+ itself, which previously rendered incorrectly. As part of this change,
+ collation name rendering across DDL and the :func:`_sql.collate`
+ construct now consistently uses the dialect's identifier preparer for
+ quoting, rather than several separate, inconsistent hand-quoting code
+ paths; as a side effect, simple lowercase collation names such as
+ ``"utf8"`` are no longer unconditionally quoted in generated DDL.
+
+ .. seealso::
+
+ :ref:`postgresql_collation`
.. versionadded:: 1.4.0b2
+.. _postgresql_collation:
+
+Schema-Qualified Collations
+----------------------------
+
+PostgreSQL supports collations that are qualified by a schema name, such as
+``CREATE COLLATION my_schema.my_collation (...)``. To refer to such a
+collation, use the :paramref:`.String.collation_schema` parameter (or
+:paramref:`_postgresql.DOMAIN.collation_schema` for a :class:`_postgresql.DOMAIN`)
+in conjunction with :paramref:`.String.collation`, rather than attempting to
+embed the schema name inside the ``collation`` string itself::
+
+ Column(
+ "data",
+ String(collation="my_collation", collation_schema="my_schema"),
+ )
+
+The above renders DDL similar to:
+
+.. sourcecode:: sql
+
+ data VARCHAR COLLATE "my_schema"."my_collation"
+
+.. versionadded:: 2.1
+
+.. seealso::
+
+ :paramref:`.String.collation`
+
+ :paramref:`.String.collation_schema`
""" # noqa: E501
options = []
if domain.collation is not None:
- options.append(f"COLLATE {self.preparer.quote(domain.collation)}")
+ collation = self.preparer.format_collation(
+ domain.collation, domain.collation_schema
+ )
+ options.append(f"COLLATE {collation}")
if domain.default is not None:
default = self.render_default_string(domain.default)
options.append(f"DEFAULT {default}")
from typing import TYPE_CHECKING
from typing import Union
+from ... import exc
from ... import schema
from ... import util
from ...sql import coercions
data_type: _TypeEngineArgument[Any],
*,
collation: Optional[str] = None,
+ collation_schema: Optional[str] = None,
default: Union[elements.TextClause, str, None] = None,
constraint_name: Optional[str] = None,
not_null: Optional[bool] = None,
If no collation is specified, the underlying data type's default
collation is used. The underlying type must be collatable if
``collation`` is specified.
+ :param collation_schema: Optional, the name of the schema in which
+ :paramref:`.DOMAIN.collation` is defined. Requires that
+ :paramref:`.DOMAIN.collation` is also present.
+
+ .. versionadded:: 2.1
+
:param default: The DEFAULT clause specifies a default value for
columns of the domain data type. The default should be a string
or a :func:`_expression.text` value.
self.data_type = type_api.to_instance(data_type)
self.default = default
self.collation = collation
+ if collation_schema is not None and collation is None:
+ raise exc.ArgumentError(
+ "the 'collation_schema' parameter of DOMAIN requires "
+ "the 'collation' parameter to also be present"
+ )
+ self.collation_schema = collation_schema
self.constraint_name = constraint_name
self.not_null = bool(not_null)
if check is not None:
def collate(
- expression: _ColumnExpressionArgument[str], collation: str
+ expression: _ColumnExpressionArgument[str],
+ collation: str,
+ collation_schema: Optional[str] = None,
) -> BinaryExpression[str]:
"""Return the clause ``expression COLLATE collation``.
The collation expression is also quoted if it is a case sensitive
identifier, e.g. contains uppercase characters.
+ :param expression: the column expression to apply a collation to.
+
+ :param collation: the name of the collation.
+
+ :param collation_schema: optional, the name of the schema in which the
+ collation is defined, for use with database backends that support
+ schema-qualified collations, currently PostgreSQL.
+
+ .. versionadded:: 2.1
+
"""
if isinstance(expression, operators.ColumnOperators):
- return expression.collate(collation) # type: ignore[return-value]
+ return expression.collate( # type: ignore[return-value]
+ collation, collation_schema=collation_schema
+ )
else:
return CollationClause._create_collation_expression(
- expression, collation
+ expression, collation, collation_schema
)
return schema_prefix + self.preparer.quote(tablename) + "." + name
def visit_collation(self, element, **kw):
- return self.preparer.format_collation(element.collation)
+ return self.preparer.format_collation(
+ element.collation, element.collation_schema
+ )
def visit_fromclause(self, fromclause, **kwargs):
return fromclause.name
return "NCLOB"
def _render_string_type(
- self, name: str, length: Optional[int], collation: Optional[str]
+ self,
+ name: str,
+ length: Optional[int],
+ collation: Optional[str],
+ collation_schema: Optional[str] = None,
+ identifier_preparer: Optional[IdentifierPreparer] = None,
+ **kw: Any,
) -> str:
text = name
if length:
text += f"({length})"
if collation:
- text += f' COLLATE "{collation}"'
+ if identifier_preparer is None:
+ identifier_preparer = self.dialect.identifier_preparer
+ text += " COLLATE " + identifier_preparer.format_collation(
+ collation, collation_schema
+ )
return text
def visit_CHAR(self, type_: sqltypes.CHAR, **kw: Any) -> str:
- return self._render_string_type("CHAR", type_.length, type_.collation)
+ return self._render_string_type(
+ "CHAR",
+ type_.length,
+ type_.collation,
+ type_.collation_schema,
+ **kw,
+ )
def visit_NCHAR(self, type_: sqltypes.NCHAR, **kw: Any) -> str:
- return self._render_string_type("NCHAR", type_.length, type_.collation)
+ return self._render_string_type(
+ "NCHAR",
+ type_.length,
+ type_.collation,
+ type_.collation_schema,
+ **kw,
+ )
def visit_VARCHAR(self, type_: sqltypes.String, **kw: Any) -> str:
return self._render_string_type(
- "VARCHAR", type_.length, type_.collation
+ "VARCHAR",
+ type_.length,
+ type_.collation,
+ type_.collation_schema,
+ **kw,
)
def visit_NVARCHAR(self, type_: sqltypes.NVARCHAR, **kw: Any) -> str:
return self._render_string_type(
- "NVARCHAR", type_.length, type_.collation
+ "NVARCHAR",
+ type_.length,
+ type_.collation,
+ type_.collation_schema,
+ **kw,
)
def visit_TEXT(self, type_: sqltypes.Text, **kw: Any) -> str:
- return self._render_string_type("TEXT", type_.length, type_.collation)
+ return self._render_string_type(
+ "TEXT",
+ type_.length,
+ type_.collation,
+ type_.collation_schema,
+ **kw,
+ )
def visit_UUID(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
return "UUID"
def visit_uuid(self, type_: sqltypes.Uuid[Any], **kw: Any) -> str:
if not type_.native_uuid or not self.dialect.supports_native_uuid:
- return self._render_string_type("CHAR", length=32, collation=None)
+ return self._render_string_type(
+ "CHAR",
+ length=32,
+ collation=None,
+ collation_schema=None,
+ **kw,
+ )
else:
return self.visit_UUID(type_, **kw)
else:
return ident
- def format_collation(self, collation_name):
+ def format_collation(self, collation_name, collation_schema=None):
if self.quote_case_sensitive_collations:
- return self.quote(collation_name)
+ name = self.quote(collation_name)
else:
- return collation_name
+ name = collation_name
+
+ if collation_schema is not None:
+ return f"{self.quote_schema(collation_schema)}.{name}"
+ else:
+ return name
def format_sequence(
self, sequence: schema.Sequence, use_schema: bool = True
def _collate_impl(
- expr: ColumnElement[str], op: OperatorType, collation: str, **kw: Any
+ expr: ColumnElement[str],
+ op: OperatorType,
+ collation: str,
+ collation_schema: Optional[str] = None,
+ **kw: Any,
) -> ColumnElement[str]:
- return CollationClause._create_collation_expression(expr, collation)
+ return CollationClause._create_collation_expression(
+ expr, collation, collation_schema
+ )
def _regexp_match_impl(
def nullslast(self) -> UnaryExpression[_T_co]: ...
- def collate(self, collation: str) -> CollationClause: ...
+ def collate(
+ self, collation: str, collation_schema: Optional[str] = None
+ ) -> CollationClause: ...
def between(
self, cleft: Any, cright: Any, symmetric: bool = False
__visit_name__ = "collation"
_traverse_internals: _TraverseInternalsType = [
- ("collation", InternalTraversal.dp_string)
+ ("collation", InternalTraversal.dp_string),
+ ("collation_schema", InternalTraversal.dp_string),
]
@classmethod
@util.preload_module("sqlalchemy.sql.sqltypes")
def _create_collation_expression(
- cls, expression: _ColumnExpressionArgument[str], collation: str
+ cls,
+ expression: _ColumnExpressionArgument[str],
+ collation: str,
+ collation_schema: Optional[str] = None,
) -> BinaryExpression[str]:
sqltypes = util.preloaded.sql_sqltypes
expr = coercions.expect(roles.ExpressionElementRole[str], expression)
if expr.type._type_affinity is sqltypes.String:
- collate_type = expr.type._with_collation(collation)
+ collate_type = expr.type._with_collation(
+ collation, collation_schema
+ )
else:
collate_type = expr.type
return BinaryExpression(
expr,
- CollationClause(collation),
+ CollationClause(collation, collation_schema),
operators.collate,
type_=collate_type,
)
- def __init__(self, collation):
+ def __init__(self, collation, collation_schema=None):
self.collation = collation
+ self.collation_schema = collation_schema
class _IdentifiedClause(Executable, ClauseElement):
flags=flags,
)
- def collate(self, collation: str) -> ColumnOperators:
+ def collate(
+ self, collation: str, collation_schema: Optional[str] = None
+ ) -> ColumnOperators:
"""Produce a :func:`_expression.collate` clause against
the parent object, given the collation string.
+ :param collation: the name of the collation.
+
+ :param collation_schema: optional, the name of the schema in which
+ the collation is defined, for use with database backends that
+ support schema-qualified collations, currently PostgreSQL.
+
+ .. versionadded:: 2.1
+
.. seealso::
:func:`_expression.collate`
"""
- return self.operate(collate, collation)
+ return self.operate(
+ collate, collation, collation_schema=collation_schema
+ )
def __radd__(self, other: Any) -> ColumnOperators:
"""Implement the ``+`` operator in reverse.
@_operator_fn
-def collate(a: Any, b: Any) -> Any:
- return a.collate(b)
+def collate(a: Any, b: Any, collation_schema: Optional[str] = None) -> Any:
+ return a.collate(b, collation_schema=collation_schema)
@_operator_fn
self,
length: Optional[int] = None,
collation: Optional[str] = None,
+ collation_schema: Optional[str] = None,
):
"""
Create a string-holding type.
to store non-ascii data. These datatypes will ensure that the
correct types are used on the database.
- """
+ :param collation_schema: Optional, the name of the schema in which
+ :paramref:`.String.collation` is defined, for use with database
+ backends that support schema-qualified collations. This is
+ currently known to be supported in PostgreSQL. Requires that
+ :paramref:`.String.collation` is also present. E.g.:
+
+ .. sourcecode:: pycon+sql
+
+ >>> from sqlalchemy import cast, select, String
+ >>> print(
+ ... select(
+ ... cast(
+ ... "some string",
+ ... String(
+ ... collation="my_collation",
+ ... collation_schema="my_schema",
+ ... ),
+ ... )
+ ... )
+ ... )
+ {printsql}SELECT CAST(:param_1 AS VARCHAR COLLATE "my_schema"."my_collation") AS anon_1
+
+ .. versionadded:: 2.1
+
+ """ # noqa: E501
self.length = length
self.collation = collation
+ if collation_schema is not None and collation is None:
+ raise exc.ArgumentError(
+ "the 'collation_schema' parameter of String requires "
+ "the 'collation' parameter to also be present"
+ )
+ self.collation_schema = collation_schema
- def _with_collation(self, collation):
+ def _with_collation(self, collation, collation_schema=None):
new_type = self.copy()
new_type.collation = collation
+ new_type.collation_schema = collation_schema
return new_type
def _resolve_for_literal(self, value):
length: Optional[int] = None
collation: Optional[str] = None
+ collation_schema: Optional[str] = None
@overload
def __init__(
return self
- def _with_collation(self, collation: str) -> Self:
+ def _with_collation(
+ self, collation: str, collation_schema: Optional[str] = None
+ ) -> Self:
"""set up error handling for the collate expression"""
raise NotImplementedError("this datatype does not support collation")
tt.impl = tt.impl_instance = typedesc
return tt
- def _with_collation(self, collation: str) -> Self:
+ def _with_collation(
+ self, collation: str, collation_schema: Optional[str] = None
+ ) -> Self:
tt = self._copy_with_check()
tt.impl = tt.impl_instance = self.impl_instance._with_collation(
- collation
+ collation, collation_schema
)
return tt
"CONSTRAINT no_bar NOT NULL CHECK (VALUE != 'bar')",
)
+ def test_domain_collation_schema(self):
+ """test #9693"""
+ self.assert_compile(
+ postgresql.CreateDomainType(
+ DOMAIN(
+ "foo",
+ Text,
+ collation="my-coll",
+ collation_schema="CollSchema",
+ )
+ ),
+ 'CREATE DOMAIN foo AS TEXT COLLATE "CollSchema"."my-coll"',
+ )
+
+ def test_domain_collation_schema_requires_collation(self):
+ assert_raises_message(
+ exc.ArgumentError,
+ "the 'collation_schema' parameter of DOMAIN requires "
+ "the 'collation' parameter to also be present",
+ DOMAIN,
+ "foo",
+ Text,
+ collation_schema="CollSchema",
+ )
+
def test_cast_domain_schema(self):
"""test #6739"""
d1 = DOMAIN("somename", Integer)
'VARCHAR(30)[] COLLATE "en_US"',
)
+ def test_array_type_render_str_collate_schema(self):
+ """test #9693"""
+ self.assert_compile(
+ postgresql.ARRAY(
+ Unicode(30, collation="en_US", collation_schema="pg_catalog")
+ ),
+ 'VARCHAR(30)[] COLLATE pg_catalog."en_US"',
+ )
+
def test_array_type_render_str_multidim(self):
self.assert_compile(
postgresql.ARRAY(Unicode(30), dimensions=2), "VARCHAR(30)[][]"
column("z", MyType1()) == column("x", MyType2()),
column("z", MyType1()) == column("x", MyType3("x")),
column("z", MyType1()) == column("x", MyType3("y")),
+ column("z", String(50, collation="x"))
+ == column("x", String(50, collation="x")),
+ column("z", String(50, collation="x"))
+ == column("x", String(50, collation="x", collation_schema="a")),
+ column("z", String(50, collation="x", collation_schema="a"))
+ == column("x", String(50, collation="x", collation_schema="b")),
)
]
"SELECT x ORDER BY x COLLATE bar",
)
+ # columns clause, schema-qualified collation
+ self.assert_compile(
+ select(column("x").collate("bar", collation_schema="myschema")),
+ "SELECT x COLLATE myschema.bar AS anon_1",
+ dialect=postgresql.dialect(),
+ )
+
def test_literal(self):
self.assert_compile(
select(literal("foo")), "SELECT :param_1 AS anon_1"
(
"default",
None,
+ None,
"SELECT CAST(t1.txt AS VARCHAR(10)) AS txt FROM t1",
None,
),
(
"explicit_mssql",
"Latin1_General_CI_AS",
+ None,
"SELECT CAST(t1.txt AS VARCHAR(10)) COLLATE Latin1_General_CI_AS AS txt FROM t1", # noqa
mssql.dialect(),
),
(
"explicit_mysql",
"utf8mb4_unicode_ci",
+ None,
"SELECT CAST(t1.txt AS CHAR(10)) AS txt FROM t1",
mysql.dialect(),
),
(
"explicit_postgresql",
"en_US",
+ None,
'SELECT CAST(t1.txt AS VARCHAR(10)) COLLATE "en_US" AS txt FROM t1', # noqa
postgresql.dialect(),
),
+ (
+ "explicit_postgresql_schema",
+ "en_US",
+ "myschema",
+ 'SELECT CAST(t1.txt AS VARCHAR(10)) COLLATE myschema."en_US" AS txt FROM t1', # noqa
+ postgresql.dialect(),
+ ),
(
"explicit_sqlite",
"NOCASE",
+ None,
'SELECT CAST(t1.txt AS VARCHAR(10)) COLLATE "NOCASE" AS txt FROM t1', # noqa
sqlite.dialect(),
),
- id_="iaaa",
+ id_="iaaaa",
)
- def test_cast_with_collate(self, collation_name, expected_sql, dialect):
+ def test_cast_with_collate(
+ self, collation_name, collation_schema, expected_sql, dialect
+ ):
t1 = Table(
"t1",
MetaData(),
- Column("txt", String(10, collation=collation_name)),
+ Column(
+ "txt",
+ String(
+ 10,
+ collation=collation_name,
+ collation_schema=collation_schema,
+ ),
+ ),
)
stmt = select(func.cast(t1.c.txt, t1.c.txt.type))
self.assert_compile(stmt, expected_sql, dialect=dialect)
def test_collate(self):
left = column("left")
right = "some collation"
- left.comparator.operate(operators.collate, right).compare(
+ assert left.comparator.operate(operators.collate, right).compare(
collate(left, right)
)
+ def test_collate_schema(self):
+ left = column("left")
+ right = "some collation"
+ assert left.comparator.operate(
+ operators.collate, right, collation_schema="some schema"
+ ).compare(collate(left, right, collation_schema="some schema"))
+ assert left.collate(right, collation_schema="some schema").compare(
+ collate(left, right, collation_schema="some schema")
+ )
+
def test_default_adapt(self):
class TypeOne(TypeEngine):
operator_classes = OperatorClass.ANY
dialect="mssql",
)
+ def test_collate_schema(self):
+ self.assert_compile(
+ column("foo").collate("fr_FR", collation_schema="MySchema"),
+ 'foo COLLATE "MySchema"."fr_FR"',
+ dialect="postgresql",
+ )
+
+ self.assert_compile(
+ column("foo").collate("fr_fr", collation_schema="myschema"),
+ "foo COLLATE myschema.fr_fr",
+ dialect="postgresql",
+ )
+
def test_join(self):
# Lower case names, should not quote
metadata = MetaData()
(lambda c1: c1.like("qpr"), "q LIKE :q_1->BINDCAST->[TEXT]"),
(
lambda c2: c2.like("qpr"),
- 'q LIKE :q_1->BINDCAST->[TEXT COLLATE "xyz"]',
+ "q LIKE :q_1->BINDCAST->[TEXT COLLATE xyz]",
),
(
# new behavior, a type with no collation passed into collate()
# on the right side bind-cast. previous to #11576 we'd only
# get TEXT for the bindcast.
lambda c1: collate(c1, "abc").like("qpr"),
- '(q COLLATE abc) LIKE :param_1->BINDCAST->[TEXT COLLATE "abc"]',
+ "(q COLLATE abc) LIKE :param_1->BINDCAST->[TEXT COLLATE abc]",
),
(
lambda c2: collate(c2, "abc").like("qpr"),
- '(q COLLATE abc) LIKE :param_1->BINDCAST->[TEXT COLLATE "abc"]',
+ "(q COLLATE abc) LIKE :param_1->BINDCAST->[TEXT COLLATE abc]",
),
argnames="testcase,expected",
)
)
self.assert_compile(
c2.like("qpr"),
- 'q LIKE :q_1->BINDCAST->[TEXT COLLATE "xyz"]',
+ "q LIKE :q_1->BINDCAST->[TEXT COLLATE xyz]",
dialect=renders_bind_cast,
)
String(50, collation="FOO"), 'VARCHAR(50) COLLATE "FOO"'
)
+ def test_string_collation_lowercase_unquoted(self):
+ """simple lowercase collation names no longer render with
+ unconditional quoting now that this renders via
+ format_collation(). #9693"""
+ self.assert_compile(
+ String(50, collation="foo"), "VARCHAR(50) COLLATE foo"
+ )
+
+ def test_string_collation_schema(self):
+ self.assert_compile(
+ String(50, collation="foo", collation_schema="MySchema"),
+ 'VARCHAR(50) COLLATE "MySchema".foo',
+ dialect="postgresql",
+ )
+
+ def test_string_collation_schema_requires_collation(self):
+ assert_raises_message(
+ exc.ArgumentError,
+ "the 'collation_schema' parameter of String requires "
+ "the 'collation' parameter to also be present",
+ String,
+ 50,
+ collation_schema="myschema",
+ )
+
def test_char_plain(self):
self.assert_compile(CHAR(), "CHAR")
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
