Add a config property descriptor along with a custom resolver and validators

packages/smithy-aws-core/src/smithy_aws_core/config/custom_resolvers.py

@@ -0,0 +1,51 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from typing import cast
+
+from smithy_core.config.resolver import ConfigResolver
+from smithy_core.retries import RetryStrategyOptions, RetryStrategyType
+
+from smithy_aws_core.config.validators import validate_max_attempts, validate_retry_mode
+
+
+def resolve_retry_strategy(
+    resolver: ConfigResolver,
+) -> tuple[RetryStrategyOptions | None, str | None]:
+    """Resolve retry strategy from multiple config keys.
+
+    Resolves both retry_mode and max_attempts from sources and constructs
+    a RetryStrategyOptions object. This allows the retry strategy to be
+    configured from multiple sources. Example: retry_mode from config file and
+    max_attempts from environment variables.
+
+    :param resolver: The config resolver to use for resolution
+
+    :returns: Tuple of (RetryStrategyOptions, source_name) if both retry_mode and max_attempts
+        are resolved. Returns (None, None) if either value is missing.
+
+        For mixed sources, the source name includes both component sources:
+        "retry_mode=environment, max_attempts=config_file"
+    """
+    # Get retry_mode
+    retry_mode, mode_source = resolver.get("retry_mode")
+
+    # Get max_attempts
+    max_attempts, attempts_source = resolver.get("max_attempts")
+
+    if retry_mode is None or max_attempts is None:
+        return None, None
+
+    retry_mode = validate_retry_mode(retry_mode, mode_source)
+    retry_mode = cast(RetryStrategyType, retry_mode)
+
+    max_attempts = validate_max_attempts(max_attempts, attempts_source)
+
+    options = RetryStrategyOptions(
+        retry_mode=retry_mode,
+        max_attempts=max_attempts,
+    )
+
+    # Construct mixed source string showing where each component came from
+    source = f"retry_mode={mode_source}, max_attempts={attempts_source}"
+
+    return (options, source)

packages/smithy-aws-core/src/smithy_aws_core/config/validators.py

@@ -0,0 +1,131 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import re
+from typing import Any, get_args
+
+from smithy_core.interfaces.retries import RetryStrategy
+from smithy_core.retries import RetryStrategyOptions, RetryStrategyType
+
+
+class ConfigValidationError(ValueError):
+    """Raised when a configuration value fails validation."""
+
+    def __init__(self, key: str, value: Any, reason: str, source: str | None = None):
+        self.key = key
+        self.value = value
+        self.reason = reason
+        self.source = source
+
+        msg = f"Invalid value for '{key}': {value!r}. {reason}"
+        if source:
+            msg += f" (from source: {source})"
+        super().__init__(msg)
+
+
+def validate_host(host_name: Any, source: str | None = None) -> str:
+    """Validate host name format.
+
+    :param host_name: The value to validate
+    :param source: The config source that provided this value
+
+    :returns: The validated value
+
+    :raises ConfigValidationError: If the value format is invalid
+    """
+    if not isinstance(host_name, str):
+        raise ConfigValidationError(
+            "host",
+            host_name,
+            f"Host must be a string, got {type(host_name).__name__}",
+            source,
+        )
+
+    pattern = r"^(?![0-9]+$)(?!-)[a-zA-Z0-9-]{,63}(?<!-)$"
+
+    if not re.match(pattern, host_name):
+        raise ConfigValidationError(
+            "host",
+            host_name,
+            "Host doesn't match the pattern.",
+            source,
+        )
+    return host_name
+
+
+def validate_retry_mode(retry_mode: Any, source: str | None = None) -> str:
+    """Validate retry mode.
+
+    Valid values: 'standard', 'simple'
+
+    :param retry_mode: The retry mode value to validate
+    :param source: The source that provided this value
+
+    :returns: The validated retry mode string
+
+    :raises: ConfigValidationError: If the retry mode is invalid
+    """
+    if not isinstance(retry_mode, str):
+        raise ConfigValidationError(
+            "retry_mode",
+            retry_mode,
+            f"Retry mode must be a string, got {type(retry_mode).__name__}",
+            source,
+        )
+
+    valid_modes = get_args(RetryStrategyType)
+
+    if retry_mode not in valid_modes:
+        raise ConfigValidationError(
+            "retry_mode",
+            retry_mode,
+            f"Retry mode must be one of {valid_modes}, got {retry_mode}",
+            source,
+        )
+
+    return retry_mode
+
+
+def validate_max_attempts(max_attempts: str | int, source: str | None = None) -> int:
+    """Validate and convert max_attempts to integer.
+
+    :param max_attempts: The max attempts value (string or int)
+    :param source: The source that provided this value
+
+    :returns: The validated max_attempts as an integer
+
+    :raises ConfigValidationError: If the value cannot be converted to an integer
+    """
+    try:
+        return int(max_attempts)
+    except (ValueError, TypeError):
+        raise ConfigValidationError(
+            "max_attempts",
+            max_attempts,
+            f"max_attempts must be a number, got {type(max_attempts).__name__}",
+            source,
+        )
+
+
+def validate_retry_strategy(
+    value: Any, source: str | None = None
+) -> RetryStrategy | RetryStrategyOptions:
+    """Validate retry strategy configuration.
+
+    :param value: The retry strategy value to validate
+    :param source: The source that provided this value
+
+    :returns: The validated retry strategy (RetryStrategy or RetryStrategyOptions)
+
+    :raises: ConfigValidationError: If the value is not a valid retry strategy type
+    """
+
+    if isinstance(value, RetryStrategy | RetryStrategyOptions):
+        return value
+
+    raise ConfigValidationError(
+        "retry_strategy",
+        value,
+        f"retry_strategy must be RetryStrategy or RetryStrategyOptions, got {type(value).__name__}",
+        source,
+    )

packages/smithy-aws-core/tests/unit/config/test_custom_resolver.py

@@ -0,0 +1,93 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from typing import Any
+
+from smithy_aws_core.config.custom_resolvers import resolve_retry_strategy
+from smithy_core.config.resolver import ConfigResolver
+from smithy_core.retries import RetryStrategyOptions
+
+
+class StubSource:
+    """A simple ConfigSource implementation for testing."""
+
+    def __init__(self, source_name: str, data: dict[str, Any] | None = None) -> None:
+        self._name = source_name
+        self._data = data or {}
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    def get(self, key: str) -> Any | None:
+        return self._data.get(key)
+
+
+class TestResolveCustomResolverRetryStrategy:
+    """Test suite for complex configuration resolution"""
+
+    def test_resolves_from_both_values(self) -> None:
+        # When both retry mode and max attempts are set
+        # It should use source names for both values
+        source = StubSource(
+            "environment", {"retry_mode": "standard", "max_attempts": "3"}
+        )
+        resolver = ConfigResolver(sources=[source])
+
+        result, source_name = resolve_retry_strategy(resolver)
+
+        assert isinstance(result, RetryStrategyOptions)
+        assert result.retry_mode == "standard"
+        assert result.max_attempts == 3
+        assert source_name == "retry_mode=environment, max_attempts=environment"
+
+    def test_tracks_different_sources_for_each_component(self) -> None:
+        source1 = StubSource("environment", {"retry_mode": "standard"})
+        source2 = StubSource("config_file", {"max_attempts": "5"})
+        resolver = ConfigResolver(sources=[source1, source2])
+
+        result, source_name = resolve_retry_strategy(resolver)
+
+        assert isinstance(result, RetryStrategyOptions)
+        assert result.retry_mode == "standard"
+        assert result.max_attempts == 5
+        assert source_name == "retry_mode=environment, max_attempts=config_file"
+
+    def test_converts_max_attempts_string_to_int(self) -> None:
+        source = StubSource(
+            "environment", {"max_attempts": "10", "retry_mode": "standard"}
+        )
+        resolver = ConfigResolver(sources=[source])
+
+        result, _ = resolve_retry_strategy(resolver)
+
+        assert isinstance(result, RetryStrategyOptions)
+        assert result.max_attempts == 10
+        assert isinstance(result.max_attempts, int)
+
+    def test_returns_none_when_only_retry_mode_set(self) -> None:
+        source = StubSource("environment", {"retry_mode": "standard"})
+        resolver = ConfigResolver(sources=[source])
+
+        result, source_name = resolve_retry_strategy(resolver)
+
+        assert result is None
+        assert source_name is None
+
+    def test_returns_none_when_only_max_attempts_set(self) -> None:
+        source = StubSource("environment", {"max_attempts": "5"})
+        resolver = ConfigResolver(sources=[source])
+
+        result, source_name = resolve_retry_strategy(resolver)
+
+        assert result is None
+        assert source_name is None
+
+    def test_returns_none_when_both_values_missing(self) -> None:
+        source = StubSource("environment", {})
+        resolver = ConfigResolver(sources=[source])
+
+        result, source_name = resolve_retry_strategy(resolver)
+
+        assert result is None
+        assert source_name is None

packages/smithy-aws-core/tests/unit/config/test_validators.py

@@ -0,0 +1,47 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Unit tests for AWS configuration validators."""
+
+import pytest
+from smithy_aws_core.config.validators import (
+    ConfigValidationError,
+    validate_host,
+    validate_max_attempts,
+    validate_retry_mode,
+)
+
+
+class TestValidators:
+    @pytest.mark.parametrize("region", ["us-east-1", "eu-west-1", "ap-south-1"])
+    def test_validate_region_accepts_valid_values(self, region: str) -> None:
+        assert validate_host(region) == region
+
+    @pytest.mark.parametrize("invalid", ["-invalid", "-east", "12345", 1234])
+    def test_validate_region_rejects_invalid_values(self, invalid: str) -> None:
+        with pytest.raises(ConfigValidationError):
+            validate_host(invalid)
+
+    @pytest.mark.parametrize("mode", ["standard", "simple"])
+    def test_validate_retry_mode_accepts_valid_values(self, mode: str) -> None:
+        assert validate_retry_mode(mode) == mode
+
+    @pytest.mark.parametrize("invalid_mode", ["some_retry", "some_retry_one", ""])
+    def test_validate_retry_mode_rejects_invalid_values(
+        self, invalid_mode: str
+    ) -> None:
+        with pytest.raises(ConfigValidationError):
+            validate_retry_mode(invalid_mode)
+
+    def test_validate_invalid_max_attempts_raises_error(self) -> None:
+        with pytest.raises(
+            ConfigValidationError, match="max_attempts must be a number"
+        ):
+            validate_max_attempts("abcd")
+
+    def test_invalid_retry_mode_error_message(self) -> None:
+        with pytest.raises(ConfigValidationError) as exc_info:
+            validate_retry_mode("random_mode")
+        assert (
+            "Invalid value for 'retry_mode': 'random_mode'. Retry mode must be one "
+            "of ('simple', 'standard'), got random_mode" in str(exc_info.value)
+        )