perf(imports): use lazy-imports for idp plugins

Author: Brooke-white

redshift_connector/iam_helper.py

@@ -2,9 +2,6 @@
 import typing
 from enum import Enum
 
-import boto3  # type: ignore
-import botocore  # type: ignore
-
 from redshift_connector.config import ClientProtocolVersion
 from redshift_connector.credentials_holder import CredentialsHolder
 from redshift_connector.error import InterfaceError
@@ -233,6 +230,9 @@ def set_iam_credentials(info: RedshiftProperty) -> None:
 # Calls the AWS SDK methods to return temporary credentials.
 # The expiration date is returned as the local time set by the client machines OS.
 def set_cluster_credentials(cred_provider: SamlCredentialsProvider, info: RedshiftProperty) -> None:
+    import boto3  # type: ignore
+    import botocore  # type: ignore
+
     try:
         credentials: CredentialsHolder = cred_provider.get_credentials()
         client = boto3.client(

redshift_connector/plugin/adfs_credentials_provider.py

@@ -2,9 +2,6 @@
 import re
 import typing
 
-import bs4  # type: ignore
-import requests
-
 from redshift_connector.error import InterfaceError
 from redshift_connector.plugin.saml_credentials_provider import SamlCredentialsProvider
 
@@ -26,6 +23,9 @@ def windows_integrated_authentication(self: "AdfsCredentialsProvider"):
         pass
 
     def form_based_authentication(self: "AdfsCredentialsProvider") -> str:
+        import bs4  # type: ignore
+        import requests
+
         url: str = "https://{host}:{port}/adfs/ls/IdpInitiatedSignOn.aspx?loginToRp=urn:amazon:webservices".format(
             host=self.idp_host, port=str(self.idpPort)
         )

redshift_connector/plugin/azure_credentials_provider.py

@@ -2,8 +2,6 @@
 import logging
 import typing
 
-import requests
-
 from redshift_connector.error import InterfaceError
 from redshift_connector.plugin.credential_provider_constants import azure_headers
 from redshift_connector.plugin.saml_credentials_provider import SamlCredentialsProvider
@@ -53,6 +51,8 @@ def get_saml_assertion(self: "AzureCredentialsProvider") -> str:
     #  Method to initiate a POST request to grab the SAML Assertion from Microsoft Azure
     #  and convert it to a SAML Response.
     def azure_oauth_based_authentication(self: "AzureCredentialsProvider") -> str:
+        import requests
+
         # endpoint to connect with Microsoft Azure to get SAML Assertion token
         url: str = "https://login.microsoftonline.com/{tenant}/oauth2/token".format(tenant=self.idp_tenant)
         # headers to pass with POST request

redshift_connector/plugin/browser_azure_credentials_provider.py

@@ -4,9 +4,6 @@
 import random
 import socket
 import typing
-import webbrowser
-
-import requests
 
 from redshift_connector.error import InterfaceError
 from redshift_connector.plugin.credential_provider_constants import azure_headers
@@ -99,6 +96,8 @@ def fetch_authorization_token(self: "BrowserAzureCredentialsProvider", listen_so
     # Initiates the request to the IDP and gets the response body
     # Get Base 64 encoded saml assertion from the response body
     def fetch_saml_response(self: "BrowserAzureCredentialsProvider", token):
+        import requests
+
         url: str = "https://login.microsoftonline.com/{tenant}/oauth2/token".format(tenant=self.idp_tenant)
         # headers to pass with POST request
         headers: typing.Dict[str, str] = azure_headers
@@ -203,6 +202,8 @@ def run_server(
 
     # Opens the default browser with the authorization request to the IDP
     def open_browser(self: "BrowserAzureCredentialsProvider", state: str) -> None:
+        import webbrowser
+
         url: str = (
             "https://login.microsoftonline.com/{tenant}"
             "/oauth2/authorize"

redshift_connector/plugin/browser_saml_credentials_provider.py

@@ -4,7 +4,6 @@
 import socket
 import typing
 import urllib.parse
-import webbrowser
 
 from redshift_connector.error import InterfaceError
 from redshift_connector.plugin.saml_credentials_provider import SamlCredentialsProvider
@@ -92,6 +91,8 @@ def run_server(self: "BrowserSamlCredentialsProvider", listen_port: int, idp_res
 
     # Opens the default browser with the authorization request to the web service.
     def open_browser(self: "BrowserSamlCredentialsProvider") -> None:
+        import webbrowser
+
         url: typing.Optional[str] = self.login_url
         if url is None:
             raise InterfaceError("the login_url could not be empty")

redshift_connector/plugin/okta_credentials_provider.py

@@ -2,9 +2,6 @@
 import logging
 import typing
 
-import bs4  # type: ignore
-import requests
-
 from redshift_connector.error import InterfaceError
 from redshift_connector.plugin.credential_provider_constants import okta_headers
 from redshift_connector.plugin.saml_credentials_provider import SamlCredentialsProvider
@@ -35,6 +32,8 @@ def get_saml_assertion(self: "OktaCredentialsProvider") -> str:
 
     # Authenticates users credentials via Okta, return Okta session token.
     def okta_authentication(self: "OktaCredentialsProvider") -> str:
+        import requests
+
         # HTTP Post request to Okta API for session token
         url: str = "https://{host}/api/v1/authn".format(host=self.idp_host)
         headers: typing.Dict[str, str] = okta_headers
@@ -70,6 +69,9 @@ def okta_authentication(self: "OktaCredentialsProvider") -> str:
 
     # Retrieves SAML assertion from Okta containing AWS roles.
     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
         )

redshift_connector/plugin/ping_credentials_provider.py

@@ -2,9 +2,6 @@
 import re
 import typing
 
-import bs4  # type: ignore
-import requests
-
 from redshift_connector.error import InterfaceError
 from redshift_connector.plugin.saml_credentials_provider import SamlCredentialsProvider
 from redshift_connector.redshift_property import RedshiftProperty
@@ -23,6 +20,9 @@ def add_parameter(self: "PingCredentialsProvider", info: RedshiftProperty) -> No
 
     # Required method to grab the SAML Response. Used in base class to refresh temporary credentials.
     def get_saml_assertion(self: "PingCredentialsProvider") -> str:
+        import bs4  # type: ignore
+        import requests
+
         self.check_required_parameters()
 
         if self.partner_sp_id is None:

redshift_connector/plugin/saml_credentials_provider.py

@@ -5,9 +5,6 @@
 import typing
 from abc import ABC, abstractmethod
 
-import boto3  # type: ignore
-import bs4  # type: ignore
-
 from redshift_connector.credentials_holder import CredentialsHolder
 from redshift_connector.error import InterfaceError
 from redshift_connector.plugin.credential_provider_constants import SAML_RESP_NAMESPACES
@@ -70,6 +67,9 @@ def get_credentials(self: "SamlCredentialsProvider") -> CredentialsHolder:
         return credentials
 
     def refresh(self: "SamlCredentialsProvider") -> None:
+        import boto3  # type: ignore
+        import bs4  # type: ignore
+
         try:
             # get SAML assertion from specific identity provider
             saml_assertion = self.get_saml_assertion()
@@ -78,7 +78,6 @@ def refresh(self: "SamlCredentialsProvider") -> None:
             raise InterfaceError(e)
         # decode SAML assertion into xml format
         doc: bytes = base64.b64decode(saml_assertion)
-
         soup = bs4.BeautifulSoup(doc, "xml")
         attrs = soup.findAll("Attribute")
         # extract RoleArn adn PrincipleArn from SAML assertion
@@ -180,6 +179,8 @@ def check_required_parameters(self: "SamlCredentialsProvider") -> None:
             raise InterfaceError("Missing required property: idp_host")
 
     def read_metadata(self: "SamlCredentialsProvider", doc: bytes) -> CredentialsHolder.IamMetadata:
+        import bs4  # type: ignore
+
         try:
             soup = bs4.BeautifulSoup(doc, "xml")
             attrs: typing.Any = []