fix: remove unneccesary boto3 call to verify region (#1042)

Author: faviansamatha

aws_advanced_python_wrapper/aws_secrets_manager_plugin.py

@@ -210,8 +210,7 @@ def _apply_secret_to_properties(self, properties: Properties):
             WrapperProperties.PASSWORD.set(properties, password_value)
 
     def _get_rds_region(self, secret_id: str, props: Properties) -> str:
-        session = self._session if self._session else boto3.Session()
-        region = self._region_utils.get_region(props, WrapperProperties.SECRETS_MANAGER_REGION.name, session=session)
+        region = self._region_utils.get_region(props, WrapperProperties.SECRETS_MANAGER_REGION.name)
 
         if region:
             return region
@@ -221,7 +220,7 @@ def _get_rds_region(self, secret_id: str, props: Properties) -> str:
             region = match.group("region")
 
         if region:
-            return self._region_utils.verify_region(region)
+            return region
         else:
             raise AwsWrapperError(
                 Messages.get_formatted("AwsSecretsManagerPlugin.MissingRequiredConfigParameter",

aws_advanced_python_wrapper/federated_plugin.py

@@ -84,7 +84,7 @@ def _connect(self, host_info: HostInfo, props: Properties, connect_func: Callabl
 
         host = IamAuthUtils.get_iam_host(props, host_info)
         port = IamAuthUtils.get_port(props, host_info, self._plugin_service.database_dialect.default_port)
-        region = self._region_utils.get_region(props, WrapperProperties.IAM_REGION.name, host, self._session)
+        region = self._region_utils.get_region(props, WrapperProperties.IAM_REGION.name, host)
         if not region:
             error_message = "RdsUtils.UnsupportedHostname"
             logger.debug(error_message, host)

aws_advanced_python_wrapper/iam_plugin.py

@@ -78,7 +78,7 @@ def _connect(self, host_info: HostInfo, props: Properties, connect_func: Callabl
             raise AwsWrapperError(Messages.get_formatted("IamAuthPlugin.IsNoneOrEmpty", WrapperProperties.USER.name))
 
         host = IamAuthUtils.get_iam_host(props, host_info)
-        region = self._region_utils.get_region(props, WrapperProperties.IAM_REGION.name, host, self._session)
+        region = self._region_utils.get_region(props, WrapperProperties.IAM_REGION.name, host)
         if not region:
             error_message = "RdsUtils.UnsupportedHostname"
             logger.debug(error_message, host)

aws_advanced_python_wrapper/okta_plugin.py

@@ -80,7 +80,7 @@ def _connect(self, host_info: HostInfo, props: Properties, connect_func: Callabl
 
         host = IamAuthUtils.get_iam_host(props, host_info)
         port = IamAuthUtils.get_port(props, host_info, self._plugin_service.database_dialect.default_port)
-        region = self._region_utils.get_region(props, WrapperProperties.IAM_REGION.name, host, self._session)
+        region = self._region_utils.get_region(props, WrapperProperties.IAM_REGION.name, host)
         if not region:
             error_message = "RdsUtils.UnsupportedHostname"
             logger.debug(error_message, host)

aws_advanced_python_wrapper/utils/region_utils.py

@@ -19,11 +19,7 @@
 if TYPE_CHECKING:
     from aws_advanced_python_wrapper.utils.properties import Properties
 
-from boto3 import Session
-
-from aws_advanced_python_wrapper.errors import AwsWrapperError
 from aws_advanced_python_wrapper.utils.log import Logger
-from aws_advanced_python_wrapper.utils.messages import Messages
 from aws_advanced_python_wrapper.utils.rdsutils import RdsUtils
 
 logger = Logger(__name__)
@@ -36,23 +32,12 @@ def __init__(self):
     def get_region(self,
                    props: Properties,
                    prop_key: str,
-                   hostname: Optional[str] = None,
-                   session: Optional[Session] = None) -> Optional[str]:
+                   hostname: Optional[str] = None) -> Optional[str]:
         region = props.get(prop_key)
         if region:
-            return self.verify_region(region, session)
-
-        return self.get_region_from_hostname(hostname, session)
-
-    def get_region_from_hostname(self, hostname: Optional[str], session: Optional[Session] = None) -> Optional[str]:
-        region = self._rds_utils.get_rds_region(hostname)
-        return self.verify_region(region, session) if region else None
+            return region
 
-    def verify_region(self, region: str, session: Optional[Session] = None) -> str:
-        session = session if session is not None else Session()
-        if region not in session.get_available_regions("rds"):
-            error_message = "AwsSdk.UnsupportedRegion"
-            logger.debug(error_message, region)
-            raise AwsWrapperError(Messages.get_formatted(error_message, region))
+        return self.get_region_from_hostname(hostname)
 
-        return region
+    def get_region_from_hostname(self, hostname: Optional[str]) -> Optional[str]:
+        return self._rds_utils.get_rds_region(hostname)