def handle_saml_assertion()

in redshift_connector/plugin/okta_credentials_provider.py [0:0]
33 lines of code
6 McCabe index (conditional complexity)

    def handle_saml_assertion(self: "OktaCredentialsProvider", okta_session_token: str) -> str:
        import bs4  # type: ignore
        import requests

        url: str = "https://{host}/home/{app_name}/{app_id}?onetimetoken={session_token}".format(
            host=self.idp_host, app_name=self.app_name, app_id=self.app_id, session_token=okta_session_token
        )
        _logger.debug("OktaAWSAppUrl: {}".format(url))

        try:
            response: "requests.Response" = requests.get(url, verify=self.do_verify_ssl_cert())
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            _logger.error("Request for SAML assertion from Okta was unsuccessful. {}".format(str(e)))
            raise InterfaceError(e)
        except requests.exceptions.Timeout as e:
            _logger.error("A timeout occurred when requesting SAML assertion from Okta")
            raise InterfaceError(e)
        except requests.exceptions.TooManyRedirects as e:
            _logger.error(
                "A error occurred when requesting SAML assertion from Okta. Verify RedshiftProperties are correct"
            )
            raise InterfaceError(e)
        except requests.exceptions.RequestException as e:
            _logger.error("A unknown error occurred when requesting SAML assertion from Okta")
            raise InterfaceError(e)

        text: str = response.text
        _logger.debug(response.content)

        try:
            soup = bs4.BeautifulSoup(text, "html.parser")
            saml_response: str = soup.find("input", {"name": "SAMLResponse"})["value"]
            return saml_response
        except Exception as e:
            _logger.error("An error occurred while parsing SAML response: {}".format(str(e)))
            raise InterfaceError(e)