def request()

in tools/cli/wsk/wskutil.py [0:0]

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

def request(method, urlString, body = '', headers = {}, auth = None, verbose = False, https_proxy = os.getenv('https_proxy', None), timeout = 60):
    url = urlparse(urlString)
    if url.scheme == 'http':
        conn = HTTPConnection(url.netloc, timeout = timeout)
    else:
        if httpRequestProps['secure'] or not hasattr(ssl, '_create_unverified_context'):
            conn = HTTPSConnection(url.netloc if https_proxy is None else https_proxy, timeout = timeout)
        else:
            conn = HTTPSConnection(url.netloc if https_proxy is None else https_proxy, context=ssl._create_unverified_context(), timeout = timeout)
        if https_proxy:
            conn.set_tunnel(url.netloc)

    if auth is not None:
        auth = base64.b64encode(auth.encode()).decode()
        headers['Authorization'] = 'Basic %s' % auth

    if verbose:
        print('========')
        print('REQUEST:')
        print('%s %s' % (method, urlString))
        print('Headers sent:')
        print(getPrettyJson(headers))
        if body != '':
            print('Body sent:')
            print(body)

    try:
        conn.request(method, urlString, body, headers)
        res = conn.getresponse()
        body = ''
        try:
            body = res.read()
        except IncompleteRead as e:
            body = e.partial

        # patch the read to return just the body since the normal read
        # can only be done once
        res.read = lambda: body

        if verbose:
            print('--------')
            print('RESPONSE:')
            print('Got response with code %s' % res.status)
            print('Body received:')
            print(res.read())
            print('========')
        return res
    except socket.timeout:
        return ErrorResponse(status = 500, error = 'request timed out at %d seconds' % timeout)
    except Exception as e:
        return ErrorResponse(status = 500, error = str(e))