def upload_single_file_native_imagestore()

in src/sfctl/custom_app.py [0:0]

• 26 lines of code
• 3 McCabe index (conditional complexity)

def upload_single_file_native_imagestore(sesh, endpoint, basename, #pylint: disable=too-many-locals,too-many-arguments
                                         rel_path, single_file, root, target_timeout):
    """
    Used by upload_to_native_imagestore to upload individual files
    of the application package to cluster

    :param sesh: A requests (module) session object.
    :param endpoint: Connection url endpoint for upload requests.
    :param basename: Image store base path.
    :param rel_path: Image store relative directory path.
    :param single_file: Filename.
    :param root: Source directory path.
    :param target_timeout: Time at which timeout would be reached.
    """
    try:
        from urllib.parse import urlparse, urlencode, urlunparse
    except ImportError:
        from urllib import urlencode
        from urlparse import urlparse, urlunparse  # pylint: disable=import-error

    current_time_left = get_timeout_left(target_timeout)   # an int representing seconds

    if current_time_left == 0:
        raise SFCTLInternalException('Upload has timed out. Consider passing a longer '
                                        'timeout duration.')

    url_path = (
        os.path.normpath(os.path.join('ImageStore', basename,
                                        rel_path, single_file))
    ).replace('\\', '/')
    fp_norm = os.path.normpath(os.path.join(root, single_file))
    with open(fp_norm, 'rb') as file_opened:
        url_parsed = list(urlparse(endpoint))
        url_parsed[2] = url_path
        url_parsed[4] = urlencode(
            {'api-version': '6.1',
                'timeout': current_time_left})
        url = urlunparse(url_parsed)

        # timeout is (connect_timeout, read_timeout)
        res = sesh.put(url, data=file_opened,
                        timeout=(get_lesser(60, current_time_left), current_time_left))

        res.raise_for_status()