Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _extract_file(self, fname, extract_dir):
"""
This method receives an argument for the archive to extract and the
destination path.
"""
with TarFile.open(fname, "r") as tar_file:
if self.members is None:
get_logger().info(
"Untarring contents of '%s' to '%s'", fname, extract_dir
)
# Unpack all files from the archive into our new folder
tar_file.extractall(path=extract_dir)
else:
for member in self.members:
get_logger().info(
"Extracting '%s' from '%s' to '%s'", member, fname, extract_dir
)
# Extract the data file from within the archive
# Python 2.7: extractfile doesn't return a context manager
data_file = tar_file.extractfile(member)
try:
# Save it to our desired file name
with open(os.path.join(extract_dir, member), "wb") as output:
output.write(data_file.read())
finally:
data_file.close()
full_path = path.resolve() / fname
action, verb = download_action(full_path, known_hash)
if action in ("download", "update"):
get_logger().info(
"%s data from '%s' to file '%s'.", verb, url, str(full_path),
)
if downloader is None:
downloader = choose_downloader(url)
stream_download(url, full_path, known_hash, downloader, pooch=None)
if known_hash is None:
get_logger().info(
"SHA256 hash of downloaded file: %s\n"
"Use this value as the 'known_hash' argument of 'pooch.retrieve'"
" to ensure that the file hasn't changed if it is downloaded again"
" in the future.",
file_hash(str(full_path)),
)
if processor is not None:
return processor(str(full_path), action, None)
return str(full_path)
- ``"download"``: File didn't exist locally and was downloaded
- ``"update"``: Local file was outdated and was re-download
- ``"fetch"``: File exists and is updated so it wasn't downloaded
pooch : :class:`pooch.Pooch`
The instance of :class:`pooch.Pooch` that is calling this.
Returns
-------
fname : str
The full path to the decompressed file.
"""
decompressed = fname + ".decomp"
if action in ("update", "download") or not os.path.exists(decompressed):
get_logger().info(
"Decompressing '%s' to '%s' using method '%s'.",
fname,
decompressed,
self.method,
)
module = self._compression_module(fname)
with open(decompressed, "w+b") as output:
with module.open(fname) as compressed:
shutil.copyfileobj(compressed, output)
return decompressed
# Will print a progress bar to standard error like:
# 100%|█████████████████████████████████████████| 336/336 [...]
"""
self._assert_file_in_registry(fname)
# Create the local data directory if it doesn't already exist
make_local_storage(str(self.abspath))
url = self.get_url(fname)
full_path = self.abspath / fname
known_hash = self.registry[fname]
action, verb = download_action(full_path, known_hash)
if action in ("download", "update"):
get_logger().info(
"%s file '%s' from '%s' to '%s'.", verb, fname, url, str(self.abspath),
)
if downloader is None:
downloader = choose_downloader(url)
stream_download(url, full_path, known_hash, downloader, pooch=self)
if processor is not None:
return processor(str(full_path), action, self)
return str(full_path)