Support result.close() for all iterator patterns

This change contains new features for 2.0 only as well as some
behaviors that will be backported to 1.4.

For 1.4 and 2.0:

Fixed issue where the underlying DBAPI cursor would not be closed when
using :class:`_orm.Query` with :meth:`_orm.Query.yield_per` and direct
iteration, if a user-defined exception case were raised within the
iteration process, interrupting the iterator. This would lead to the usual
MySQL-related issues with server side cursors out of sync.

For 1.4 only:

A similar scenario can occur when using :term:`2.x` executions with direct
use of :class:`.Result`, in that case the end-user code has access to the
:class:`.Result` itself and should call :meth:`.Result.close` directly.
Version 2.0 will feature context-manager calling patterns to address this
use case.  However within the 1.4 scope, ensured that ``.close()`` methods
are available on all :class:`.Result` implementations including
:class:`.ScalarResult`, :class:`.MappingResult`.

For 2.0 only:

To better support the use case of iterating :class:`.Result` and
:class:`.AsyncResult` objects where user-defined exceptions may interrupt
the iteration, both objects as well as variants such as
:class:`.ScalarResult`, :class:`.MappingResult`,
:class:`.AsyncScalarResult`, :class:`.AsyncMappingResult` now support
context manager usage, where the result will be closed at the end of
iteration.

Corrected various typing issues within the engine and async engine
packages.

Fixes: #8710
Change-Id: I3166328bfd3900957eb33cbf1061d0495c9df670

1 files changed, 86 insertions, 0 deletions

diff --git a/test/ext/mypy/plain_files/engines.py b/test/ext/mypy/plain_files/engines.py
new file mode 100644
index 000000000..c920ad55d
--- /dev/null
+++ b/test/ext/mypy/plain_files/engines.py
@@ -0,0 +1,86 @@
+from sqlalchemy import create_engine
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+
+def regular() -> None:
+
+    e = create_engine("sqlite://")
+
+    # EXPECTED_TYPE: Engine
+    reveal_type(e)
+
+    with e.connect() as conn:
+
+        # EXPECTED_TYPE: Connection
+        reveal_type(conn)
+
+        result = conn.execute(text("select * from table"))
+
+        # EXPECTED_TYPE: CursorResult[Any]
+        reveal_type(result)
+
+    with e.begin() as conn:
+
+        # EXPECTED_TYPE: Connection
+        reveal_type(conn)
+
+        result = conn.execute(text("select * from table"))
+
+        # EXPECTED_TYPE: CursorResult[Any]
+        reveal_type(result)
+
+
+async def asyncio() -> None:
+    e = create_async_engine("sqlite://")
+
+    # EXPECTED_TYPE: AsyncEngine
+    reveal_type(e)
+
+    async with e.connect() as conn:
+
+        # EXPECTED_TYPE: AsyncConnection
+        reveal_type(conn)
+
+        result = await conn.execute(text("select * from table"))
+
+        # EXPECTED_TYPE: CursorResult[Any]
+        reveal_type(result)
+
+        # stream with direct await
+        async_result = await conn.stream(text("select * from table"))
+
+        # EXPECTED_TYPE: AsyncResult[Any]
+        reveal_type(async_result)
+
+        # stream with context manager
+        async with conn.stream(
+            text("select * from table")
+        ) as ctx_async_result:
+            # EXPECTED_TYPE: AsyncResult[Any]
+            reveal_type(ctx_async_result)
+
+        # stream_scalars with direct await
+        async_scalar_result = await conn.stream_scalars(
+            text("select * from table")
+        )
+
+        # EXPECTED_TYPE: AsyncScalarResult[Any]
+        reveal_type(async_scalar_result)
+
+        # stream_scalars with context manager
+        async with conn.stream_scalars(
+            text("select * from table")
+        ) as ctx_async_scalar_result:
+            # EXPECTED_TYPE: AsyncScalarResult[Any]
+            reveal_type(ctx_async_scalar_result)
+
+    async with e.begin() as conn:
+
+        # EXPECTED_TYPE: AsyncConnection
+        reveal_type(conn)
+
+        result = await conn.execute(text("select * from table"))
+
+        # EXPECTED_TYPE: CursorResult[Any]
+        reveal_type(result)