def run_command()

in dynalab_cli/upload.py [0:0]

• 96 lines of code
• 7 McCabe index (conditional complexity)

    def run_command(self):
        # validate config
        try:
            self.config_handler.validate_config()
        except AssertionError as err:
            print(
                f"Error: {err}.\nPlease fix your config file by",
                "dynalab-cli init --amend",
            )
            exit(1)
        else:
            config = self.config_handler.load_config()
            print("Config file validated")

        # set up exclude files for tarball
        print("Tarballing the project directory...")
        tmp_dir = os.path.join(self.config_handler.dynalab_dir, self.args.name, "tmp")
        os.makedirs(tmp_dir, exist_ok=True)
        exclude_list_file = os.path.join(tmp_dir, "exclude.txt")
        self.config_handler.write_exclude_filelist(
            exclude_list_file, self.args.name, exclude_model=False
        )

        # tarball
        tmp_tarball_dir = tempfile.TemporaryDirectory()
        tarball = os.path.join(tmp_tarball_dir.name, f"{self.args.name}.tar.gz")
        process = subprocess.run(
            ["tar", f"--exclude-from={exclude_list_file}", "-czf", tarball, "."],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            universal_newlines=True,
        )
        if process.returncode != 0:
            raise RuntimeError(
                f"Error in tarballing the current directory {process.stderr}"
            )
        # upload to s3
        print(
            "Uploading files to S3. For large submissions, the progress bar may "
            "hang a while even after uploading reaches 100%. Please do not kill it..."
        )
        url = f"{DYNABENCH_API}/models/upload/s3"

        payload = encoder.MultipartEncoder(
            {
                "name": self.args.name,
                "taskCode": config["task"],
                "tarball": (
                    f"{self.args.name}.tar.gz",
                    open(tarball, "rb"),
                    "application/octet-stream",
                ),
            }
        )
        with tqdm(
            desc="Uploading",
            total=payload.len,
            dynamic_ncols=True,
            unit="B",
            unit_scale=True,
            unit_divisor=1024,
        ) as pbar:
            payload_monitor = encoder.MultipartEncoderMonitor(
                payload, lambda monitor: pbar.update(monitor.bytes_read - pbar.n)
            )
            headers = {
                **AccessToken().get_headers(),
                "Content-Type": payload.content_type,
            }
            r = requests.post(url, data=payload_monitor, headers=headers)

        try:
            r.raise_for_status()
        except requests.exceptions.HTTPError as ex:
            if r.status_code == 429:
                hr_diff, threshold = get_task_submission_limit(config["task"])
                print(
                    f"Failed to submit model {self.args.name} "
                    f"due to submission limit exceeded. No more than {threshold} "
                    f"submissions allowed every {hr_diff} hours for task {config['task']}."
                )
            else:
                print(f"Failed to submit model due to: {ex}")
        except Exception as ex:
            print(f"Failed to submit model due to: {ex}")
        # TODO: show which email address it is: API to fetch email address?
        else:
            print(
                f"Your model {self.args.name} has been uploaded to S3 and "
                f"will be deployed shortly. "
                f"You will get an email notification when your model is available "
                f"on Dynabench."
            )
        finally:
            os.makedirs(self.config_handler.submission_dir, exist_ok=True)
            submission = os.path.join(
                self.config_handler.submission_dir,
                datetime.now().strftime("%b-%d-%Y-%H-%M-%S-")
                + os.path.basename(tarball),
            )
            shutil.move(tarball, submission)
            tmp_tarball_dir.cleanup()
            print(
                f"You can inspect the prepared model submission locally at {submission}"
            )