comparison data_manager/bakta_build_database.py @ 6:97b1b5ad1cda draft default tip

planemo upload for repository https://github.com/galaxyproject/tools-iuc/tree/master/data_managers/data_manager_build_bakta_database commit e0ce56ac52cff0e8f85e546440d28ca46853b11d

5:baceff842902

    def download(self):
        bakta_path = Path(self.db_dir).joinpath(self.tar_name)
        try:
            with bakta_path.open("wb") as fh_out, requests.get(
                    self.db_url, stream=True) as resp:
                total_length = resp.headers.get("content-length")
                if total_length is None:  # no content length header
                    for data in resp.iter_content(chunk_size=1024 * 1024):
                        fh_out.write(data)
                else:
                    for data in resp.iter_content(chunk_size=1024 * 1024):
                        fh_out.write(data)
            print(f"Download bakta database {self.db_version}")
            self.tarball_path = bakta_path
        except IOError:
            print(
                f"ERROR: Could not download file from Zenodo!"

            with self.tarball_path.open("rb") as fh_in, tarfile.open(
                fileobj=fh_in, mode="r:gz"
            ) as tar_file:
                tar_file.extractall(path=db_path)
                print(f"Untar the database in {db_path}")

            if not self.test_mode:
                self.move_files(db_path=db_path)

        except OSError:
            sys.exit(f"ERROR: Could not extract {self.tar_name} " f"to {db_path}")

    def move_files(self, db_path):
        if db_path.joinpath("db-light").is_dir():
            input_dir = db_path.joinpath("db-light")
        elif db_path.joinpath("db").is_dir():
            input_dir = db_path.joinpath("db")
        output_dir = db_path
        for file in input_dir.iterdir():
            if file.is_file():  # to avoid moving amrfinder-plus folder
                input = input_dir.joinpath(file)
                output = output_dir.joinpath(file)
                input.rename(output)

    def calc_md5_sum(self, buffer_size=1048576):
        tarball_path = Path(self.db_dir).joinpath(self.tar_name)
        md5 = hashlib.md5()
        with tarball_path.open("rb") as fh:

6:97b1b5ad1cda

    def download(self):
        bakta_path = Path(self.db_dir).joinpath(self.tar_name)
        try:
            with bakta_path.open("wb") as fh_out, requests.get(
                    self.db_url, stream=True) as resp:
                # total_length = resp.headers.get("content-length")
                for data in resp.iter_content(chunk_size=1024 * 1024):
                    fh_out.write(data)
            print(f"Download bakta database {self.db_version}")
            self.tarball_path = bakta_path
        except IOError:
            print(
                f"ERROR: Could not download file from Zenodo!"

            with self.tarball_path.open("rb") as fh_in, tarfile.open(
                fileobj=fh_in, mode="r:gz"
            ) as tar_file:
                tar_file.extractall(path=db_path)
                print(f"Untar the database in {db_path}")
        except OSError:
            sys.exit(f"ERROR: Could not extract {self.tar_name} " f"to {db_path}")
        if not self.test_mode:
            self.move_files(db_path=db_path)
        self.db_dir = db_path.resolve()

    def delete_folder(self, path):
        for sub in path.iterdir():
            if sub.is_dir() and sub.name != "latest":
                self.delete_folder(sub)
            else:
                sub.unlink()
        path.rmdir()

    def move_files(self, db_path):
        if db_path.joinpath("db-light").is_dir():
            input_dir = db_path.joinpath("db-light")
        elif db_path.joinpath("db").is_dir():
            input_dir = db_path.joinpath("db")
        output_dir = db_path
        for file in input_dir.iterdir():
            if file.is_file():  # to avoid moving amrfinder-plus folder
                output = output_dir.joinpath(file.name)
                file.rename(output)
        self.delete_folder(input_dir)

    def calc_md5_sum(self, buffer_size=1048576):
        tarball_path = Path(self.db_dir).joinpath(self.tar_name)
        md5 = hashlib.md5()
        with tarball_path.open("rb") as fh: