def run_query()

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

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

    def run_query(self, query, user):
        query = parse_query(query)

        if not isinstance(query, dict):
            raise QueryParseError(
                "Query should be a YAML object describing the URL to query."
            )

        if "url" not in query:
            raise QueryParseError("Query must include 'url' option.")


        method = query.get("method", "get")
        request_options = project(query, ("params", "headers", "data", "auth", "json"))

        fields = query.get("fields")
        path = query.get("path")

        if isinstance(request_options.get("auth", None), list):
            request_options["auth"] = tuple(request_options["auth"])
        elif self.configuration.get("username") or self.configuration.get("password"):
            request_options["auth"] = (
                self.configuration.get("username"),
                self.configuration.get("password"),
            )

        if method not in ("get", "post"):
            raise QueryParseError("Only GET or POST methods are allowed.")

        if fields and not isinstance(fields, list):
            raise QueryParseError("'fields' needs to be a list.")

        response, error = self.get_response(
            query["url"], http_method=method, **request_options
        )

        if error is not None:
            return None, error

        data = json_dumps(parse_json(response.json(), path, fields))

        if data:
            return data, None
        else:
            return None, "Got empty response from '{}'.".format(query["url"])