Handle mappings passed to ``execution_options``.

Fixed a bug in :meth:`_asyncio.AsyncSession.execute` and
:meth:`_asyncio.AsyncSession.stream` that required ``execution_options``
to be an instance of ``immutabledict`` when defined. It now
correctly accepts any mapping.

Fixes: #6943
Change-Id: Ic09de480dc2da1b0bdce25acb60b8f01371971f9

3 files changed, 17 insertions, 4 deletions

diff --git a/lib/sqlalchemy/ext/asyncio/session.py b/lib/sqlalchemy/ext/asyncio/session.py
index a10621eef..5c6e7f5a7 100644
--- a/lib/sqlalchemy/ext/asyncio/session.py
+++ b/lib/sqlalchemy/ext/asyncio/session.py
@@ -14,6 +14,9 @@ from ...orm import Session
 from ...orm import state as _instance_state
 from ...util.concurrency import greenlet_spawn
 
+_EXECUTE_OPTIONS = util.immutabledict({"prebuffer_rows": True})
+_STREAM_OPTIONS = util.immutabledict({"stream_results": True})
+
 
 @util.create_proxy_methods(
     Session,
@@ -140,7 +143,12 @@ class AsyncSession(ReversibleProxy):
         """Execute a statement and return a buffered
         :class:`_engine.Result` object."""
 
-        execution_options = execution_options.union({"prebuffer_rows": True})
+        if execution_options:
+            execution_options = util.immutabledict(execution_options).union(
+                _EXECUTE_OPTIONS
+            )
+        else:
+            execution_options = _EXECUTE_OPTIONS
 
         return await greenlet_spawn(
             self.sync_session.execute,
@@ -205,7 +213,12 @@ class AsyncSession(ReversibleProxy):
         """Execute a statement and return a streaming
         :class:`_asyncio.AsyncResult` object."""
 
-        execution_options = execution_options.union({"stream_results": True})
+        if execution_options:
+            execution_options = util.immutabledict(execution_options).union(
+                _STREAM_OPTIONS
+            )
+        else:
+            execution_options = _STREAM_OPTIONS
 
         result = await greenlet_spawn(
             self.sync_session.execute,
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 4747d0bba..fd484b52b 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -1833,7 +1833,7 @@ class BulkUDCompileState(CompileState):
         return (
             statement,
             util.immutabledict(execution_options).union(
-                dict(_sa_orm_update_options=update_options)
+                {"_sa_orm_update_options": update_options}
             ),
         )
 
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index af803a1b0..0bdd5cc95 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1581,7 +1581,7 @@ class Session(_SessionClassMethods):
         :param execution_options: optional dictionary of execution options,
          which will be associated with the statement execution.  This
          dictionary can provide a subset of the options that are accepted
-         by :meth:`_future.Connection.execution_options`, and may also
+         by :meth:`_engine.Connection.execution_options`, and may also
          provide additional options understood only in an ORM context.
 
         :param bind_arguments: dictionary of additional arguments to determine