refactor: config properties retrieval and validation

Author: karenc-bq

aws_advanced_python_wrapper/simple_read_write_splitting_plugin.py

@@ -41,53 +41,24 @@ class EndpointBasedConnectionHandler(ConnectionHandler):
     """Endpoint based implementation of connection handling logic."""
 
     def __init__(self, plugin_service: PluginService, props: Properties):
-        srw_read_endpoint = WrapperProperties.SRW_READ_ENDPOINT.get(props)
-        if srw_read_endpoint is None:
-            raise AwsWrapperError(
-                Messages.get_formatted(
-                    "SimpleReadWriteSplittingPlugin.MissingRequiredConfigParameter",
-                    WrapperProperties.SRW_READ_ENDPOINT.name,
-                )
-            )
-        self._read_endpoint: str = srw_read_endpoint
-
-        srw_write_endpoint = WrapperProperties.SRW_WRITE_ENDPOINT.get(props)
-        if srw_write_endpoint is None:
-            raise AwsWrapperError(
-                Messages.get_formatted(
-                    "SimpleReadWriteSplittingPlugin.MissingRequiredConfigParameter",
-                    WrapperProperties.SRW_WRITE_ENDPOINT.name,
-                )
-            )
-        self._write_endpoint: str = srw_write_endpoint
+        self._read_endpoint: str = EndpointBasedConnectionHandler._verify_parameter(
+            WrapperProperties.SRW_READ_ENDPOINT, props, str, required=True
+        )
+        self._write_endpoint: str = EndpointBasedConnectionHandler._verify_parameter(
+            WrapperProperties.SRW_WRITE_ENDPOINT, props, str, required=True
+        )
 
-        self._verify_new_connections: bool = (
-            WrapperProperties.SRW_VERIFY_NEW_CONNECTIONS.get_bool(props)
+        self._verify_new_connections: bool = self._verify_parameter(
+            WrapperProperties.SRW_VERIFY_NEW_CONNECTIONS, props, bool
         )
-        if self._verify_new_connections is True:
-            srw_connect_retry_timeout_ms: int = (
-                WrapperProperties.SRW_CONNECT_RETRY_TIMEOUT_MS.get_int(props)
-            )
-            if srw_connect_retry_timeout_ms <= 0:
-                raise ValueError(
-                    Messages.get_formatted(
-                        "SimpleReadWriteSplittingPlugin.IncorrectConfiguration",
-                        WrapperProperties.SRW_CONNECT_RETRY_TIMEOUT_MS.name,
-                    )
-                )
-            self._connect_retry_timeout_ms: int = srw_connect_retry_timeout_ms
 
-            srw_connect_retry_interval_ms: int = (
-                WrapperProperties.SRW_CONNECT_RETRY_INTERVAL_MS.get_int(props)
+        if self._verify_new_connections:
+            self._connect_retry_timeout_ms: int = self._verify_parameter(
+                WrapperProperties.SRW_CONNECT_RETRY_TIMEOUT_MS, props, int, lambda x: x > 0
+            )
+            self._connect_retry_interval_ms: int = self._verify_parameter(
+                WrapperProperties.SRW_CONNECT_RETRY_INTERVAL_MS, props, int, lambda x: x > 0
             )
-            if srw_connect_retry_interval_ms <= 0:
-                raise ValueError(
-                    Messages.get_formatted(
-                        "SimpleReadWriteSplittingPlugin.IncorrectConfiguration",
-                        WrapperProperties.SRW_CONNECT_RETRY_INTERVAL_MS.name,
-                    )
-                )
-            self._connect_retry_interval_ms: int = srw_connect_retry_interval_ms
 
             self._verify_opened_connection_type: Optional[HostRole] = (
                 EndpointBasedConnectionHandler._parse_connection_type(
@@ -305,6 +276,27 @@ def _create_host_info(self, endpoint, role: HostRole) -> HostInfo:
             host=host, port=port, role=role, availability=HostAvailability.AVAILABLE
         )
 
+    @staticmethod
+    def _verify_parameter(prop, props, expected_type, validator=None, required=False):
+        value = prop.get_type(props, expected_type)
+        if required:
+            if value is None:
+                raise AwsWrapperError(
+                    Messages.get_formatted(
+                        "SimpleReadWriteSplittingPlugin.MissingRequiredConfigParameter",
+                        prop.name,
+                    )
+                )
+
+        if validator and not validator(value):
+            raise ValueError(
+                Messages.get_formatted(
+                    "SimpleReadWriteSplittingPlugin.IncorrectConfiguration",
+                    prop.name,
+                )
+            )
+        return value
+
     def _delay(self):
         sleep(self._connect_retry_interval_ms / 1000)
 

aws_advanced_python_wrapper/utils/properties.py

@@ -11,8 +11,9 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
+
 import copy
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, TypeVar, Type
 from urllib.parse import unquote
 
 from aws_advanced_python_wrapper.errors import AwsWrapperError
@@ -25,6 +26,8 @@ def put_if_absent(self, key: str, value: Any):
             self[key] = value
 
 
+T = TypeVar('T')
+
 class WrapperProperty:
     def __init__(
         self, name: str, description: str, default_value: Optional[Any] = None
@@ -41,34 +44,29 @@ def get(self, props: Properties) -> Optional[str]:
             return props.get(self.name, self.default_value)
         return props.get(self.name)
 
+    def get_typed(self, props: Properties, type_class: Type[T]) -> T:
+        value = props.get(self.name, self.default_value) if self.default_value else props.get(self.name)
+        if value is None:
+            return -1 if type_class in (int, float) else None
+        if type_class == bool:
+            if isinstance(value, bool):
+                return value
+            return value.lower() == "true" if isinstance(value, str) else bool(value)
+        return type_class(value)
+
     def get_or_default(self, props: Properties) -> str:
         if not self.default_value:
             raise ValueError(f"No default value found for property {self}")
         return props.get(self.name, self.default_value)
 
     def get_int(self, props: Properties) -> int:
-        if self.default_value:
-            return int(props.get(self.name, self.default_value))
-
-        val = props.get(self.name)
-        return int(val) if val else -1
+        return self.get_typed(props, int)
 
     def get_float(self, props: Properties) -> float:
-        if self.default_value:
-            return float(props.get(self.name, self.default_value))
-
-        val = props.get(self.name)
-        return float(val) if val else -1
+        return self.get_typed(props, float)
 
     def get_bool(self, props: Properties) -> bool:
-        if not self.default_value:
-            value = props.get(self.name)
-        else:
-            value = props.get(self.name, self.default_value)
-        if isinstance(value, bool):
-            return value
-        else:
-            return value is not None and value.lower() == "true"
+        return self.get_typed(props, bool)
 
     def set(self, props: Properties, value: Any):
         props[self.name] = value