def get_file()

in src/huggingface_hub/hf_file_system.py [0:0]

• 40 lines of code
• 12 McCabe index (conditional complexity)

    def get_file(self, rpath, lpath, callback=_DEFAULT_CALLBACK, outfile=None, **kwargs) -> None:
        """
        Copy single remote file to local.

        <Tip warning={true}>

            Note: When possible, use `HfApi.hf_hub_download()` for better performance.

        </Tip>

        Args:
            rpath (`str`):
                Remote path to download from.
            lpath (`str`):
                Local path to download to.
            callback (`Callback`, *optional*):
                Optional callback to track download progress. Defaults to no callback.
            outfile (`IO`, *optional*):
                Optional file-like object to write to. If provided, `lpath` is ignored.

        """
        revision = kwargs.get("revision")
        unhandled_kwargs = set(kwargs.keys()) - {"revision"}
        if not isinstance(callback, (NoOpCallback, TqdmCallback)) or len(unhandled_kwargs) > 0:
            # for now, let's not handle custom callbacks
            # and let's not handle custom kwargs
            return super().get_file(rpath, lpath, callback=callback, outfile=outfile, **kwargs)

        # Taken from https://github.com/fsspec/filesystem_spec/blob/47b445ae4c284a82dd15e0287b1ffc410e8fc470/fsspec/spec.py#L883
        if isfilelike(lpath):
            outfile = lpath
        elif self.isdir(rpath):
            os.makedirs(lpath, exist_ok=True)
            return None

        if isinstance(lpath, (str, Path)):  # otherwise, let's assume it's a file-like object
            os.makedirs(os.path.dirname(lpath), exist_ok=True)

        # Open file if not already open
        close_file = False
        if outfile is None:
            outfile = open(lpath, "wb")
            close_file = True
        initial_pos = outfile.tell()

        # Custom implementation of `get_file` to use `http_get`.
        resolve_remote_path = self.resolve_path(rpath, revision=revision)
        expected_size = self.info(rpath, revision=revision)["size"]
        callback.set_size(expected_size)
        try:
            http_get(
                url=hf_hub_url(
                    repo_id=resolve_remote_path.repo_id,
                    revision=resolve_remote_path.revision,
                    filename=resolve_remote_path.path_in_repo,
                    repo_type=resolve_remote_path.repo_type,
                    endpoint=self.endpoint,
                ),
                temp_file=outfile,
                displayed_filename=rpath,
                expected_size=expected_size,
                resume_size=0,
                headers=self._api._build_hf_headers(),
                _tqdm_bar=callback.tqdm if isinstance(callback, TqdmCallback) else None,
            )
            outfile.seek(initial_pos)
        finally:
            # Close file only if we opened it ourselves
            if close_file:
                outfile.close()