def __cache_file_locally()

in python/dpu_utils/utils/richpath.py [0:0]

• 31 lines of code
• 10 McCabe index (conditional complexity)

    def __cache_file_locally(self, num_retries: int=1) -> LocalPath:
        cached_file_path = self.__cached_file_path
        cached_file_path_etag = cached_file_path+'.etag'  # Create an .etag file containing the object etag
        old_etag = None
        if os.path.exists(cached_file_path_etag):
            with open(cached_file_path_etag) as f:
                old_etag = f.read()

        try:
            os.makedirs(os.path.dirname(cached_file_path), exist_ok=True)
            # The next invocation to the blob service may fail and delete the current file. Store it elsewhere
            new_filepath = cached_file_path+'.new'

            if old_etag is not None:
                downloader = self.__blob_client.download_blob(etag=old_etag, match_condition=MatchConditions.IfModified)
            else:
                downloader = self.__blob_client.download_blob()
            with open(new_filepath, 'wb') as f:
                downloader.readinto(f)

            os.rename(new_filepath, cached_file_path)
            with open(cached_file_path_etag, 'w') as f:
                f.write(downloader.properties['etag'])
        except HttpResponseError as responseError:
            if responseError.status_code != 304:  # HTTP 304: Not Modified
                raise
        except Exception as e:
            if os.path.exists(cached_file_path):
                os.remove(cached_file_path)   # On failure, remove the cached file, if it exits.
                os.remove(cached_file_path_etag)
            if num_retries == 0:
                raise
            else:
                self.__cache_file_locally(num_retries-1)
        return LocalPath(cached_file_path)