def api_request()

in components/frontend_streamlit/src/api.py [0:0]
59 lines of code
17 McCabe index (conditional complexity)

def api_request(method: str, api_url: str,
                request_body: dict = None, auth_token: str = None):
  """ Make API request with error handling. """

  st.session_state.error_msg = None
  try:
    resp = None
    logging.info("api_url=%s, auth_token=%s", api_url, auth_token)

    # global processing of llm_type param
    if request_body and isinstance(request_body, dict):
      llm_type = request_body.get("llm_type", None)
      if llm_type == "default":
        del request_body["llm_type"]

    resp, resp_dict, status_code = dispatch_api(method,
                                                api_url,
                                                request_body,
                                                auth_token)

    if status_code == 401 or resp_dict.get("success", False) is False:
      # refresh token with existing creds and retry on failure to authenticate
      username = st.session_state.get("username", None)
      password = st.session_state.get("password", None)
      logging.info("API responds 401 error. attempting to re-auth with "
                  "username: %s", username)
      if username and password:
        auth_token = login_user(username, password)
        resp, resp_dict, status_code = dispatch_api(method,
                                                    api_url,
                                                    request_body,
                                                    auth_token)

      if status_code == 401:
        logging.error(
            "Unauthorized when calling API: %s", api_url)
        st.session_state.error_msg = \
            "Unauthorized or session expired. " \
            "Please [login]({APP_BASE_PATH}/Login) again."

      if resp_dict.get("success", False) is False:
        msg = resp_dict.get("message", "no message returned")
        st.session_state.error_msg = f"API call failed: {msg} "

    if status_code != 200:
      logging.error("Error with status %s: %s", status_code, str(resp))
      st.session_state.error_msg = \
          f"Error with status {status_code}: {str(resp)}"

    return resp

  except requests.exceptions.ConnectionError as e:
    logging.error(e)
    st.session_state.error_msg = \
        "Unable to connect to backend APIs. Please try again later."

  except RuntimeError as e:
    logging.error(e)
    st.session_state.error_msg = str(e)

  except json.decoder.JSONDecodeError as e:
    logging.error("Unable to parse response: %s", resp)
    logging.error(e)
    st.session_state.error_msg = \
        f"Unable to decode response from backend APIs: {resp}"

  finally:
    if st.session_state.error_msg:
      st.error(st.session_state.error_msg)

      if st.session_state.get("debug", False):
        with st.expander("Expand to see detail:"):
          st.write(f"API URL: {api_url}")
          st.write(e)
      st.stop()