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.
| 1479 | |
| 1480 | |
| 1481 | class FullyBufferedCursorFetchStrategy(CursorFetchStrategy): |
| 1482 | """A cursor strategy that buffers rows fully upon creation. |
| 1483 | |
| 1484 | Used for operations where a result is to be delivered |
| 1485 | after the database conversation can not be continued, |
| 1486 | such as MSSQL INSERT...OUTPUT after an autocommit. |
| 1487 | |
| 1488 | """ |
| 1489 | |
| 1490 | __slots__ = ("_rowbuffer", "alternate_cursor_description") |
| 1491 | |
| 1492 | def __init__( |
| 1493 | self, |
| 1494 | dbapi_cursor: Optional[DBAPICursor], |
| 1495 | alternate_description: Optional[_DBAPICursorDescription] = None, |
| 1496 | initial_buffer: Optional[Iterable[Any]] = None, |
| 1497 | ): |
| 1498 | self.alternate_cursor_description = alternate_description |
| 1499 | if initial_buffer is not None: |
| 1500 | self._rowbuffer = collections.deque(initial_buffer) |
| 1501 | else: |
| 1502 | assert dbapi_cursor is not None |
| 1503 | self._rowbuffer = collections.deque(dbapi_cursor.fetchall()) |
| 1504 | |
| 1505 | def yield_per( |
| 1506 | self, result: CursorResult[Any], dbapi_cursor: DBAPICursor, num: int |
| 1507 | ) -> Any: |
| 1508 | pass |
| 1509 | |
| 1510 | def soft_close( |
| 1511 | self, result: CursorResult[Any], dbapi_cursor: Optional[DBAPICursor] |
| 1512 | ) -> None: |
| 1513 | self._rowbuffer.clear() |
| 1514 | super().soft_close(result, dbapi_cursor) |
| 1515 | |
| 1516 | def hard_close( |
| 1517 | self, result: CursorResult[Any], dbapi_cursor: Optional[DBAPICursor] |
| 1518 | ) -> None: |
| 1519 | self._rowbuffer.clear() |
| 1520 | super().hard_close(result, dbapi_cursor) |
| 1521 | |
| 1522 | def fetchone( |
| 1523 | self, |
| 1524 | result: CursorResult[Any], |
| 1525 | dbapi_cursor: DBAPICursor, |
| 1526 | hard_close: bool = False, |
| 1527 | ) -> Any: |
| 1528 | if self._rowbuffer: |
| 1529 | return self._rowbuffer.popleft() |
| 1530 | else: |
| 1531 | result._soft_close(hard=hard_close) |
| 1532 | return None |
| 1533 | |
| 1534 | def fetchmany( |
| 1535 | self, |
| 1536 | result: CursorResult[Any], |
| 1537 | dbapi_cursor: DBAPICursor, |
| 1538 | size: Optional[int] = None, |
no outgoing calls
no test coverage detected