def fetch_dataframe()

in redshift_connector/cursor.py [0:0]

• 24 lines of code
• 9 McCabe index (conditional complexity)

    def fetch_dataframe(self: "Cursor", num: typing.Optional[int] = None) -> typing.Optional["pandas.DataFrame"]:
        """
        Fetches a user defined number of rows of a query result as a :class:`pandas.DataFrame`.

        Parameters
        ----------
        num : Optional[int] The number of rows to retrieve. If unspecified, all rows will be retrieved

        Returns
        -------
        A `pandas.DataFrame` containing field values making up a row. A column label, derived from the row description of the table, is provided.:typing.Optional["pandas.Dataframe"]
        """
        try:
            import pandas
        except ModuleNotFoundError:
            raise ModuleNotFoundError(MISSING_MODULE_ERROR_MSG.format(module="pandas"))

        columns: typing.Optional[typing.List[typing.Union[str, bytes]]] = None
        try:
            columns = [column[0].decode().lower() for column in self.description]
        except UnicodeError as e:
            warn(
                "Unable to decode column names. Byte values will be used for pandas dataframe column labels.",
                stacklevel=2,
            )
            columns = [column[0].lower() for column in self.description]
        except:
            warn("No row description was found. pandas dataframe will be missing column labels.", stacklevel=2)

        if num:
            fetcheddata: tuple = self.fetchmany(num)
        else:
            fetcheddata = self.fetchall()

        result: typing.List = [tuple(column for column in rows) for rows in fetcheddata]
        if len(result) == 0:
            return None
        return pandas.DataFrame(result, columns=columns)