A cursor strategy that buffers rows fully upon creation. Used for operations where a result is to be delivered after the database conversation can not be continued, such as MSSQL INSERT...OUTPUT after an autocommit.
| 1365 | |
| 1366 | |
| 1367 | class FullyBufferedCursorFetchStrategy(CursorFetchStrategy): |
| 1368 | """A cursor strategy that buffers rows fully upon creation. |
| 1369 | |
| 1370 | Used for operations where a result is to be delivered |
| 1371 | after the database conversation can not be continued, |
| 1372 | such as MSSQL INSERT...OUTPUT after an autocommit. |
| 1373 | |
| 1374 | """ |
| 1375 | |
| 1376 | __slots__ = ("_rowbuffer", "alternate_cursor_description") |
| 1377 | |
| 1378 | def __init__( |
| 1379 | self, |
| 1380 | dbapi_cursor: Optional[DBAPICursor], |
| 1381 | alternate_description: Optional[_DBAPICursorDescription] = None, |
| 1382 | initial_buffer: Optional[Iterable[Any]] = None, |
| 1383 | ): |
| 1384 | self.alternate_cursor_description = alternate_description |
| 1385 | if initial_buffer is not None: |
| 1386 | self._rowbuffer = collections.deque(initial_buffer) |
| 1387 | else: |
| 1388 | assert dbapi_cursor is not None |
| 1389 | self._rowbuffer = collections.deque(dbapi_cursor.fetchall()) |
| 1390 | |
| 1391 | def yield_per( |
| 1392 | self, result: CursorResult[Any], dbapi_cursor: DBAPICursor, num: int |
| 1393 | ) -> Any: |
| 1394 | pass |
| 1395 | |
| 1396 | def soft_close( |
| 1397 | self, result: CursorResult[Any], dbapi_cursor: Optional[DBAPICursor] |
| 1398 | ) -> None: |
| 1399 | self._rowbuffer.clear() |
| 1400 | super().soft_close(result, dbapi_cursor) |
| 1401 | |
| 1402 | def hard_close( |
| 1403 | self, result: CursorResult[Any], dbapi_cursor: Optional[DBAPICursor] |
| 1404 | ) -> None: |
| 1405 | self._rowbuffer.clear() |
| 1406 | super().hard_close(result, dbapi_cursor) |
| 1407 | |
| 1408 | def fetchone( |
| 1409 | self, |
| 1410 | result: CursorResult[Any], |
| 1411 | dbapi_cursor: DBAPICursor, |
| 1412 | hard_close: bool = False, |
| 1413 | ) -> Any: |
| 1414 | if self._rowbuffer: |
| 1415 | return self._rowbuffer.popleft() |
| 1416 | else: |
| 1417 | result._soft_close(hard=hard_close) |
| 1418 | return None |
| 1419 | |
| 1420 | def fetchmany( |
| 1421 | self, |
| 1422 | result: CursorResult[Any], |
| 1423 | dbapi_cursor: DBAPICursor, |
| 1424 | size: Optional[int] = None, |
no outgoing calls
no test coverage detected