4 files changed, 34 insertions, 7 deletions
diff --git a/lib/sqlalchemy/ext/asyncio/engine.py b/lib/sqlalchemy/ext/asyncio/engine.py
index bb51a4d22..fb05f512e 100644
--- a/lib/sqlalchemy/ext/asyncio/engine.py
+++ b/lib/sqlalchemy/ext/asyncio/engine.py
@@ -48,6 +48,7 @@ if TYPE_CHECKING:
     from ...engine.url import URL
     from ...pool import Pool
     from ...pool import PoolProxiedConnection
+    from ...sql._typing import _InfoType
     from ...sql.base import Executable
 
 
@@ -241,8 +242,8 @@ class AsyncConnection(
 
         return await greenlet_spawn(getattr, self._proxied, "connection")
 
-    @property
-    def info(self) -> Dict[str, Any]:
+    @util.ro_non_memoized_property
+    def info(self) -> _InfoType:
         """Return the :attr:`_engine.Connection.info` dictionary of the
         underlying :class:`_engine.Connection`.
 
@@ -464,7 +465,7 @@ class AsyncConnection(
          * :class:`_expression.TextClause` and
            :class:`_expression.TextualSelect`
          * :class:`_schema.DDL` and objects which inherit from
-           :class:`_schema.DDLElement`
+           :class:`_schema.ExecutableDDLElement`
 
         :param parameters: parameters which will be bound into the statement.
          This may be either a dictionary of parameter names to values,
diff --git a/lib/sqlalchemy/ext/asyncio/result.py b/lib/sqlalchemy/ext/asyncio/result.py
index a9db822a6..d0337554c 100644
--- a/lib/sqlalchemy/ext/asyncio/result.py
+++ b/lib/sqlalchemy/ext/asyncio/result.py
@@ -710,7 +710,14 @@ _RT = TypeVar("_RT", bound="Result")
 
 async def _ensure_sync_result(result: _RT, calling_method: Any) -> _RT:
     cursor_result: CursorResult
-    if not result._is_cursor:
+
+    try:
+        is_cursor = result._is_cursor
+    except AttributeError:
+        # legacy execute(DefaultGenerator) case
+        return result
+
+    if not is_cursor:
         cursor_result = getattr(result, "raw", None)  # type: ignore
     else:
         cursor_result = result  # type: ignore
diff --git a/lib/sqlalchemy/ext/asyncio/scoping.py b/lib/sqlalchemy/ext/asyncio/scoping.py
index 0d6ae92b4..33cf3f745 100644
--- a/lib/sqlalchemy/ext/asyncio/scoping.py
+++ b/lib/sqlalchemy/ext/asyncio/scoping.py
@@ -442,6 +442,7 @@ class async_scoped_session:
         self,
         statement: Executable,
         params: Optional[_CoreAnyExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -914,6 +915,7 @@ class async_scoped_session:
         self,
         statement: Executable,
         params: Optional[_CoreSingleExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -944,6 +946,7 @@ class async_scoped_session:
         self,
         statement: Executable,
         params: Optional[_CoreSingleExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -980,6 +983,7 @@ class async_scoped_session:
         self,
         statement: Executable,
         params: Optional[_CoreAnyExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -1007,6 +1011,7 @@ class async_scoped_session:
         self,
         statement: Executable,
         params: Optional[_CoreSingleExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
diff --git a/lib/sqlalchemy/ext/asyncio/session.py b/lib/sqlalchemy/ext/asyncio/session.py
index 7d63b084c..1422f99a3 100644
--- a/lib/sqlalchemy/ext/asyncio/session.py
+++ b/lib/sqlalchemy/ext/asyncio/session.py
@@ -55,6 +55,7 @@ if TYPE_CHECKING:
     from ...orm.session import _PKIdentityArgument
     from ...orm.session import _SessionBind
     from ...orm.session import _SessionBindKey
+    from ...sql._typing import _InfoType
     from ...sql.base import Executable
     from ...sql.elements import ClauseElement
     from ...sql.selectable import ForUpdateArg
@@ -260,6 +261,7 @@ class AsyncSession(ReversibleProxy[Session]):
         self,
         statement: Executable,
         params: Optional[_CoreAnyExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -294,6 +296,7 @@ class AsyncSession(ReversibleProxy[Session]):
         self,
         statement: Executable,
         params: Optional[_CoreSingleExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -306,19 +309,28 @@ class AsyncSession(ReversibleProxy[Session]):
 
         """
 
-        result = await self.execute(
+        if execution_options:
+            execution_options = util.immutabledict(execution_options).union(
+                _EXECUTE_OPTIONS
+            )
+        else:
+            execution_options = _EXECUTE_OPTIONS
+
+        result = await greenlet_spawn(
+            self.sync_session.scalar,
             statement,
             params=params,
             execution_options=execution_options,
             bind_arguments=bind_arguments,
             **kw,
         )
-        return result.scalar()
+        return result
 
     async def scalars(
         self,
         statement: Executable,
         params: Optional[_CoreSingleExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -383,6 +395,7 @@ class AsyncSession(ReversibleProxy[Session]):
         self,
         statement: Executable,
         params: Optional[_CoreAnyExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -414,6 +427,7 @@ class AsyncSession(ReversibleProxy[Session]):
         self,
         statement: Executable,
         params: Optional[_CoreSingleExecuteParams] = None,
+        *,
         execution_options: _ExecuteOptionsParameter = util.EMPTY_DICT,
         bind_arguments: Optional[_BindArguments] = None,
         **kw: Any,
@@ -1277,7 +1291,7 @@ class async_sessionmaker:
         class_: Type[AsyncSession] = AsyncSession,
         autoflush: bool = True,
         expire_on_commit: bool = True,
-        info: Optional[Dict[Any, Any]] = None,
+        info: Optional[_InfoType] = None,
         **kw: Any,
     ):
         r"""Construct a new :class:`.async_sessionmaker`.