feat(postgres): add expression-level COLLATE support

changelog.d/40.added.md

@@ -0,0 +1 @@
+Added PostgreSQL expression-level `COLLATE` support with schema-qualified collation validation.

src/rhosocial/activerecord/backend/impl/postgres/__init__.py

@@ -18,6 +18,7 @@
 """
 
 from .backend import PostgresBackend, AsyncPostgresBackend
+from .collation import PostgresCollation
 from .config import PostgresConnectionConfig
 from .dialect import PostgresDialect
 from .transaction import PostgresTransactionManager, AsyncPostgresTransactionManager

src/rhosocial/activerecord/backend/impl/postgres/collation.py

@@ -0,0 +1,43 @@
+# src/rhosocial/activerecord/backend/impl/postgres/collation.py
+"""
+PostgreSQL collation names supported by the dialect whitelist.
+"""
+
+from enum import Enum
+from typing import Optional, Tuple
+
+
+class PostgresCollation(Enum):
+    """Common PostgreSQL collations for expression-level COLLATE."""
+
+    C = "C"
+    POSIX = "POSIX"
+    UCS_BASIC = "ucs_basic"
+    UND_X_ICU = "und-x-icu"
+
+
+_BUILTIN_COLLATIONS = {
+    PostgresCollation.C.value,
+    PostgresCollation.POSIX.value,
+    PostgresCollation.UCS_BASIC.value,
+}
+
+_VERSION_SENSITIVE_COLLATIONS = {
+    PostgresCollation.UND_X_ICU.value: (10, 0, 0),
+}
+
+_POSTGRES_COLLATIONS = _BUILTIN_COLLATIONS | set(_VERSION_SENSITIVE_COLLATIONS)
+
+
+def validate_postgres_collation_name(
+    name: str,
+    version: Optional[Tuple[int, int, int]] = None,
+) -> str:
+    if name not in _POSTGRES_COLLATIONS:
+        raise ValueError(f"Unsupported PostgreSQL collation: {name!r}")
+    required_version = _VERSION_SENSITIVE_COLLATIONS.get(name)
+    if required_version is not None and version is not None and version < required_version:
+        raise ValueError(
+            f"PostgreSQL collation requires PostgreSQL {required_version[0]}.{required_version[1]}+: {name!r}"
+        )
+    return name

src/rhosocial/activerecord/backend/impl/postgres/dialect.py

@@ -9,10 +9,12 @@
 from typing import Any, Dict, Tuple, Optional, TYPE_CHECKING
 
 if TYPE_CHECKING:
+    from rhosocial.activerecord.backend.expression.collation import CollateExpression
     from .function_versions import FunctionSupportInfo, FunctionVersionRequirement
 
 from rhosocial.activerecord.backend.dialect.base import SQLDialectBase
 from rhosocial.activerecord.backend.dialect.mixins import (
+    CollationMixin,
     CTEMixin,
     FilterClauseMixin,
     WindowFunctionMixin,
@@ -41,6 +43,7 @@
     ConstraintMixin,
 )
 from rhosocial.activerecord.backend.dialect.protocols import (
+    CollationSupport,
     CTESupport,
     FilterClauseSupport,
     WindowFunctionSupport,
@@ -133,6 +136,8 @@
     PostgresIntrospectionCapabilityMixin,
 )
 
+from .collation import validate_postgres_collation_name
+
 # PostgreSQL-specific imports
 from .protocols import (
     PostgresExtensionSupport,
@@ -221,6 +226,7 @@
 
 class PostgresDialect(
     SQLDialectBase,
+    CollationMixin,
     SetOperationMixin,
     TruncateMixin,
     ILIKEMixin,
@@ -310,6 +316,7 @@ class PostgresDialect(
     PostgresStoredProcedureMixin,
     PostgresAdvisoryLockMixin,
     # Protocol supports
+    CollationSupport,
     SetOperationSupport,
     TruncateSupport,
     ILIKESupport,
@@ -464,6 +471,22 @@ def get_server_version(self) -> Tuple[int, int, int]:
         """Return the PostgreSQL version this dialect is configured for."""
         return self.version
 
+    def supports_collate_expression(self) -> bool:
+        """PostgreSQL supports expression-level COLLATE."""
+        return True
+
+    def validate_collation_name(self, expr: "CollateExpression") -> str:
+        """Validate PostgreSQL collation names and return their SQL representation."""
+        schema = expr.collation_options.get("schema")
+        unsupported = set(expr.collation_options) - {"schema"}
+        if unsupported:
+            options = ", ".join(sorted(unsupported))
+            raise UnsupportedFeatureError(self.name, f"COLLATE options: {options}")
+        validate_postgres_collation_name(expr.collation_name, getattr(self, "version", None))
+        if schema is not None:
+            return f"{self.format_identifier(str(schema))}.{self.format_identifier(expr.collation_name)}"
+        return self.format_identifier(expr.collation_name)
+
     # region Protocol Support Checks based on version
     def supports_basic_cte(self) -> bool:
         """Basic CTEs are supported since PostgreSQL 8.4."""

tests/rhosocial/activerecord_postgres_test/feature/backend/test_postgres_protocol_conformance.py

@@ -77,6 +77,7 @@ def get_postgres_protocols():
     from typing import runtime_checkable
 
     protocols = [
+        dialect_protocols.CollationSupport,
         dialect_protocols.CTESupport,
         dialect_protocols.FilterClauseSupport,
         dialect_protocols.WindowFunctionSupport,

tests/rhosocial/activerecord_postgres_test/feature/query/test_collation_expression.py

@@ -0,0 +1,69 @@
+# tests/rhosocial/activerecord_postgres_test/feature/query/test_collation_expression.py
+"""
+Tests for expression-level COLLATE support on PostgreSQL.
+"""
+
+import pytest
+
+from rhosocial.activerecord.backend.expression import Column, Literal
+from rhosocial.activerecord.backend.impl.postgres import PostgresCollation, PostgresDialect
+
+
+@pytest.fixture
+def dialect():
+    return PostgresDialect(version=(15, 0, 0))
+
+
+class TestPostgresCollationExpression:
+    def test_column_collate_generates_sql(self, dialect):
+        expr = Column(dialect, "name", table="users").collate(PostgresCollation.C)
+
+        sql, params = expr.to_sql()
+
+        assert sql == '"users"."name" COLLATE "C"'
+        assert params == ()
+
+    def test_literal_collate_preserves_parameter_binding(self, dialect):
+        expr = Literal(dialect, "Alice").collate(PostgresCollation.POSIX)
+
+        sql, params = expr.to_sql()
+
+        assert sql == '%s COLLATE "POSIX"'
+        assert params == ("Alice",)
+
+    def test_schema_qualified_collation_generates_sql(self, dialect):
+        expr = Column(dialect, "name").collate(PostgresCollation.C.value, schema="pg_catalog")
+
+        sql, params = expr.to_sql()
+
+        assert sql == '"name" COLLATE "pg_catalog"."C"'
+        assert params == ()
+
+    def test_rejects_unsupported_collation(self, dialect):
+        expr = Column(dialect, "name").collate("unknown_ci")
+
+        with pytest.raises(ValueError, match="Unsupported PostgreSQL collation"):
+            expr.to_sql()
+
+    def test_rejects_version_sensitive_collation_on_older_version(self):
+        dialect = PostgresDialect(version=(9, 6, 0))
+        expr = Column(dialect, "name").collate(PostgresCollation.UND_X_ICU)
+
+        with pytest.raises(ValueError, match="requires PostgreSQL 10.0"):
+            expr.to_sql()
+
+    def test_collate_executes_order_by(self, annotated_query_fixtures):
+        SearchableItem = (
+            annotated_query_fixtures[0]
+            if isinstance(annotated_query_fixtures, tuple)
+            else annotated_query_fixtures
+        )
+        SearchableItem(name="b", tags=[]).save()
+        SearchableItem(name="A", tags=[]).save()
+        SearchableItem(name="c", tags=[]).save()
+
+        results = SearchableItem.query().order_by(
+            SearchableItem.c.name.collate(PostgresCollation.C)
+        ).all()
+
+        assert [item.name for item in results] == ["A", "b", "c"]