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,52 @@
+# 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_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 or None, source_name or None).
+        Returns (None, None) if neither retry_mode nor max_attempts is set.
+
+        For mixed sources, the source string 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 neither is set, return None
+    if retry_mode is None and max_attempts is None:
+        return (None, None)
+
+    if retry_mode is not None:
+        retry_mode = validate_retry_mode(retry_mode, mode_source)
+        retry_mode = cast(RetryStrategyType, retry_mode)
+
+    # Construct options with defaults
+    options = RetryStrategyOptions(
+        retry_mode=retry_mode or "standard",
+        max_attempts=int(max_attempts) if max_attempts else None,
+    )
+
+    # Construct mixed source string showing where each component came from
+    source = f"retry_mode={mode_source or 'unresolved'}, max_attempts={attempts_source or 'unresolved'}"
+
+    return (options, source)

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

@@ -0,0 +1,116 @@
+# 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_region(region_name: Any, source: str | None = None) -> str | None:
+    """Validate AWS region format.
+
+    Valid formats:
+    - us-east-1, us-west-2, eu-west-1, etc.
+    - Pattern: {partition}-{region}-{number}
+
+    :param region_name: The region value to validate
+    :param source: The config source that provided this value
+
+    :returns: The validated region string, or None if value is None
+
+    :raises ConfigValidationError: If the region format is invalid
+    """
+    if not isinstance(region_name, str):
+        raise ConfigValidationError(
+            "region",
+            region_name,
+            f"Region must be a string, got {type(region_name).__name__}",
+            source,
+        )
+
+    pattern = r"^(?![0-9]+$)(?!-)[a-zA-Z0-9-]{,63}(?<!-)$"
+
+    if not re.match(pattern, region_name):
+        raise ConfigValidationError(
+            "region",
+            region_name,
+            "Region doesn't match the pattern (e.g., 'us-west-2', 'eu-central-1')",
+            source,
+        )
+    return region_name
+
+
+def validate_retry_mode(retry_mode: Any, source: str | None = None) -> str | None:
+    """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, or None if value is None
+
+    :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 = set(get_args(RetryStrategyType))
+
+    if retry_mode not in valid_modes:
+        raise ConfigValidationError(
+            "retry_mode",
+            retry_mode,
+            f"Retry mode must be one of {RetryStrategyType}, got {retry_mode}",
+            source,
+        )
+
+    return retry_mode
+
+
+def validate_retry_strategy(value: Any, source: str | None = None) -> Any:
+    """Validate retry strategy configuration.
+
+    :param value: The retry strategy value to validate (None is allowed and returns None)
+    :param source: The source that provided this value (for error messages)
+
+    :returns: The validated retry strategy (RetryStrategy or RetryStrategyOptions)
+
+    :raises: ConfigValidationError: If the value is not a valid retry strategy type
+    """
+    # Allow RetryStrategy instances
+    if isinstance(value, RetryStrategy):
+        return value
+
+    # Allow RetryStrategyOptions instances
+    if isinstance(value, RetryStrategyOptions):
+        return value
+
+    raise ConfigValidationError(
+        "retry_strategy",
+        value,
+        f"Retry strategy must be a RetryStrategy or RetryStrategyOptions got {type(value).__name__}",
+        source,
+    )

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

@@ -0,0 +1,109 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for custom resolver functions."""
+
+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_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 isinstance(result, RetryStrategyOptions)
+        assert result.retry_mode == "standard"
+        assert result.max_attempts is None
+        assert source_name == "retry_mode=environment, max_attempts=unresolved"
+
+    def test_resolves_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 isinstance(result, RetryStrategyOptions)
+        assert result.retry_mode == "standard"
+        assert result.max_attempts == 5
+        assert source_name == "retry_mode=unresolved, max_attempts=environment"
+
+    def test_resolves_from_both_values_when_set(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_returns_none_when_neither_value_set(self) -> None:
+        source = StubSource("environment", {})
+        resolver = ConfigResolver(sources=[source])
+
+        result, source_name = resolve_retry_strategy(resolver)
+        # It should return (None, None) when values not set
+        assert result is None
+        assert source_name is None
+
+    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_tracks_source_when_only_max_attempts_set(self) -> None:
+        source1 = StubSource("environment", {})
+        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"  # Default
+        assert result.max_attempts == 5
+        assert source_name == "retry_mode=unresolved, max_attempts=config_file"
+
+    def test_converts_max_attempts_string_to_int(self) -> None:
+        source = StubSource("environment", {"max_attempts": "10"})
+        resolver = ConfigResolver(sources=[source])
+
+        result, _ = resolve_retry_strategy(resolver)
+
+        assert isinstance(result, RetryStrategyOptions)
+        assert result.max_attempts == 10
+        assert isinstance(result.max_attempts, int)

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

@@ -0,0 +1,32 @@
+# 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_region,
+    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_region(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_region(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)