def _get_connection()

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

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

    def _get_connection(self):
        host = self.configuration["host"]

        scheme = self.configuration.get("http_scheme", "https")

        # if path is set but is missing initial slash, append it
        path = self.configuration.get("http_path", "")
        if path and path[0] != "/":
            path = "/" + path

        # if port is set prepend colon
        port = self.configuration.get("port", "")
        if port:
            port = ":" + str(port)

        http_uri = "{}://{}{}{}".format(scheme, host, port, path)

        # create transport
        transport = THttpClient.THttpClient(http_uri)

        # if username or password is set, add Authorization header
        username = self.configuration.get("username", "")
        password = self.configuration.get("http_password", "")
        if username or password:
            auth = base64.b64encode(username.encode("ascii") + b":" + password.encode("ascii"))
            transport.setCustomHeaders({"Authorization": "Basic " + auth.decode()})

        # create connection
        connection = hive.connect(thrift_transport=transport)

        return connection