docs: def _direct_upload(file_obj, file_name, fields, session, samples_resource):
      """Uploads a single file-like object via our validating proxy. Maintains compatibility with direct upload
      to a user's S3 bucket as well in case we disable our validating proxy.
  
      Parameters
      ----------
      file_obj : `FASTXInterleave`, `FilePassthru`, or a file-like object
          A wrapper around a pair of fastx files (`FASTXInterleave`) or a single fastx file. In the
          case of paired files, they will be interleaved and uploaded uncompressed. In the case of a
          single file, it will simply be passed through (`FilePassthru`) to One Codex, compressed
          or otherwise. If a file-like object is given, its mime-type will be sent as 'text/plain'.
      file_name : `string`
          The file_name you wish to associate this fastx file with at One Codex.
      fields : `dict`
          Additional data fields to include as JSON in the POST. Must include 'sample_id' and
          'upload_url' at a minimum.
      samples_resource : `onecodex.models.Samples`
          Wrapped potion-client object exposing `init_upload` and `confirm_upload` routes to mainline.
  
      Raises
      ------
      RetryableUploadException
          In cases where the proxy is temporarily down or we experience connectivity issues
  
      UploadException
          In other cases where the proxy determines the upload is invalid and should *not* be retried.
      """
  
      # need an OrderedDict to preserve field order for S3, required for Python 2.7
      multipart_fields = OrderedDict()
  
      for k, v in fields["additional_fields"].items():
          multipart_fields[str(k)] = str(v)
  
      # this attribute is only in FASTXInterleave and FilePassthru
      mime_type = getattr(file_obj, "mime_type", "text/plain")
      multipart_fields["file"] = (file_name, file_obj, mime_type)
      encoder = MultipartEncoder(multipart_fields)
      upload_request = None
  
      try:
          upload_request = session.post(
              fields["upload_url"],
              data=encoder,
              headers={"Content-Type": encoder.content_type},
              auth={},
          )
      except requests.exceptions.ConnectionError:
          pass
  
      # If we expect a status *always* try to check it,
      # waiting up to 4 hours for buffering to complete (~30-50GB file gzipped)
      if "status_url" in fields["additional_fields"]:
          now = time.time()
          while time.time() < (now + 60 * 60 * 4):
              try:
                  resp = session.post(
                      fields["additional_fields"]["status_url"],
                      json={"sample_id": fields["sample_id"]},
                  )
                  resp.raise_for_status()
              except (ValueError, requests.exceptions.RequestException) as e:
                  logging.debug("Retrying due to error: {}".format(e))
                  raise RetryableUploadException(
                      "Unexpected failure of direct upload proxy. Retrying..."
                  )
  
              if resp.json() and resp.json().get("complete", True) is False:
                  logging.debug("Blocking on waiting for proxy to complete (in progress)...")
                  time.sleep(30)
              else:
                  break
  
          # Return is successfully processed
          if resp.json().get("code") in [200, 201]:
              file_obj.close()
              return
          elif resp.json().get("code") == 500:
              logging.debug("Retrying due to 500 from proxy...")
              raise RetryableUploadException("Unexpected issue with direct upload proxy. Retrying...")
          else:
              raise_api_error(resp, state="upload")
  
      # Direct to S3 case
      else:
          file_obj.close()
          if upload_request.status_code not in [200, 201]:
              raise RetryableUploadException("Unknown connectivity issue with proxy upload.")
  
          # Issue a callback -- this only happens in the direct-to-S3 case
          try:
              if not fields["additional_fields"].get("callback_url"):
                  samples_resource.confirm_upload(
                      {"sample_id": fields["sample_id"], "upload_type": "standard"}
                  )
          except requests.exceptions.HTTPError as e:
              raise_api_error(e.response, state="callback")
          except requests.exceptions.ConnectionError:
              raise_connectivity_error()