Function xopen

src/datasets/utils/file_utils.py:945–1010

Extend `open` function to support remote files using `fsspec`. It also has a retry mechanism in case connection fails. The `args` and `kwargs` are passed to `fsspec.open`, except `token` which is used for queries to private repos on huggingface.co Args: file (`str`): Path name

(file: str, mode="r", *args, download_config: Optional[DownloadConfig] = None, **kwargs)

def xopen(file: str, mode="r", *args, download_config: Optional[DownloadConfig] = None, **kwargs):
    """Extend `open` function to support remote files using `fsspec`.

    It also has a retry mechanism in case connection fails.
    The `args` and `kwargs` are passed to `fsspec.open`, except `token` which is used for queries to private repos on huggingface.co

    Args:
        file (`str`): Path name of the file to be opened.
        mode (`str`, *optional*, default "r"): Mode in which the file is opened.
        *args: Arguments to be passed to `fsspec.open`.
        download_config : mainly use token or storage_options to support different platforms and auth types.
        **kwargs: Keyword arguments to be passed to `fsspec.open`.

    Returns:
        file object
    """
    # This works as well for `xopen(str(Path(...)))`
    file_str = _as_str(file)
    main_hop, *rest_hops = file_str.split("::")
    if is_local_path(main_hop):
        # ignore fsspec-specific kwargs
        kwargs.pop("block_size", None)
        return open(main_hop, mode, *args, **kwargs)
    # add headers and cookies for authentication on the HF Hub and for Google Drive
    file, storage_options = _prepare_path_and_storage_options(file_str, download_config=download_config)
    kwargs = {**kwargs, **(storage_options or {})}

    max_retries = config.STREAMING_OPEN_MAX_RETRIES

    disconnect_err = None
    for retry in range(1, max_retries + 1):
        try:
            fs, fs_token, paths = fsspec.get_fs_token_paths(
                file,
                mode,
                storage_options=kwargs,
            )
            file_obj = fs.open(paths[0], mode)
            if hasattr(fs, "of") and hasattr(fs.of, "__exit__"):
                file_obj._fs = fs  # keep a reference or the fs might close the file on gc
            break
        except CONNECTION_ERRORS_TO_RETRY as err:
            disconnect_err = err
            logger.warning(
                f"Failed to connect to remote data host. Retrying in {config.STREAMING_OPEN_RETRY_INTERVAL}sec [{retry}/{max_retries}]"
            )
            time.sleep(config.STREAMING_OPEN_RETRY_INTERVAL)
        except ValueError as e:
            if str(e) == "Cannot seek streaming HTTP file":
                raise NonStreamableDatasetError(
                    "Streaming is not possible for this dataset because data host server doesn't support HTTP range "
                    "requests. You can still load this dataset in non-streaming mode by passing `streaming=False` (default)"
                ) from e
            else:
                raise
        except FileNotFoundError:
            if file.startswith(config.HF_ENDPOINT):
                raise FileNotFoundError(