From 57f684b4b4a661c78de0b5953603984714f01e0b Mon Sep 17 00:00:00 2001
From: Mike Bayer <mike_mp@zzzcomputing.com>
Date: Mon, 5 Jan 2015 21:38:19 -0500
Subject: [PATCH] - Fixed bug where if an exception were thrown at the start of
 a :class:`.Query` before it fetched results, particularly when row processors
 can't be formed, the cursor would stay open with results pending and not
 actually be closed.  This is typically only an issue on an interpreter like
 Pypy where the cursor isn't immediately GC'ed, and can in some circumstances
 lead to transactions/ locks being open longer than is desirable. fixes #3285

---
 doc/build/changelog/changelog_09.rst | 13 ++++++
 lib/sqlalchemy/orm/loading.py        | 66 +++++++++++++++-------------
 test/orm/test_loading.py             | 33 ++++++++++++--
 3 files changed, 78 insertions(+), 34 deletions(-)

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index e0f46eb66f..81a26d1872 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -14,6 +14,19 @@
 .. changelog::
     :version: 0.9.9
 
+    .. change::
+        :tags: bug, orm, pypy
+        :versions: 1.0.0
+        :tickets: 3285
+
+        Fixed bug where if an exception were thrown at the start of a
+        :class:`.Query` before it fetched results, particularly when
+        row processors can't be formed, the cursor would stay open with
+        results pending and not actually be closed.  This is typically only
+        an issue on an interpreter like Pypy where the cursor isn't
+        immediately GC'ed, and can in some circumstances lead to transactions/
+        locks being open longer than is desirable.
+
     .. change::
         :tags: change, mysql
         :versions: 1.0.0
diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index 380afcdc72..fdc7875458 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -42,41 +42,45 @@ def instances(query, cursor, context):
             def filter_fn(row):
                 return tuple(fn(x) for x, fn in zip(row, filter_fns))
 
-    (process, labels) = \
-        list(zip(*[
-            query_entity.row_processor(query,
-                                       context, cursor)
-            for query_entity in query._entities
-        ]))
-
-    if not single_entity:
-        keyed_tuple = util.lightweight_named_tuple('result', labels)
-
-    while True:
-        context.partials = {}
-
-        if query._yield_per:
-            fetch = cursor.fetchmany(query._yield_per)
-            if not fetch:
-                break
-        else:
-            fetch = cursor.fetchall()
+    try:
+        (process, labels) = \
+            list(zip(*[
+                query_entity.row_processor(query,
+                                           context, cursor)
+                for query_entity in query._entities
+            ]))
+
+        if not single_entity:
+            keyed_tuple = util.lightweight_named_tuple('result', labels)
+
+        while True:
+            context.partials = {}
+
+            if query._yield_per:
+                fetch = cursor.fetchmany(query._yield_per)
+                if not fetch:
+                    break
+            else:
+                fetch = cursor.fetchall()
 
-        if single_entity:
-            proc = process[0]
-            rows = [proc(row) for row in fetch]
-        else:
-            rows = [keyed_tuple([proc(row) for proc in process])
-                    for row in fetch]
+            if single_entity:
+                proc = process[0]
+                rows = [proc(row) for row in fetch]
+            else:
+                rows = [keyed_tuple([proc(row) for proc in process])
+                        for row in fetch]
 
-        if filtered:
-            rows = util.unique_list(rows, filter_fn)
+            if filtered:
+                rows = util.unique_list(rows, filter_fn)
 
-        for row in rows:
-            yield row
+            for row in rows:
+                yield row
 
-        if not query._yield_per:
-            break
+            if not query._yield_per:
+                break
+    except Exception as err:
+        cursor.close()
+        util.raise_from_cause(err)
 
 
 @util.dependencies("sqlalchemy.orm.query")
diff --git a/test/orm/test_loading.py b/test/orm/test_loading.py
index 97c08ea290..f86477ec29 100644
--- a/test/orm/test_loading.py
+++ b/test/orm/test_loading.py
@@ -1,13 +1,40 @@
 from . import _fixtures
 from sqlalchemy.orm import loading, Session, aliased
-from sqlalchemy.testing.assertions import eq_
+from sqlalchemy.testing.assertions import eq_, assert_raises
 from sqlalchemy.util import KeyedTuple
-
-# class InstancesTest(_fixtures.FixtureTest):
+from sqlalchemy.testing import mock
 # class GetFromIdentityTest(_fixtures.FixtureTest):
 # class LoadOnIdentTest(_fixtures.FixtureTest):
 # class InstanceProcessorTest(_fixture.FixtureTest):
 
+
+class InstancesTest(_fixtures.FixtureTest):
+    run_setup_mappers = 'once'
+    run_inserts = 'once'
+    run_deletes = None
+
+    @classmethod
+    def setup_mappers(cls):
+        cls._setup_stock_mapping()
+
+    def test_cursor_close_w_failed_rowproc(self):
+        User = self.classes.User
+        s = Session()
+
+        q = s.query(User)
+
+        ctx = q._compile_context()
+        cursor = mock.Mock()
+        q._entities = [
+            mock.Mock(row_processor=mock.Mock(side_effect=Exception("boom")))
+        ]
+        assert_raises(
+            Exception,
+            list, loading.instances(q, cursor, ctx)
+        )
+        assert cursor.close.called, "Cursor wasn't closed"
+
+
 class MergeResultTest(_fixtures.FixtureTest):
     run_setup_mappers = 'once'
     run_inserts = 'once'
-- 
2.47.3