def _get_statement_result()

in awswrangler/data_api/rds.py [0:0]

• 18 lines of code
• 7 McCabe index (conditional complexity)

    def _get_statement_result(self, request_id: str) -> pd.DataFrame:
        try:
            result = cast("ExecuteStatementResponseTypeDef", self.results.pop(request_id))
        except KeyError as exception:
            raise KeyError(f"Request {request_id} not found in results {self.results}") from exception

        if "records" not in result:
            return pd.DataFrame()

        rows: list[list[Any]] = []
        column_types = [col.get("typeName") for col in result["columnMetadata"]]

        for record in result["records"]:
            row: list[Any] = [
                _connector.DataApiConnector._get_column_value(column, col_type)  # type: ignore[arg-type]
                for column, col_type in zip(record, column_types)
            ]
            rows.append(row)

        column_names: list[str] = [column["name"] for column in result["columnMetadata"]]
        dataframe = pd.DataFrame(rows, columns=column_names)
        return dataframe