code: def _request(
      cls,
      monitor,
      path,
      method,
      data=None,
      retry_409_path=None,
      return_raw_resp=False,
  ):
      if cls.INFO is None:
          raise MetaflowException(
              "Missing Metaflow Service URL. "
              "Specify with METAFLOW_SERVICE_URL environment variable"
          )
      supported_methods = ("GET", "PATCH", "POST")
      if method not in supported_methods:
          raise MetaflowException(
              "Only these methods are supported: %s, but got %s"
              % (supported_methods, method)
          )
      url = os.path.join(cls.INFO, path.lstrip("/"))
      for i in range(SERVICE_RETRY_COUNT):
          try:
              if method == "GET":
                  if monitor:
                      with monitor.measure(
                          "metaflow.service_metadata.get"
                      ):
                          resp = requests.get(
                              url, headers=SERVICE_HEADERS.copy()
                          )
                  else:
                      resp = requests.get(
                          url, headers=SERVICE_HEADERS.copy()
                      )
              elif method == "POST":
                  if monitor:
                      with monitor.measure(
                          "metaflow.service_metadata.post"
                      ):
                          resp = requests.post(
                              url,
                              headers=SERVICE_HEADERS.copy(),
                              json=data,
                          )
                  else:
                      resp = requests.post(
                          url, headers=SERVICE_HEADERS.copy(), json=data
                      )
              elif method == "PATCH":
                  if monitor:
                      with monitor.measure(
                          "metaflow.service_metadata.patch"
                      ):
                          resp = requests.patch(
                              url,
                              headers=SERVICE_HEADERS.copy(),
                              json=data,
                          )
                  else:
                      resp = requests.patch(
                          url, headers=SERVICE_HEADERS.copy(), json=data
                      )
              else:
                  raise MetaflowInternalError(
                      "Unexpected HTTP method %s" % (method,)
                  )
          except MetaflowInternalError:
              raise
          except:  # noqa E722
              if monitor:
                  with monitor.count(
                      "metaflow.service_metadata.failed_request"
                  ):
                      if i == SERVICE_RETRY_COUNT - 1:
                          raise
              else:
                  if i == SERVICE_RETRY_COUNT - 1:
                      raise
              resp = None
          else:
              if return_raw_resp:
                  return resp, True
              if resp.status_code < 300:
                  return resp.json(), True
              elif resp.status_code == 409 and data is not None:
                  # a special case: the post fails due to a conflict
                  # this could occur when we missed a success response
                  # from the first POST request but the request
                  # actually went though, so a subsequent POST
                  # returns 409 (conflict) or we end up with a
                  # conflict while running on AWS Step Functions
                  # instead of retrying the post we retry with a get since
                  # the record is guaranteed to exist
                  if retry_409_path:
                      v, _ = cls._request(
                          monitor, retry_409_path, "GET"
                      )
                      return v, False
                  else:
                      return None, False
              elif resp.status_code != 503:
                  raise ServiceException(
                      "Metadata request (%s) failed (code %s): %s"
                      % (path, resp.status_code, resp.text),
                      resp.status_code,
                      resp.text,
                  )
          time.sleep(2**i)
  
      if resp:
          raise ServiceException(
              "Metadata request (%s) failed (code %s): %s"
              % (path, resp.status_code, resp.text),
              resp.status_code,
              resp.text,
          )
      else:
          raise ServiceException("Metadata request (%s) failed" % path)
  
imports: import os
  import random
  import requests
  import time
  from metaflow.exception import MetaflowException, MetaflowTaggingError, MetaflowInternalError
  from metaflow.metaflow_config import SERVICE_RETRY_COUNT, SERVICE_HEADERS, SERVICE_URL
  from metaflow.metadata import MetadataProvider
  from metaflow.metadata.heartbeat import HB_URL_KEY
  from metaflow.sidecar import Message, MessageTypes, Sidecar
  from metaflow.util import version_parse
label_desc: Use image node_id 23 for calling the ServiceMetadataProvider obj's underlying member method code with example usage: obj._request(cls, monitor, path, method, data, retry_409_path, return_raw_resp) and returns: resp,  True, resp,  True, v,  False, None,  False