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","(",")","`","."]
docstr: Write blocks for a file-based datasource.
docstr_tok: ["Write","blocks","for","a","file-based","datasource","."]
code: def write(
      self,
      blocks: Iterable[Block],
      ctx: TaskContext,
      path: str,
      dataset_uuid: str,
      filesystem: Optional["pyarrow.fs.FileSystem"] = None,
      try_create_dir: bool = True,
      open_stream_args: Optional[Dict[str, Any]] = None,
      block_path_provider: Optional[
          BlockWritePathProvider
      ] = None,  # Deprecated
      filename_provider: Optional[FilenameProvider] = None,
      write_args_fn: Callable[[], Dict[str, Any]] = lambda: {},
      file_format: Optional[str] = None,
      _block_udf: Optional[Callable[[Block], Block]] = None,
      **write_args,
  ) -> WriteResult:
      """Write blocks for a file-based datasource."""
      # `FileBasedDatasource` subclasses expose a `_FILE_EXTENSIONS` attribute. It
      # represents a list of supported file extensions. If the user doesn't specify
      # a file format, we default to the first extension in the list.
      if file_format is None:
          file_format = self._FILE_EXTENSIONS
          if isinstance(file_format, list):
              file_format = file_format[0]
  
      path, filesystem = _resolve_paths_and_filesystem(path, filesystem)
      path = path[0]
      _write_block_to_file = self._write_block
      _write_row_to_file = self._write_row
  
      if open_stream_args is None:
          open_stream_args = {}
  
      if block_path_provider is not None:
          warnings.warn(
              "`block_path_provider` has been deprecated in favor of "
              "`filename_provider`. For more information, see "
              "https://docs.ray.io/en/master/data/api/doc/ray.data.datasource.FilenameProvider.html",  # noqa: E501
              DeprecationWarning,
          )
  
      if filename_provider is None and block_path_provider is None:
          filename_provider = _DefaultFilenameProvider(
              dataset_uuid=dataset_uuid, file_format=file_format
          )
  
      num_rows_written = 0
      block_idx = 0
      for block in blocks:
          if _block_udf is not None:
              block = _block_udf(block)
  
          block = BlockAccessor.for_block(block)
          if block.num_rows() == 0:
              continue
  
          fs = _unwrap_s3_serialization_workaround(filesystem)
  
          if self._WRITE_FILE_PER_ROW:
              for row_idx, row in enumerate(
                  block.iter_rows(public_row_format=False)
              ):
                  if filename_provider is not None:
                      filename = filename_provider.get_filename_for_row(
                          row, ctx.task_idx, block_idx, row_idx
                      )
                  else:  # Legacy code path
                      filename = (
                          f"{dataset_uuid}_{ctx.task_idx:06}_{block_idx:06}_"
                          f"{row_idx:06}.{file_format}"
                      )
                  write_path = posixpath.join(path, filename)
  
                  logger.get_logger().debug(
                      f"Writing {write_path} file."
                  )
                  with _open_file_with_retry(
                      write_path,
                      lambda: fs.open_output_stream(
                          write_path, **open_stream_args
                      ),
                  ) as f:
                      _write_row_to_file(
                          f,
                          row,
                          writer_args_fn=write_args_fn,
                          file_format=file_format,
                          **write_args,
                      )
          else:
              if filename_provider is not None:
                  filename = filename_provider.get_filename_for_block(
                      block, ctx.task_idx, block_idx
                  )
                  write_path = posixpath.join(path, filename)
              else:  # Legacy code path
                  write_path = block_path_provider(
                      path,
                      filesystem=filesystem,
                      dataset_uuid=dataset_uuid,
                      task_index=ctx.task_idx,
                      block_index=block_idx,
                      file_format=file_format,
                  )
  
              logger.get_logger().debug(f"Writing {write_path} file.")
              with _open_file_with_retry(
                  write_path,
                  lambda: fs.open_output_stream(
                      write_path, **open_stream_args
                  ),
              ) as f:
                  _write_block_to_file(
                      f,
                      block,
                      writer_args_fn=write_args_fn,
                      **write_args,
                  )
  
          num_rows_written += block.num_rows()
          block_idx += 1
  
      if num_rows_written == 0:
          logger.get_logger().warning(
              f"Skipping writing empty dataset with UUID {dataset_uuid} at {path}",
          )
          return "skip"
  
      # TODO: decide if we want to return richer object when the task
      # succeeds.
      return "ok"
  
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 10 for calling the FileBasedDatasource obj's underlying member method code with example usage: obj.write(blocks, ctx, path, dataset_uuid, filesystem, try_create_dir, open_stream_args, block_path_provider, filename_provider, write_args_fn, file_format, _block_udf) and returns: str, str