class_docstr: File-based datasource for reading files.
  
  Don't use this class directly. Instead, subclass it and implement `_read_stream()`.
class_docstr_tok: ["File-based","datasource","for","reading","files",".","Do","n't","use","this","class","directly",".","Instead",",","subclass","it","and","implement","`","_read_stream","(",")","`","."]
code: def __init__(
      self,
      paths: Union[str, List[str]],
      *,
      filesystem: Optional["pyarrow.fs.FileSystem"] = None,
      schema: Optional[Union[type, "pyarrow.lib.Schema"]] = None,
      open_stream_args: Optional[Dict[str, Any]] = None,
      meta_provider: BaseFileMetadataProvider = DefaultFileMetadataProvider(),
      partition_filter: PathPartitionFilter = None,
      partitioning: Partitioning = None,
      ignore_missing_paths: bool = False,
      shuffle: Union[Literal["files"], None] = None,
      include_paths: bool = False,
      file_extensions: Optional[List[str]] = None,
  ):
      _check_pyarrow_version()
  
      self._supports_distributed_reads = not _is_local_scheme(paths)
      if (
          not self._supports_distributed_reads
          and ray.util.client.ray.is_connected()
      ):
          raise ValueError(
              "Because you're using Ray Client, read tasks scheduled on the Ray "
              "cluster can't access your local files. To fix this issue, store "
              "files in cloud storage or a distributed filesystem like NFS."
          )
  
      self._schema = schema
      self._open_stream_args = open_stream_args
      self._meta_provider = meta_provider
      self._partition_filter = partition_filter
      self._partitioning = partitioning
      self._ignore_missing_paths = ignore_missing_paths
      self._include_paths = include_paths
      paths, self._filesystem = _resolve_paths_and_filesystem(
          paths, filesystem
      )
      paths, file_sizes = map(
          list,
          zip(
              *meta_provider.expand_paths(
                  paths,
                  self._filesystem,
                  partitioning,
                  ignore_missing_paths=ignore_missing_paths,
              )
          ),
      )
  
      if ignore_missing_paths and len(paths) == 0:
          raise ValueError(
              "None of the provided paths exist. "
              "The 'ignore_missing_paths' field is set to True."
          )
  
      if self._partition_filter is not None:
          # Use partition filter to skip files which are not needed.
          path_to_size = dict(zip(paths, file_sizes))
          paths = self._partition_filter(paths)
          file_sizes = [path_to_size[p] for p in paths]
          if len(paths) == 0:
              raise ValueError(
                  "No input files found to read. Please double check that "
                  "'partition_filter' field is set properly."
              )
  
      if file_extensions is not None:
          path_to_size = dict(zip(paths, file_sizes))
          paths = [
              p
              for p in paths
              if _has_file_extension(p, file_extensions)
          ]
          file_sizes = [path_to_size[p] for p in paths]
          if len(paths) == 0:
              raise ValueError(
                  "No input files found to read with the following file extensions: "
                  f"{file_extensions}. Please double check that "
                  "'file_extensions' field is set properly."
              )
  
      self._file_metadata_shuffler = None
      if shuffle == "files":
          self._file_metadata_shuffler = np.random.default_rng()
  
      # Read tasks serialize `FileBasedDatasource` instances, and the list of paths
      # can be large. To avoid slow serialization speeds, we store a reference to
      # the paths rather than the paths themselves.
      self._paths_ref = ray.put(paths)
      self._file_sizes_ref = ray.put(file_sizes)
  
imports: import io
  import pathlib
  import posixpath
  import warnings
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator, List, Literal, Optional, Union
  import numpy
  import ray
  from ray._private.utils import _add_creatable_buckets_param_if_s3_uri
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data._internal.util import _check_pyarrow_version, _is_local_scheme, call_with_retry, make_async_gen
  from ray.data.block import Block, BlockAccessor
  from ray.data.context import DataContext
  from ray.data.datasource.block_path_provider import BlockWritePathProvider
  from ray.data.datasource.datasource import Datasource, ReadTask, WriteResult
  from ray.data.datasource.file_meta_provider import BaseFileMetadataProvider, DefaultFileMetadataProvider
  from ray.data.datasource.filename_provider import FilenameProvider, _DefaultFilenameProvider
  from ray.data.datasource.partitioning import Partitioning, PathPartitionFilter, PathPartitionParser
  from ray.data.datasource.path_util import _has_file_extension, _resolve_paths_and_filesystem
  from ray.util.annotations import Deprecated, DeveloperAPI, PublicAPI
label_desc: Use image node_id 1 to create a new FileBasedDatasource object from inherited base classes: Datasource with example: obj = FileBasedDatasource(paths)