--- /dev/null
+.. change::
+ :tags: bug, orm
+ :tickets: 12769
+
+ Improved the behavior of standalone "operators" like :func:`_sql.desc`,
+ :func:`_sql.asc`, :func:`_sql.all_`, etc. so that they consult the given
+ expression object for an overriding method for that operator, even if the
+ object is not itself a ``ClauseElement``, such as if it's an ORM attribute.
+ This allows custom comparators for things like :func:`_orm.composite` to
+ provide custom implementations of methods like ``desc()``, ``asc()``, etc.
+
+
+.. change::
+ :tags: usecase, orm
+ :tickets: 12769
+
+ Added default implementations of :meth:`.ColumnOperators.desc`,
+ :meth:`.ColumnOperators.asc`, :meth:`.ColumnOperators.nulls_first`,
+ :meth:`.ColumnOperators.nulls_last` to :func:`_orm.composite` attributes,
+ by default applying the modifier to all contained columns. Can be
+ overridden using a custom comparator.
:members:
:special-members:
+.. autoclass:: OrderByList
+ :members:
+
.. autoclass:: Over
:members:
from .sql.expression import nullslast as nullslast
from .sql.expression import Operators as Operators
from .sql.expression import or_ as or_
+from .sql.expression import OrderByList as OrderByList
from .sql.expression import outerjoin as outerjoin
from .sql.expression import outparam as outparam
from .sql.expression import Over as Over
def __ge__(self, other: Any) -> ColumnElement[bool]:
return self._compare(operators.ge, other)
+ def desc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
+ return expression.OrderByList(
+ [e.desc() for e in self._comparable_elements]
+ )
+
+ def asc(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
+ return expression.OrderByList(
+ [e.asc() for e in self._comparable_elements]
+ )
+
+ def nulls_first(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
+ return expression.OrderByList(
+ [e.nulls_first() for e in self._comparable_elements]
+ )
+
+ def nulls_last(self) -> operators.OrderingOperators: # type: ignore[override] # noqa: E501
+ return expression.OrderByList(
+ [e.nulls_last() for e in self._comparable_elements]
+ )
+
# what might be interesting would be if we create
# an instance of the composite class itself with
# the columns as data members, then use "hybrid style" comparison
from typing import Union
from . import coercions
+from . import operators
from . import roles
from .base import _NoArg
from .coercions import _document_text_coercion
from .elements import FunctionFilter
from .elements import Label
from .elements import Null
+from .elements import OrderByList
from .elements import Over
from .elements import TextClause
from .elements import True_
:func:`_expression.any_`
"""
- return CollectionAggregate._create_all(expr)
+ if isinstance(expr, operators.ColumnOperators):
+ return expr.all_()
+ else:
+ return CollectionAggregate._create_all(expr)
def and_( # type: ignore[empty-body]
:func:`_expression.all_`
"""
- return CollectionAggregate._create_any(expr)
+ if isinstance(expr, operators.ColumnOperators):
+ return expr.any_()
+ else:
+ return CollectionAggregate._create_any(expr)
+
+
+@overload
+def asc(
+ column: Union[str, "ColumnElement[_T]"],
+) -> UnaryExpression[_T]: ...
+
+
+@overload
+def asc(
+ column: _ColumnExpressionOrStrLabelArgument[_T],
+) -> Union[OrderByList, UnaryExpression[_T]]: ...
def asc(
column: _ColumnExpressionOrStrLabelArgument[_T],
-) -> UnaryExpression[_T]:
+) -> Union[OrderByList, UnaryExpression[_T]]:
"""Produce an ascending ``ORDER BY`` clause element.
e.g.::
:meth:`_expression.Select.order_by`
"""
- return UnaryExpression._create_asc(column)
+
+ if isinstance(column, operators.OrderingOperators):
+ return column.asc() # type: ignore[unused-ignore]
+ else:
+ return UnaryExpression._create_asc(column)
def collate(
identifier, e.g. contains uppercase characters.
"""
- return CollationClause._create_collation_expression(expression, collation)
+ if isinstance(expression, operators.ColumnOperators):
+ return expression.collate(collation) # type: ignore
+ else:
+ return CollationClause._create_collation_expression(
+ expression, collation
+ )
def between(
return ColumnClause(text, type_, is_literal, _selectable)
+@overload
+def desc(
+ column: Union[str, "ColumnElement[_T]"],
+) -> UnaryExpression[_T]: ...
+
+
+@overload
def desc(
column: _ColumnExpressionOrStrLabelArgument[_T],
-) -> UnaryExpression[_T]:
+) -> Union[OrderByList, UnaryExpression[_T]]: ...
+
+
+def desc(
+ column: _ColumnExpressionOrStrLabelArgument[_T],
+) -> Union[OrderByList, UnaryExpression[_T]]:
"""Produce a descending ``ORDER BY`` clause element.
e.g.::
:meth:`_expression.Select.order_by`
"""
- return UnaryExpression._create_desc(column)
+ if isinstance(column, operators.OrderingOperators):
+ return column.desc() # type: ignore[unused-ignore]
+ else:
+ return UnaryExpression._create_desc(column)
def distinct(expr: _ColumnExpressionArgument[_T]) -> UnaryExpression[_T]:
:data:`.func`
""" # noqa: E501
- return UnaryExpression._create_distinct(expr)
+ if isinstance(expr, operators.ColumnOperators):
+ return expr.distinct()
+ else:
+ return UnaryExpression._create_distinct(expr)
def bitwise_not(expr: _ColumnExpressionArgument[_T]) -> UnaryExpression[_T]:
"""
-
- return UnaryExpression._create_bitwise_not(expr)
+ if isinstance(expr, operators.ColumnOperators):
+ return expr.bitwise_not()
+ else:
+ return UnaryExpression._create_bitwise_not(expr)
def extract(field: str, expr: _ColumnExpressionArgument[Any]) -> Extract:
return Null._instance()
-def nulls_first(column: _ColumnExpressionArgument[_T]) -> UnaryExpression[_T]:
+@overload
+def nulls_first(
+ column: "ColumnElement[_T]",
+) -> UnaryExpression[_T]: ...
+
+
+@overload
+def nulls_first(
+ column: _ColumnExpressionArgument[_T],
+) -> Union[OrderByList, UnaryExpression[_T]]: ...
+
+
+def nulls_first(
+ column: _ColumnExpressionArgument[_T],
+) -> Union[OrderByList, UnaryExpression[_T]]:
"""Produce the ``NULLS FIRST`` modifier for an ``ORDER BY`` expression.
:func:`.nulls_first` is intended to modify the expression produced
:meth:`_expression.Select.order_by`
""" # noqa: E501
- return UnaryExpression._create_nulls_first(column)
+ if isinstance(column, operators.OrderingOperators):
+ return column.nulls_first()
+ else:
+ return UnaryExpression._create_nulls_first(column)
+
+
+@overload
+def nulls_last(
+ column: "ColumnElement[_T]",
+) -> UnaryExpression[_T]: ...
+
+
+@overload
+def nulls_last(
+ column: _ColumnExpressionArgument[_T],
+) -> Union[OrderByList, UnaryExpression[_T]]: ...
-def nulls_last(column: _ColumnExpressionArgument[_T]) -> UnaryExpression[_T]:
+def nulls_last(
+ column: _ColumnExpressionArgument[_T],
+) -> Union[OrderByList, UnaryExpression[_T]]:
"""Produce the ``NULLS LAST`` modifier for an ``ORDER BY`` expression.
:func:`.nulls_last` is intended to modify the expression produced
:meth:`_expression.Select.order_by`
""" # noqa: E501
- return UnaryExpression._create_nulls_last(column)
+ if isinstance(column, operators.OrderingOperators):
+ return column.nulls_last()
+ else:
+ return UnaryExpression._create_nulls_last(column)
def or_( # type: ignore[empty-body]
def visit_element_list(self, element, **kw):
return self._generate_delimited_list(element.clauses, " ", **kw)
+ def visit_order_by_list(self, element, **kw):
+ return self._generate_delimited_list(element.clauses, ", ", **kw)
+
def visit_clauselist(self, clauselist, **kw):
sep = clauselist.operator
if sep is None:
self.clauses = tuple(clauses)
+class OrderByList(
+ roles.OrderByRole,
+ operators.OrderingOperators,
+ DQLDMLClauseElement,
+):
+ """Describe a list of clauses that will be comma separated to nest
+ within an ORDER BY.
+
+ .. versionadded:: 2.1
+
+ """
+
+ __visit_name__ = "order_by_list"
+
+ _traverse_internals: _TraverseInternalsType = [
+ ("clauses", InternalTraversal.dp_clauseelement_tuple),
+ ]
+
+ clauses: List[ColumnElement[Any]]
+
+ def __init__(
+ self,
+ clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
+ ):
+ text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
+ self._text_converter_role = text_converter_role
+
+ self.clauses = [
+ coercions.expect(
+ text_converter_role, clause, apply_propagate_attrs=self
+ )
+ for clause in clauses
+ ]
+
+ def __iter__(self) -> Iterator[ColumnElement[Any]]:
+ return iter(self.clauses)
+
+ def __len__(self) -> int:
+ return len(self.clauses)
+
+ @property
+ def _select_iterable(self) -> _SelectIterable:
+ return itertools.chain.from_iterable(
+ [elem._select_iterable for elem in self.clauses]
+ )
+
+ @util.ro_non_memoized_property
+ def _from_objects(self) -> List[FromClause]:
+ return list(itertools.chain(*[c._from_objects for c in self.clauses]))
+
+ def self_group(
+ self, against: Optional[OperatorType] = None
+ ) -> Union[Self, Grouping[Any]]:
+ return self
+
+ def desc(self) -> OrderByList:
+ return OrderByList([e.desc() for e in self.clauses])
+
+ def asc(self) -> OrderByList:
+ return OrderByList([e.asc() for e in self.clauses])
+
+ def nulls_first(self) -> OrderByList:
+ return OrderByList([e.nulls_first() for e in self.clauses])
+
+ def nulls_last(self) -> OrderByList:
+ return OrderByList([e.nulls_last() for e in self.clauses])
+
+
class ClauseList(
roles.InElementRole,
roles.OrderByRole,
__visit_name__ = "clauselist"
- # this is used only by the ORM in a legacy use case for
- # composite attributes
+ # Used by ORM context.py to identify ClauseList objects in legacy
+ # composite attribute queries (see test_query_cols_legacy test)
_is_clause_list = True
_traverse_internals: _TraverseInternalsType = [
def _create_desc(
cls, column: _ColumnExpressionOrStrLabelArgument[_T]
) -> UnaryExpression[_T]:
+
return UnaryExpression(
coercions.expect(roles.ByOfRole, column),
modifier=operators.desc_op,
from .elements import literal as literal
from .elements import literal_column as literal_column
from .elements import Null as Null
+from .elements import OrderByList as OrderByList
from .elements import Over as Over
from .elements import quoted_name as quoted_name
from .elements import ReleaseSavepointClause as ReleaseSavepointClause
)
-class ColumnOperators(Operators):
+class OrderingOperators(Operators):
+ """defines ORDER BY operators, which can operate as single expressions
+ or comma-separated lists
+
+ """
+
+ __slots__ = ()
+
+ if typing.TYPE_CHECKING:
+
+ def operate(
+ self, op: OperatorType, *other: Any, **kwargs: Any
+ ) -> OrderingOperators: ...
+
+ def reverse_operate(
+ self, op: OperatorType, other: Any, **kwargs: Any
+ ) -> OrderingOperators: ...
+
+ def desc(self) -> OrderingOperators:
+ """Produce a :func:`_expression.desc` clause against the
+ parent object."""
+ return self.operate(desc_op)
+
+ def asc(self) -> OrderingOperators:
+ """Produce a :func:`_expression.asc` clause against the
+ parent object."""
+ return self.operate(asc_op)
+
+ def nulls_first(self) -> OrderingOperators:
+ """Produce a :func:`_expression.nulls_first` clause against the
+ parent object.
+
+ .. versionchanged:: 1.4 The ``nulls_first()`` operator is
+ renamed from ``nullsfirst()`` in previous releases.
+ The previous name remains available for backwards compatibility.
+ """
+ return self.operate(nulls_first_op)
+
+ # deprecated 1.4; see #5435
+ if TYPE_CHECKING:
+
+ def nullsfirst(self) -> OrderingOperators: ...
+
+ else:
+ nullsfirst = nulls_first
+
+ def nulls_last(self) -> OrderingOperators:
+ """Produce a :func:`_expression.nulls_last` clause against the
+ parent object.
+
+ .. versionchanged:: 1.4 The ``nulls_last()`` operator is
+ renamed from ``nullslast()`` in previous releases.
+ The previous name remains available for backwards compatibility.
+ """
+ return self.operate(nulls_last_op)
+
+ # deprecated 1.4; see #5429
+ if TYPE_CHECKING:
+
+ def nullslast(self) -> OrderingOperators: ...
+
+ else:
+ nullslast = nulls_last
+
+
+class ColumnOperators(OrderingOperators):
"""Defines boolean, comparison, and other operators for
:class:`_expression.ColumnElement` expressions.
flags=flags,
)
- def desc(self) -> ColumnOperators:
- """Produce a :func:`_expression.desc` clause against the
- parent object."""
- return self.operate(desc_op)
-
- def asc(self) -> ColumnOperators:
- """Produce a :func:`_expression.asc` clause against the
- parent object."""
- return self.operate(asc_op)
-
- def nulls_first(self) -> ColumnOperators:
- """Produce a :func:`_expression.nulls_first` clause against the
- parent object.
-
- .. versionchanged:: 1.4 The ``nulls_first()`` operator is
- renamed from ``nullsfirst()`` in previous releases.
- The previous name remains available for backwards compatibility.
- """
- return self.operate(nulls_first_op)
-
- # deprecated 1.4; see #5435
- if TYPE_CHECKING:
-
- def nullsfirst(self) -> ColumnOperators: ...
-
- else:
- nullsfirst = nulls_first
-
- def nulls_last(self) -> ColumnOperators:
- """Produce a :func:`_expression.nulls_last` clause against the
- parent object.
-
- .. versionchanged:: 1.4 The ``nulls_last()`` operator is
- renamed from ``nullslast()`` in previous releases.
- The previous name remains available for backwards compatibility.
- """
- return self.operate(nulls_last_op)
-
- # deprecated 1.4; see #5429
- if TYPE_CHECKING:
-
- def nullslast(self) -> ColumnOperators: ...
-
- else:
- nullslast = nulls_last
-
def collate(self, collation: str) -> ColumnOperators:
"""Produce a :func:`_expression.collate` clause against
the parent object, given the collation string.
import random
import sqlalchemy as sa
+from sqlalchemy import asc
+from sqlalchemy import desc
from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import insert
from sqlalchemy import inspect
from sqlalchemy import Integer
+from sqlalchemy import nulls_first
+from sqlalchemy import nulls_last
+from sqlalchemy import OrderByList
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
diff_y = clauses[1] - other.y
return diff_x * diff_x + diff_y * diff_y <= d * d
+ def desc(self):
+ clauses = self.__clause_element__().clauses
+ return OrderByList([clauses[0].desc(), clauses[1].asc()])
+
self.mapper_registry.map_imperatively(
Edge,
edge,
Edge = self.classes.Edge
s = fixture_session()
self.assert_compile(
- s.query(Edge).order_by(Edge.start, Edge.end),
+ s.query(Edge).order_by(Edge.start, Edge.end.desc()),
+ "SELECT edge.id AS edge_id, edge.x1 AS edge_x1, "
+ "edge.y1 AS edge_y1, edge.x2 AS edge_x2, edge.y2 AS edge_y2 "
+ "FROM edge ORDER BY edge.x1, edge.y1, edge.x2 DESC, edge.y2 DESC",
+ )
+
+ self.assert_compile(
+ s.query(Edge).order_by(
+ Edge.start.asc().nulls_first(), Edge.end.nulls_last()
+ ),
+ "SELECT edge.id AS edge_id, edge.x1 AS edge_x1, "
+ "edge.y1 AS edge_y1, edge.x2 AS edge_x2, edge.y2 AS edge_y2 "
+ "FROM edge ORDER BY edge.x1 ASC NULLS FIRST, "
+ "edge.y1 ASC NULLS FIRST, edge.x2 NULLS LAST, edge.y2 NULLS LAST",
+ )
+
+ # Test using standalone ops syntax
+
+ self.assert_compile(
+ s.query(Edge).order_by(
+ nulls_first(asc(Edge.start)), nulls_last(Edge.end)
+ ),
+ "SELECT edge.id AS edge_id, edge.x1 AS edge_x1, "
+ "edge.y1 AS edge_y1, edge.x2 AS edge_x2, edge.y2 AS edge_y2 "
+ "FROM edge ORDER BY edge.x1 ASC NULLS FIRST, "
+ "edge.y1 ASC NULLS FIRST, edge.x2 NULLS LAST, edge.y2 NULLS LAST",
+ )
+
+ def test_order_by_custom(self):
+ """test #12769"""
+ self._fixture(True)
+ Edge = self.classes.Edge
+ s = fixture_session()
+
+ self.assert_compile(
+ s.query(Edge).order_by(Edge.start.desc()),
+ "SELECT edge.id AS edge_id, edge.x1 AS edge_x1, "
+ "edge.y1 AS edge_y1, edge.x2 AS edge_x2, edge.y2 AS edge_y2 "
+ "FROM edge "
+ "ORDER BY edge.x1 DESC, edge.y1 ASC",
+ )
+
+ self.assert_compile(
+ s.query(Edge).order_by(Edge.start.desc().nulls_first()),
+ "SELECT edge.id AS edge_id, edge.x1 AS edge_x1, "
+ "edge.y1 AS edge_y1, edge.x2 AS edge_x2, edge.y2 AS edge_y2 "
+ "FROM edge "
+ "ORDER BY edge.x1 DESC NULLS FIRST, edge.y1 ASC NULLS FIRST",
+ )
+
+ self.assert_compile(
+ s.query(Edge).order_by(desc(Edge.start)),
"SELECT edge.id AS edge_id, edge.x1 AS edge_x1, "
"edge.y1 AS edge_y1, edge.x2 AS edge_x2, edge.y2 AS edge_y2 "
- "FROM edge ORDER BY edge.x1, edge.y1, edge.x2, edge.y2",
+ "FROM edge "
+ "ORDER BY edge.x1 DESC, edge.y1 ASC",
)
def test_order_by_aliased(self):
from sqlalchemy.sql.elements import ElementList
from sqlalchemy.sql.elements import Immutable
from sqlalchemy.sql.elements import Null
+from sqlalchemy.sql.elements import OrderByList
from sqlalchemy.sql.elements import Slice
from sqlalchemy.sql.elements import TypeClause
from sqlalchemy.sql.elements import UnaryExpression
ElementList([table_a.c.a]),
ElementList([table_a.c.a, table_a.c.b]),
),
+ lambda: (
+ table_a.c.a,
+ OrderByList([table_a.c.a]),
+ OrderByList(
+ [table_a.c.a, OrderByList([table_a.c.b, table_b.c.a])]
+ ),
+ ),
lambda: (_textual_label_reference("a"), _textual_label_reference("b")),
lambda: (
text("select a, b from table").columns(a=Integer, b=String),
from sqlalchemy import nullslast
from sqlalchemy import Numeric
from sqlalchemy import or_
+from sqlalchemy import OrderByList
from sqlalchemy import outerjoin
from sqlalchemy import over
from sqlalchemy import schema
select(t1).order_by(t1.c.d),
"SELECT t1.id, t1.a, t1.c, t1.e FROM t1 ORDER BY t1.d",
)
+
+
+class OrderByListTest(fixtures.TestBase, AssertsCompiledSQL):
+ __dialect__ = "default"
+
+ def test_order_by_list(self):
+ """Test standalone OrderByList with various operators"""
+ col1 = Column("x", Integer)
+ col2 = Column("y", Integer)
+
+ # Test basic OrderByList creation
+ order_list = OrderByList([col1, col2])
+ self.assert_compile(
+ select(literal(1)).order_by(order_list),
+ "SELECT :param_1 AS anon_1 ORDER BY x, y",
+ )
+
+ # Test OrderByList with desc
+ order_list_desc = order_list.desc()
+ self.assert_compile(
+ select(literal(1)).order_by(order_list_desc),
+ "SELECT :param_1 AS anon_1 ORDER BY x DESC, y DESC",
+ )
+
+ # Test OrderByList with asc
+ order_list_asc = order_list.asc()
+ self.assert_compile(
+ select(literal(1)).order_by(order_list_asc),
+ "SELECT :param_1 AS anon_1 ORDER BY x ASC, y ASC",
+ )
+
+ # Test OrderByList with nulls_first
+ order_list_nf = order_list.nulls_first()
+ self.assert_compile(
+ select(literal(1)).order_by(order_list_nf),
+ "SELECT :param_1 AS anon_1 ORDER BY x NULLS FIRST, y NULLS FIRST",
+ )
+
+ # Test OrderByList with nulls_last
+ order_list_nl = order_list.nulls_last()
+ self.assert_compile(
+ select(literal(1)).order_by(order_list_nl),
+ "SELECT :param_1 AS anon_1 ORDER BY x NULLS LAST, y NULLS LAST",
+ )
+
+ def test_order_by_list_chained_ops(self):
+ """Test chained operations on OrderByList"""
+ col1 = Column("x", Integer)
+ col2 = Column("y", Integer)
+
+ order_list = OrderByList([col1, col2])
+
+ # Test chained desc().nulls_first()
+ chained = order_list.desc().nulls_first()
+ self.assert_compile(
+ select(literal(1)).order_by(chained),
+ "SELECT :param_1 AS anon_1 ORDER BY x DESC NULLS FIRST, "
+ "y DESC NULLS FIRST",
+ )
import pickle
import re
+from sqlalchemy import all_
from sqlalchemy import and_
+from sqlalchemy import any_
+from sqlalchemy import asc
from sqlalchemy import between
from sqlalchemy import bindparam
+from sqlalchemy import bitwise_not
+from sqlalchemy import desc
+from sqlalchemy import distinct
from sqlalchemy import exc
from sqlalchemy import Float
from sqlalchemy import Integer
from sqlalchemy import LargeBinary
from sqlalchemy import literal_column
from sqlalchemy import not_
+from sqlalchemy import nulls_first
+from sqlalchemy import nulls_last
from sqlalchemy import Numeric
from sqlalchemy import or_
+from sqlalchemy import SQLColumnExpression
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy.schema import Column
from sqlalchemy.schema import MetaData
from sqlalchemy.schema import Table
-from sqlalchemy.sql import all_
-from sqlalchemy.sql import any_
-from sqlalchemy.sql import asc
from sqlalchemy.sql import coercions
from sqlalchemy.sql import collate
from sqlalchemy.sql import column
from sqlalchemy.sql import compiler
-from sqlalchemy.sql import desc
from sqlalchemy.sql import false
from sqlalchemy.sql import LABEL_STYLE_TABLENAME_PLUS_COL
from sqlalchemy.sql import literal
from sqlalchemy.sql.expression import tuple_
from sqlalchemy.sql.expression import UnaryExpression
from sqlalchemy.sql.expression import union
+from sqlalchemy.sql.operators import ColumnOperators
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import combinations
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_not
+from sqlalchemy.testing import mock
from sqlalchemy.testing import resolve_lambda
from sqlalchemy.testing.assertions import expect_deprecated
from sqlalchemy.types import ARRAY
select(py_op(c1, c2)),
f"SELECT c1 {sql_op} c2 AS anon_1",
)
+
+
+class StandaloneOperatorTranslateTest(
+ fixtures.TestBase, testing.AssertsCompiledSQL
+):
+ __dialect__ = "default"
+
+ def _combinations(fn):
+ return testing.combinations(
+ desc,
+ asc,
+ nulls_first,
+ nulls_last,
+ any_,
+ all_,
+ distinct,
+ bitwise_not,
+ collate,
+ )(fn)
+
+ @_combinations
+ def test_move(self, operator):
+ m1 = column("q")
+ m2 = mock.Mock()
+
+ class MyCustomThing(roles.ByOfRole, SQLColumnExpression):
+ def __clause_element__(self):
+ return m1
+
+ @property
+ def comparator(self):
+ return Comparator()
+
+ def operate(
+ self,
+ op,
+ *other,
+ **kwargs,
+ ):
+ return op(self.comparator, *other, **kwargs)
+
+ def reverse_operate(
+ self,
+ op,
+ *other,
+ **kwargs,
+ ):
+ return op(other, self.comparator, **kwargs)
+
+ class Comparator(ColumnOperators):
+ def _operate(self, *arg, **kw):
+ return m2
+
+ setattr(Comparator, operator.__name__, Comparator._operate)
+
+ mc = MyCustomThing()
+
+ if operator is collate:
+ result = operator(mc, "some collation")
+ else:
+ result = operator(mc)
+
+ is_(result, m2)
+
+ @_combinations
+ def test_text(self, operator):
+ if operator is collate:
+ result = operator(text("foo"), "some collation")
+ else:
+ result = operator(text("foo"))
+
+ # Assert that the operation completed without crashing
+ # and returned a valid SQL expression
+ assert result is not None
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
