self.dialect.set_connection_execution_options(c, opt)
return c
+ def get_execution_options(self):
+ """ Get the non-SQL options which will take effect during execution.
+
+ .. seealso::
+
+ :meth:`.Connection.execution_options`
+ """
+ return self._execution_options
+
@property
def closed(self):
"""Return True if this connection is closed."""
"""
return OptionEngine(self, opt)
+ def get_execution_options(self):
+ """ Get the non-SQL options which will take effect during execution.
+
+ .. seealso::
+
+ :meth:`.Engine.execution_options`
+ """
+ return self._execution_options
+
@property
def name(self):
"""String name of the :class:`~sqlalchemy.engine.interfaces.Dialect`
c2_branch = c2.connect()
eq_(c2_branch._execution_options, {"foo": "bar"})
+ def test_get_engine_execution_options(self):
+ engine = testing_engine("sqlite://")
+ engine.dialect = Mock()
+ e2 = engine.execution_options(foo="bar")
+
+ eq_(e2.get_execution_options(), {"foo": "bar"})
+
+ def test_get_connection_execution_options(self):
+ engine = testing_engine("sqlite://", options=dict(_initialize=False))
+ engine.dialect = Mock()
+ conn = engine.connect()
+ c = conn.execution_options(foo="bar")
+
+ eq_(c.get_execution_options(), {"foo": "bar"})
+
class EngineEventsTest(fixtures.TestBase):
__requires__ = ("ad_hoc_engines",)