Fixed regression where the :meth:`_engine.CursorResult.fetchmany` method
would fail to autoclose a server-side cursor (i.e. when ``stream_results``
or ``yield_per`` is in use, either Core or ORM oriented results) when the
results were fully exhausted.
All :class:`_result.Result` objects will now consistently raise
:class:`_exc.ResourceClosedError` if they are used after a hard close,
which includes the "hard close" that occurs after calling "single row or
value" methods like :meth:`_result.Result.first` and
:meth:`_result.Result.scalar`. This was already the behavior of the most
common class of result objects returned for Core statement executions, i.e.
those based on :class:`_engine.CursorResult`, so this behavior is not new.
However, the change has been extended to properly accommodate for the ORM
"filtering" result objects returned when using 2.0 style ORM queries,
which would previously behave in "soft closed" style of returning empty
results, or wouldn't actually "soft close" at all and would continue
yielding from the underlying cursor.
As part of this change, also added :meth:`_result.Result.close` to the base
:class:`_result.Result` class and implemented it for the filtered result
implementations that are used by the ORM, so that it is possible to call
the :meth:`_engine.CursorResult.close` method on the underlying
:class:`_engine.CursorResult` when the the ``yield_per`` execution option
is in use to close a server side cursor before remaining ORM results have
been fetched. This was again already available for Core result sets but the
change makes it available for 2.0 style ORM results as well.
Fixes: #7274
Change-Id: Id4acdfedbcab891582a7f8edd2e2e7d20d868e53
(cherry picked from commit
33824a9c06ca555ad208a9925bc7b40fe489fc72)
--- /dev/null
+.. change::
+ :tags: bug, core, regression
+ :tickets: 7274
+ :versions: 2.0.0b1
+
+ Fixed regression where the :meth:`_engine.CursorResult.fetchmany` method
+ would fail to autoclose a server-side cursor (i.e. when ``stream_results``
+ or ``yield_per`` is in use, either Core or ORM oriented results) when the
+ results were fully exhausted.
+
+.. change::
+ :tags: bug, orm
+ :tickets: 7274
+ :versions: 2.0.0b1
+
+ All :class:`_result.Result` objects will now consistently raise
+ :class:`_exc.ResourceClosedError` if they are used after a hard close,
+ which includes the "hard close" that occurs after calling "single row or
+ value" methods like :meth:`_result.Result.first` and
+ :meth:`_result.Result.scalar`. This was already the behavior of the most
+ common class of result objects returned for Core statement executions, i.e.
+ those based on :class:`_engine.CursorResult`, so this behavior is not new.
+ However, the change has been extended to properly accommodate for the ORM
+ "filtering" result objects returned when using 2.0 style ORM queries,
+ which would previously behave in "soft closed" style of returning empty
+ results, or wouldn't actually "soft close" at all and would continue
+ yielding from the underlying cursor.
+
+ As part of this change, also added :meth:`_result.Result.close` to the base
+ :class:`_result.Result` class and implemented it for the filtered result
+ implementations that are used by the ORM, so that it is possible to call
+ the :meth:`_engine.CursorResult.close` method on the underlying
+ :class:`_engine.CursorResult` when the the ``yield_per`` execution option
+ is in use to close a server side cursor before remaining ORM results have
+ been fetched. This was again already available for Core result sets but the
+ change makes it available for 2.0 style ORM results as well.
+
)
def _buffer_rows(self, result, dbapi_cursor):
+ """this is currently used only by fetchone()."""
+
size = self._bufsize
try:
if size < 1:
lb = len(buf)
if size > lb:
try:
- buf.extend(dbapi_cursor.fetchmany(size - lb))
+ new = dbapi_cursor.fetchmany(size - lb)
except BaseException as e:
self.handle_exception(result, dbapi_cursor, e)
+ else:
+ if not new:
+ result._soft_close()
+ else:
+ buf.extend(new)
result = buf[0:size]
self._rowbuffer = collections.deque(buf[size:])
"""
-
if (not hard and self._soft_closed) or (hard and self.closed):
return
def _soft_close(self, hard=False):
raise NotImplementedError()
+ def close(self):
+ """close this :class:`_result.Result`.
+
+ The behavior of this method is implementation specific, and is
+ not implemented by default. The method should generally end
+ the resources in use by the result object and also cause any
+ subsequent iteration or row fetching to raise
+ :class:`.ResourceClosedError`.
+
+ .. versionadded:: 1.4.27 - ``.close()`` was previously not generally
+ available for all :class:`_result.Result` classes, instead only
+ being available on the :class:`_engine.CursorResult` returned for
+ Core statement executions. As most other result objects, namely the
+ ones used by the ORM, are proxying a :class:`_engine.CursorResult`
+ in any case, this allows the underlying cursor result to be closed
+ from the outside facade for the case when the ORM query is using
+ the ``yield_per`` execution option where it does not immediately
+ exhaust and autoclose the database cursor.
+
+ """
+ self._soft_close(hard=True)
+
@_generative
def yield_per(self, num):
"""Configure the row-fetching strategy to fetch num rows at a time.
"""
+ _hard_closed = False
+
def __init__(
self,
cursor_metadata,
self.raw = raw
self._source_supports_scalars = _source_supports_scalars
- def _soft_close(self, **kw):
+ def _soft_close(self, hard=False, **kw):
+ if hard:
+ self._hard_closed = True
+ if self.raw is not None:
+ self.raw._soft_close(hard=hard, **kw)
self.iterator = iter([])
+ self._reset_memoizations()
+
+ def _raise_hard_closed(self):
+ raise exc.ResourceClosedError("This result object is closed.")
def _raw_row_iterator(self):
return self.iterator
def _fetchiter_impl(self):
+ if self._hard_closed:
+ self._raise_hard_closed()
return self.iterator
def _fetchone_impl(self, hard_close=False):
+ if self._hard_closed:
+ self._raise_hard_closed()
+
row = next(self.iterator, _NO_ROW)
if row is _NO_ROW:
self._soft_close(hard=hard_close)
return row
def _fetchall_impl(self):
+ if self._hard_closed:
+ self._raise_hard_closed()
+
try:
return list(self.iterator)
finally:
self._soft_close()
def _fetchmany_impl(self, size=None):
+ if self._hard_closed:
+ self._raise_hard_closed()
+
return list(itertools.islice(self.iterator, 0, size))
self._yield_per = num
self.iterator = itertools.chain.from_iterable(self.chunks(num))
+ def _soft_close(self, **kw):
+ super(ChunkedIteratorResult, self)._soft_close(**kw)
+ self.chunks = lambda size: []
+
def _fetchmany_impl(self, size=None):
if self.dynamic_yield_per:
self.iterator = itertools.chain.from_iterable(self.chunks(size))
*[r._attributes for r in results]
)
- def close(self):
- self._soft_close(hard=True)
-
- def _soft_close(self, hard=False):
+ def _soft_close(self, hard=False, **kw):
for r in self._results:
- r._soft_close(hard=hard)
+ r._soft_close(hard=hard, **kw)
if hard:
self.closed = True
row = result.first()
eq_(row, (1, 1, 1))
- eq_(result.all(), [])
+ # note this is a behavior change in 1.4.27 due to
+ # adding a real result.close() to Result, previously this would
+ # return an empty list. this is already the
+ # behavior with CursorResult, but was mis-implemented for
+ # other non-cursor result sets.
+ assert_raises(exc.ResourceClosedError, result.all)
def test_one_unique(self):
# assert that one() counts rows after uniqueness has been applied.
eq_(result.scalar(), 1)
- eq_(result.all(), [])
+ # note this is a behavior change in 1.4.27 due to
+ # adding a real result.close() to Result, previously this would
+ # return an empty list. this is already the
+ # behavior with CursorResult, but was mis-implemented for
+ # other non-cursor result sets.
+ assert_raises(exc.ResourceClosedError, result.all)
def test_partition(self):
result = self._fixture()
from sqlalchemy.testing.assertions import assert_raises
from sqlalchemy.testing.assertions import assert_raises_message
from sqlalchemy.testing.assertions import eq_
+from sqlalchemy.testing.assertions import expect_raises
from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing.assertions import is_not_none
from sqlalchemy.testing.assertsql import CompiledSQL
run_setup_mappers = "each"
run_inserts = "each"
+ __backend__ = True
+
def _eagerload_mappings(self, addresses_lazy=True, user_lazy=True):
User, Address = self.classes("User", "Address")
users, addresses = self.tables("users", "addresses")
except StopIteration:
pass
+ def test_we_can_close_cursor(self):
+ """test new usecase close() added along with #7274"""
+ self._eagerload_mappings()
+
+ User = self.classes.User
+
+ sess = fixture_session()
+
+ stmt = select(User).execution_options(yield_per=15)
+ result = sess.execute(stmt)
+
+ with mock.patch.object(result.raw, "_soft_close") as mock_close:
+ two_results = result.fetchmany(2)
+ eq_(len(two_results), 2)
+
+ eq_(mock_close.mock_calls, [])
+
+ result.close()
+
+ eq_(mock_close.mock_calls, [mock.call(hard=True)])
+
+ with expect_raises(sa.exc.ResourceClosedError):
+ result.fetchmany(10)
+
+ with expect_raises(sa.exc.ResourceClosedError):
+ result.fetchone()
+
+ with expect_raises(sa.exc.ResourceClosedError):
+ result.all()
+
+ result.close()
+
+ @testing.combinations("fetchmany", "fetchone", "fetchall")
+ def test_cursor_is_closed_on_exhausted(self, fetch_method):
+ """test #7274"""
+ self._eagerload_mappings()
+
+ User = self.classes.User
+
+ sess = fixture_session()
+
+ stmt = select(User).execution_options(yield_per=15)
+ result = sess.execute(stmt)
+
+ with mock.patch.object(result.raw, "_soft_close") as mock_close:
+ # call assertions are implementation specific.
+ # test needs that _soft_close called at least once and without
+ # the hard=True flag
+ if fetch_method == "fetchmany":
+ while True:
+ buf = result.fetchmany(2)
+ if not buf:
+ break
+ eq_(mock_close.mock_calls, [mock.call()])
+ elif fetch_method == "fetchall":
+ eq_(len(result.all()), 4)
+ eq_(
+ mock_close.mock_calls, [mock.call(), mock.call(hard=False)]
+ )
+ elif fetch_method == "fetchone":
+ while True:
+ row = result.fetchone()
+ if row is None:
+ break
+ eq_(
+ mock_close.mock_calls, [mock.call(), mock.call(hard=False)]
+ )
+ else:
+ assert False
+
+ # soft closed, we can still get an empty result
+ eq_(result.all(), [])
+
+ # real closed
+ result.close()
+ assert_raises(sa.exc.ResourceClosedError, result.all)
+
def test_yield_per_and_execution_options_legacy(self):
self._eagerload_mappings()
# buffer of 98, plus buffer of 99 - 89, 10 rows
eq_(len(result.cursor_strategy._rowbuffer), 10)
+ @testing.combinations(True, False, argnames="close_on_init")
+ @testing.combinations(
+ "fetchone", "fetchmany", "fetchall", argnames="fetch_style"
+ )
+ def test_buffered_fetch_auto_soft_close(
+ self, connection, close_on_init, fetch_style
+ ):
+ """test #7274"""
+
+ table = self.tables.test
+
+ connection.execute(
+ table.insert(),
+ [{"x": i, "y": "t_%d" % i} for i in range(15, 30)],
+ )
+
+ result = connection.execute(table.select().limit(15))
+ assert isinstance(result.cursor_strategy, _cursor.CursorFetchStrategy)
+
+ if close_on_init:
+ # close_on_init - the initial buffering will exhaust the cursor,
+ # should soft close immediately
+ result = result.yield_per(30)
+ else:
+ # not close_on_init - soft close will occur after fetching an
+ # empty buffer
+ result = result.yield_per(5)
+ assert isinstance(
+ result.cursor_strategy, _cursor.BufferedRowCursorFetchStrategy
+ )
+
+ with mock.patch.object(result, "_soft_close") as soft_close:
+ if fetch_style == "fetchone":
+ while True:
+ row = result.fetchone()
+
+ if row:
+ eq_(soft_close.mock_calls, [])
+ else:
+ # fetchone() is also used by first(), scalar()
+ # and one() which want to embed a hard close in one
+ # step
+ eq_(soft_close.mock_calls, [mock.call(hard=False)])
+ break
+ elif fetch_style == "fetchmany":
+ while True:
+ rows = result.fetchmany(5)
+
+ if rows:
+ eq_(soft_close.mock_calls, [])
+ else:
+ eq_(soft_close.mock_calls, [mock.call()])
+ break
+ elif fetch_style == "fetchall":
+ rows = result.fetchall()
+
+ eq_(soft_close.mock_calls, [mock.call()])
+ else:
+ assert False
+
+ result.close()
+
def test_buffered_fetchmany_yield_per_all(self, connection):
table = self.tables.test
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
