def run_query()

in redash/query_runner/vertica.py [0:0]

• 44 lines of code
• 8 McCabe index (conditional complexity)

    def run_query(self, query, user):
        import vertica_python

        if query == "":
            json_data = None
            error = "Query is empty"
            return json_data, error

        connection = None
        try:
            conn_info = {
                "host": self.configuration.get("host", ""),
                "port": self.configuration.get("port", 5433),
                "user": self.configuration.get("user", ""),
                "password": self.configuration.get("password", ""),
                "database": self.configuration.get("database", ""),
                "read_timeout": self.configuration.get("read_timeout", 600),
            }

            if self.configuration.get("connection_timeout"):
                conn_info["connection_timeout"] = self.configuration.get(
                    "connection_timeout"
                )

            connection = vertica_python.connect(**conn_info)
            cursor = connection.cursor()
            logger.debug("Vertica running query: %s", query)
            cursor.execute(query)

            if cursor.description is not None:
                columns_data = [
                    (i[0], types_map.get(i[1], None)) for i in cursor.description
                ]

                columns = self.fetch_columns(columns_data)
                rows = [
                    dict(zip(([c["name"] for c in columns]), r))
                    for r in cursor.fetchall()
                ]

                data = {"columns": columns, "rows": rows}
                json_data = json_dumps(data)
                error = None
            else:
                json_data = None
                error = "No data was returned."

            cursor.close()
        finally:
            if connection:
                connection.close()

        return json_data, error