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 get_read_tasks(self, parallelism: int) -> List[ReadTask]:
      import numpy as np
  
      ctx = DataContext.get_current()
      open_stream_args = self._open_stream_args
      partitioning = self._partitioning
  
      paths = self._paths()
      file_sizes = self._file_sizes()
  
      if self._file_metadata_shuffler is not None:
          files_metadata = list(zip(paths, file_sizes))
          shuffled_files_metadata = [
              files_metadata[i]
              for i in self._file_metadata_shuffler.permutation(
                  len(files_metadata)
              )
          ]
          paths, file_sizes = list(
              map(list, zip(*shuffled_files_metadata))
          )
  
      read_stream = self._read_stream
      filesystem = _wrap_s3_serialization_workaround(self._filesystem)
  
      if open_stream_args is None:
          open_stream_args = {}
  
      open_input_source = self._open_input_source
  
      def read_files(
          read_paths: Iterable[str],
      ) -> Iterable[Block]:
          nonlocal filesystem, open_stream_args, partitioning
  
          DataContext._set_current(ctx)
          fs = _unwrap_s3_serialization_workaround(filesystem)
          for read_path in read_paths:
              partitions: Dict[str, str] = {}
              if partitioning is not None:
                  parse = PathPartitionParser(partitioning)
                  partitions = parse(read_path)
  
              with _open_file_with_retry(
                  read_path,
                  lambda: open_input_source(
                      fs, read_path, **open_stream_args
                  ),
              ) as f:
                  for block in read_stream(f, read_path):
                      if partitions:
                          block = _add_partitions(block, partitions)
                      if self._include_paths:
                          block_accessor = BlockAccessor.for_block(
                              block
                          )
                          block = block_accessor.append_column(
                              "path",
                              [read_path] * block_accessor.num_rows(),
                          )
                      yield block
  
      def create_read_task_fn(read_paths, num_threads):
          def read_task_fn():
              nonlocal num_threads, read_paths
  
              if num_threads > 0:
                  if len(read_paths) < num_threads:
                      num_threads = len(read_paths)
  
                  logger.get_logger().debug(
                      f"Reading {len(read_paths)} files with {num_threads} threads."
                  )
  
                  yield from make_async_gen(
                      iter(read_paths),
                      read_files,
                      num_workers=num_threads,
                  )
              else:
                  logger.get_logger().debug(
                      f"Reading {len(read_paths)} files."
                  )
                  yield from read_files(read_paths)
  
          return read_task_fn
  
      # fix https://github.com/ray-project/ray/issues/24296
      parallelism = min(parallelism, len(paths))
  
      read_tasks = []
      for read_paths, file_sizes in zip(
          np.array_split(paths, parallelism),
          np.array_split(file_sizes, parallelism),
      ):
          if len(read_paths) <= 0:
              continue
  
          meta = self._meta_provider(
              read_paths,
              self._schema,
              rows_per_file=self._rows_per_file(),
              file_sizes=file_sizes,
          )
  
          read_task_fn = create_read_task_fn(
              read_paths, self._NUM_THREADS_PER_TASK
          )
  
          read_task = ReadTask(read_task_fn, meta)
  
          read_tasks.append(read_task)
  
      return read_tasks
  
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 5 for calling the FileBasedDatasource obj's underlying member method code with example usage: obj.get_read_tasks(parallelism) and returns: read_tasks, read_task_fn