def run_query()

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

• 38 lines of code
• 11 McCabe index (conditional complexity)

    def run_query(self, query, user):
        engine = None
        try:
            engine = self._connect()

            if not query.endswith(";"):
                query = query + ";"

            result = engine.execute(query)

            columns = []
            rows = []

            # When running a count query it returns the value as a string, in which case
            # we transform it into a dictionary to be the same as regular queries.
            if isinstance(result, str):
                # when count < scanned_count, dql returns a string with number of rows scanned
                value = result.split(" (")[0]
                if value:
                    value = int(value)
                result = [{"value": value}]

            for item in result:
                if not columns:
                    for k, v in item.items():
                        columns.append(
                            {
                                "name": k,
                                "friendly_name": k,
                                "type": types_map.get(str(type(v)).upper(), None),
                            }
                        )
                rows.append(item)

            data = {"columns": columns, "rows": rows}
            json_data = json_dumps(data)
            error = None
        except ParseException as e:
            error = "Error parsing query at line {} (column {}):\n{}".format(
                e.lineno, e.column, e.line
            )
            json_data = None
        except (KeyboardInterrupt, JobTimeoutException):
            if engine and engine.connection:
                engine.connection.cancel()
            raise

        return json_data, error