implement python_impl to custom_op for basic ORM evaluator extensibility

Added new parameter :paramref:`_sql.Operators.op.python_impl`, available
from :meth:`_sql.Operators.op` and also when using the
:class:`_sql.Operators.custom_op` constructor directly, which allows an
in-Python evaluation function to be provided along with the custom SQL
operator. This evaluation function becomes the implementation used when the
operator object is used given plain Python objects as operands on both
sides, and in particular is compatible with the
``synchronize_session='evaluate'`` option used with
:ref:`orm_expression_update_delete`.

Fixes: #3162
Change-Id: If46ba6a0e303e2180a177ba418a8cafe9b42608e

1 files changed, 23 insertions, 0 deletions

diff --git a/test/sql/test_operators.py b/test/sql/test_operators.py
index 9e47f217f..2c77c39f3 100644
--- a/test/sql/test_operators.py
+++ b/test/sql/test_operators.py
@@ -55,6 +55,7 @@ from sqlalchemy.sql.expression import union
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import combinations
 from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises_message
 from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
@@ -3286,6 +3287,28 @@ class CustomOpTest(fixtures.TestBase):
         )
         is_(expr.type, some_return_type)
 
+    def test_python_impl(self):
+        """test #3162"""
+        c = column("x")
+        c2 = column("y")
+        op1 = c.op("$", python_impl=lambda a, b: a > b)(c2).operator
+
+        is_(op1(3, 5), False)
+        is_(op1(5, 3), True)
+
+    def test_python_impl_not_present(self):
+        """test #3162"""
+        c = column("x")
+        c2 = column("y")
+        op1 = c.op("$")(c2).operator
+
+        with expect_raises_message(
+            exc.InvalidRequestError,
+            r"Custom operator '\$' can't be used with plain Python objects "
+            "unless it includes the 'python_impl' parameter.",
+        ):
+            op1(3, 5)
+
 
 class TupleTypingTest(fixtures.TestBase):
     def _assert_types(self, expr):