@retry.with_exponential_backoff(num_retries=4, retry_filter=lambda _: True)
def _gcs_file_copy(from_path, to_path, encoding=''):
  """Copy a local file to a GCS location with retries for transient errors."""
  if not encoding:
    command_args = ['gsutil', '-m', '-q', 'cp', from_path, to_path]
  else:
    encoding = 'Content-Type:' + encoding
    command_args = ['gsutil', '-m', '-q', '-h', encoding, 'cp', from_path,
                    to_path]
  logging.info('Executing command: %s', command_args)
  popen = processes.Popen(command_args, stdout=processes.PIPE,
                          stderr=processes.PIPE)
  stdoutdata, stderrdata = popen.communicate()
  if popen.returncode != 0:
    raise ValueError(
        'Failed to copy GCS file from %s to %s (stdout=%s, stderr=%s).' % (
            from_path, to_path, stdoutdata, stderrdata))


# -----------------------------------------------------------------------------
# TextFileSource, TextFileSink.
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/io/fileio.py/_gcs_file_copy/original