code: def put(
      files=None,
      filelist=None,
      inputs=None,
      num_workers=None,
      verbose=None,
      overwrite=True,
      listing=None,
      s3role=None,
      s3sessionvars=None,
      s3clientparams=None,
      inject_failure=0,
  ):
      if inputs is not None and filelist is not None:
          raise RuntimeError(
              "Cannot specify inputs and filelist at the same time"
          )
      if inputs is not None and filelist is None:
          filelist = inputs
  
      is_transient_retry = False
  
      def _files():
          nonlocal is_transient_retry
          line_idx = 0
          for local, url in files:
              local_file = url_unquote(local)
              if not os.path.exists(local_file):
                  exit(ERROR_LOCAL_FILE_NOT_FOUND, local_file)
              yield line_idx, local_file, url_unquote(url), None, None
              line_idx += 1
          if filelist:
              # NOTE: We are assuming that the idx is properly set. This is only used
              # by the transient failure retry mechanism and users should not use it
              # directly. This will not work, for example, if only some lines have
              # an idx specified (in some cases)
              for line in open(filelist, mode="rb"):
                  r = json.loads(line)
                  input_line_idx = r.get("idx")
                  if input_line_idx is not None:
                      # We only have input indices if we have a transient retry.
                      is_transient_retry = True
                  else:
                      input_line_idx = line_idx
                  line_idx += 1
                  local = r["local"]
                  url = r["url"]
                  content_type = r.get("content_type", None)
                  metadata = r.get("metadata", None)
                  encryption = r.get("encryption", None)
                  if not os.path.exists(local):
                      exit(ERROR_LOCAL_FILE_NOT_FOUND, local)
                  yield input_line_idx, local, url, content_type, metadata, encryption
  
      def _make_url(
          idx, local, user_url, content_type, metadata, encryption
      ):
          src = urlparse(user_url)
          url = S3Url(
              url=user_url,
              bucket=src.netloc,
              path=src.path.lstrip("/"),
              local=local,
              prefix=None,
              content_type=content_type,
              metadata=metadata,
              idx=idx,
              encryption=encryption,
          )
          if src.scheme != "s3":
              exit(ERROR_INVALID_URL, url)
          if not src.path:
              exit(ERROR_NOT_FULL_PATH, url)
          return url
  
      s3config = S3Config(
          s3role,
          json.loads(s3sessionvars) if s3sessionvars else None,
          json.loads(s3clientparams) if s3clientparams else None,
      )
  
      urls = list(starmap(_make_url, _files()))
      ul_op = "upload"
      if not overwrite:
          ul_op = "info_upload"
      sz_results = process_urls(
          ul_op, urls, verbose, inject_failure, num_workers, s3config
      )
      retry_lines = []
      out_lines = []
      denied_url = None
      for url, sz in zip(urls, sz_results):
          # sz is None if the file wasn't uploaded (no overwrite), 0 if uploaded OK
          # or the error code if not (error code here will only be
          # ERROR_TRANSIENT or ERROR_URL_ACCESS_DENIED
          if sz is None:
              if listing:
                  # We keep a position for it in our out list in case of retries
                  out_lines.append(
                      "%d %s\n"
                      % (url.idx, TRANSIENT_RETRY_LINE_CONTENT)
                  )
              continue
          elif listing and sz == 0:
              out_lines.append(
                  format_result_line(url.idx, url.url) + "\n"
              )
          elif sz == -ERROR_TRANSIENT:
              retry_lines.append(
                  json.dumps(
                      {
                          "idx": url.idx,
                          "url": url.url,
                          "local": url.local,
                          "content_type": url.content_type,
                          "metadata": url.metadata,
                          "encryption": url.encryption,
                      }
                  )
                  + "\n"
              )
              # Output something to get a total count the first time around
              if not is_transient_retry:
                  out_lines.append(
                      "%d %s\n"
                      % (url.idx, TRANSIENT_RETRY_LINE_CONTENT)
                  )
          elif sz == -ERROR_URL_ACCESS_DENIED:
              # We do NOT break because we want to be able to accurately report all
              # the files uploaded after retries.
              denied_url = url
      if denied_url is not None:
          exit(ERROR_URL_ACCESS_DENIED, denied_url)
  
      if out_lines:
          sys.stdout.writelines(out_lines)
          sys.stdout.flush()
  
      if retry_lines:
          sys.stderr.write("%s\n" % TRANSIENT_RETRY_START_LINE)
          sys.stderr.writelines(retry_lines)
          sys.stderr.flush()
          sys.exit(ERROR_TRANSIENT)
  
imports: from __future__ import print_function
  import json
  import time
  import math
  import random
  import re
  import sys
  import os
  import traceback
  from collections import namedtuple
  from functools import partial, wraps
  from hashlib import sha1
  from tempfile import NamedTemporaryFile
  from multiprocessing import Process, Queue
  from itertools import starmap, chain, islice
  from boto3.s3.transfer import TransferConfig
  from metaflow._vendor import click
  from metaflow.util import TempDir, url_quote, url_unquote
  from metaflow.multicore_utils import parallel_map
  from metaflow.plugins.datatools.s3.s3util import aws_retry, read_in_chunks, get_timestamp, TRANSIENT_RETRY_LINE_CONTENT, TRANSIENT_RETRY_START_LINE
  import metaflow.tracing
label_desc: Use image node_id 18 for calling a global function with example usage: put(files, filelist, inputs, num_workers, verbose, overwrite, listing, s3role, s3sessionvars, s3clientparams, inject_failure) and returns: url