def dicom_store_proxy()

in pathology/dicom_proxy/dicom_store_util.py [0:0]

• 50 lines of code
• 8 McCabe index (conditional complexity)

def dicom_store_proxy(params: Optional[List[str]] = None) -> flask.Response:
  """Processes generic DICOM Store proxy request and return results.

  Args:
    params: Optional parameters to add to proxy request.

  Returns:
    flask.Response
  """
  method = flask_util.get_method()
  # Convert proxy web request url into url against DICOM store.
  full_path = _get_flask_full_path_removing_unsupported_proxy_params()
  cloud_logging_client.info(f'Entry Proxy request: {full_path}')
  if params is not None and params:
    full_path = _add_parameters_to_url(full_path, params)
  url = _remove_dicom_proxy_url_path_prefix(full_path)
  url = dicom_url_util.get_dicom_store_url(url)
  # The accept(key):value pair in the HTTP request defines the image/data
  # format of what the client is requesting. Copy the header key:value from
  # the request and use it in the call to the DICOM store.
  header = flask_util.norm_dict_keys(
      flask_util.get_headers(),
      [
          flask_util.ACCEPT_HEADER_KEY,
          flask_util.CONTENT_TYPE_HEADER_KEY,
          flask_util.CONTENT_LENGTH_HEADER_KEY,
          flask_util.ACCEPT_ENCODING_KEY,
      ],
  )
  headers = AuthSession(flask_util.get_headers()).add_to_header(header)
  cloud_logging_client.info(
      f'Proxied URL: {url}',
      {
          'proxy_headers': logging_util.mask_privileged_header_values(headers),
          'proxy_method': method,
      },
  )
  # Proxy using same method called with
  # Stream data from proxy to caller to avoid fully loading response in server
  if method == flask_util.GET:
    response = requests.get(url, headers=headers, stream=True)
  elif method == flask_util.POST:
    response = requests.post(
        url, headers=headers, stream=True, data=flask_util.get_data()
    )
  elif method == flask_util.DELETE:
    response = requests.delete(url, headers=headers, stream=True)
  else:
    msg = 'Unsupported HTTP request method.'
    cloud_logging_client.error(msg, {'method': method, 'url': url})
    return flask.Response(
        response=str(msg),
        status=http.HTTPStatus.BAD_REQUEST.value,
        content_type='text/plain',
    )
  fl_response = flask.Response(
      _stream_proxy_content(response),
      status=response.status_code,
      direct_passthrough=False,
  )
  fl_response.headers.clear()
  for key, value in response.headers.items():
    fl_response.headers[key] = value
  return fl_response