def upload()

in rules_jvm_export/jvm_export/support/deploy.py [0:0]

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

def upload(url, curl_opts, local_fn, remote_fn):
    print(f"  publishing {remote_fn}")
    u = urlparse(url)
    if u.scheme == "file":
        destination = Path(u.path).expanduser() / remote_fn
        destination.parent.mkdir(parents=True, exist_ok=True)
        shutil.copyfile(src=local_fn, dst=destination)
    else:
        failed = False
        try:
            upload_status_code = (
                sp.check_output(
                    [
                        "curl",
                        "--silent",
                        "--output",
                        "/dev/stderr",
                        "--write-out",
                        "%{http_code}",
                        *curl_opts,
                        "--upload-file",
                        local_fn,
                        urljoin(url, remote_fn),
                    ]
                )
                .decode()
                .strip()
            )
        except Exception:
            # Catch the exception to avoid credential getting logged
            failed = True
        if failed:
            raise Exception("upload of {} to {} failed".format(local_fn, urljoin(url, remote_fn)))
        if upload_status_code not in {"200", "201"}:
            raise Exception(
                "upload of {} failed, got HTTP status code {}".format(
                    local_fn, upload_status_code
                )
            )