)
)
+ def test_before_compile_execution_options(self):
+ User = self.classes.User
+
+ @event.listens_for(query.Query, "before_compile", retval=True)
+ def _before_compile(query):
+ return query.execution_options(my_option=True)
+
+ opts = {}
+
+ @event.listens_for(testing.db, "before_cursor_execute")
+ def _before_cursor_execute(
+ conn, cursor, statement, parameters, context, executemany
+ ):
+ opts.update(context.execution_options)
+
+ sess = create_session(bind=testing.db, autocommit=False)
+ sess.query(User).first()
+ eq_(opts["my_option"], True)
+
class RefreshFlushInReturningTest(fixtures.MappedTest):
"""test [ticket:3427].
from sqlalchemy import column
from sqlalchemy import desc
from sqlalchemy import distinct
-from sqlalchemy import event
from sqlalchemy import exc as sa_exc
from sqlalchemy import exists
from sqlalchemy import ForeignKey
q1 = sess.query(User).execution_options(**execution_options)
q1.all()
- def test_options_before_compile(self):
- User = self.classes.User
-
- @event.listens_for(Query, "before_compile", retval=True)
- def _before_compile(query):
- return query.execution_options(my_option=True)
-
- opts = {}
-
- @event.listens_for(testing.db, "before_cursor_execute")
- def _before_cursor_execute(
- conn, cursor, statement, parameters, context, executemany
- ):
- opts.update(context.execution_options)
-
- sess = create_session(bind=testing.db, autocommit=False)
- sess.query(User).first()
- eq_(opts["my_option"], True)
-
class BooleanEvalTest(fixtures.TestBase, testing.AssertsCompiledSQL):
"""test standalone booleans being wrapped in an AsBoolean, as well