(
self,
result: CursorResult[Any],
dbapi_cursor: DBAPICursor,
size: Optional[int] = None,
)
| 1438 | return self._rowbuffer.popleft() |
| 1439 | |
| 1440 | def fetchmany( |
| 1441 | self, |
| 1442 | result: CursorResult[Any], |
| 1443 | dbapi_cursor: DBAPICursor, |
| 1444 | size: Optional[int] = None, |
| 1445 | ) -> Any: |
| 1446 | if size is None: |
| 1447 | return self.fetchall(result, dbapi_cursor) |
| 1448 | |
| 1449 | rb = self._rowbuffer |
| 1450 | lb = len(rb) |
| 1451 | close = False |
| 1452 | if size > lb: |
| 1453 | try: |
| 1454 | new = dbapi_cursor.fetchmany(size - lb) |
| 1455 | except BaseException as e: |
| 1456 | self.handle_exception(result, dbapi_cursor, e) |
| 1457 | else: |
| 1458 | if not new: |
| 1459 | # defer closing since it may clear the row buffer |
| 1460 | close = True |
| 1461 | else: |
| 1462 | rb.extend(new) |
| 1463 | |
| 1464 | res = [rb.popleft() for _ in range(min(size, len(rb)))] |
| 1465 | if close: |
| 1466 | result._soft_close() |
| 1467 | return res |
| 1468 | |
| 1469 | def fetchall( |
| 1470 | self, result: CursorResult[Any], dbapi_cursor: DBAPICursor |
nothing calls this directly
no test coverage detected