Skip to content

Cursor result

A simple wrapper to hold cursor results after the cursor is closed.

This class provides a cursor-compatible interface for accessing query results after the actual database cursor has been closed.

Source code in pum/sql_content.py
class CursorResult:
    """A simple wrapper to hold cursor results after the cursor is closed.

    This class provides a cursor-compatible interface for accessing query results
    after the actual database cursor has been closed.
    """

    def __init__(
        self, results: list | None = None, description: Any | None = None, rowcount: int = 0
    ):
        self._pum_results = results
        self._pum_description = description
        self._pum_rowcount = rowcount
        self._pum_index = 0

    @property
    def description(self):
        """Return the column description (compatible with cursor.description)."""
        return self._pum_description

    @property
    def rowcount(self):
        """Return the number of rows (compatible with cursor.rowcount)."""
        return self._pum_rowcount

    def fetchall(self):
        """Return all results (compatible with cursor.fetchall())."""
        return self._pum_results if self._pum_results is not None else []

    def fetchone(self):
        """Return the next result (compatible with cursor.fetchone())."""
        if self._pum_results is None or self._pum_index >= len(self._pum_results):
            return None
        result = self._pum_results[self._pum_index]
        self._pum_index += 1
        return result

    def fetchmany(self, size: int = 1):
        """Return the next `size` results (compatible with cursor.fetchmany())."""
        if self._pum_results is None:
            return []
        results = self._pum_results[self._pum_index : self._pum_index + size]
        self._pum_index += len(results)
        return results

description property

description

Return the column description (compatible with cursor.description).

rowcount property

rowcount

Return the number of rows (compatible with cursor.rowcount).

fetchall

fetchall()

Return all results (compatible with cursor.fetchall()).

Source code in pum/sql_content.py
def fetchall(self):
    """Return all results (compatible with cursor.fetchall())."""
    return self._pum_results if self._pum_results is not None else []

fetchmany

fetchmany(size: int = 1)

Return the next size results (compatible with cursor.fetchmany()).

Source code in pum/sql_content.py
def fetchmany(self, size: int = 1):
    """Return the next `size` results (compatible with cursor.fetchmany())."""
    if self._pum_results is None:
        return []
    results = self._pum_results[self._pum_index : self._pum_index + size]
    self._pum_index += len(results)
    return results

fetchone

fetchone()

Return the next result (compatible with cursor.fetchone()).

Source code in pum/sql_content.py
def fetchone(self):
    """Return the next result (compatible with cursor.fetchone())."""
    if self._pum_results is None or self._pum_index >= len(self._pum_results):
        return None
    result = self._pum_results[self._pum_index]
    self._pum_index += 1
    return result