Replace raw source strings with typed SourceInfo dataclasses (#652)

* Replace raw source strings with typed SourceInfo dataclasses

Author: ubaskota

codegen/core/src/main/java/software/amazon/smithy/python/codegen/generators/ConfigGenerator.java

@@ -343,9 +343,9 @@ private void generateConfig(GenerationContext context, PythonWriter writer) {
 
         // Only add config resolution imports if there are descriptor properties
         if (hasDescriptors) {
-            writer.addDependency(SmithyPythonDependency.SMITHY_CORE);
-            writer.addImport("smithy_core.config.property", "ConfigProperty");
-            writer.addImport("smithy_core.config.resolver", "ConfigResolver");
+            writer.addDependency(SmithyPythonDependency.SMITHY_AWS_CORE);
+            writer.addImport("smithy_aws_core.config.property", "ConfigProperty");
+            writer.addImport("smithy_aws_core.config.resolver", "ConfigResolver");
             writer.addImport("smithy_aws_core.config.sources", "EnvironmentSource");
 
             // Add validator and resolver imports for properties that use descriptors
@@ -580,17 +580,19 @@ public void write(PythonWriter writer, String previousText, ConfigSection sectio
             }
 
             writer.write(previousText);
+            writer.addImport("smithy_aws_core.config.source_info", "SourceInfo");
 
             writer.write("""
 
-                        def get_source(self, key: str) -> str | None:
+                        def get_source(self, key: str) -> SourceInfo | None:
                             \"""Get the source that provided a configuration value.
 
                             Args:
                                 key: The configuration key (e.g., 'region', 'retry_strategy')
 
                             Returns:
-                                The source name ('instance', 'environment', etc.),
+                                The source info (SimpleSource('source_name') or
+                                ComplexSource({"retry_mode": "source1", "max_attempts": "source2"})),
                                 or None if the key hasn't been resolved yet.
                             \"""
                             cached = self.__dict__.get(f'_cache_{key}')

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

@@ -1,15 +1,16 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-from smithy_core.config.resolver import ConfigResolver
 from smithy_core.retries import RetryStrategyOptions
 
+from smithy_aws_core.config.resolver import ConfigResolver
+from smithy_aws_core.config.source_info import ComplexSource, SourceName
 from smithy_aws_core.config.validators import validate_max_attempts, validate_retry_mode
 
 
 def resolve_retry_strategy(
     resolver: ConfigResolver,
-) -> tuple[RetryStrategyOptions | None, str | None]:
+) -> tuple[RetryStrategyOptions | None, ComplexSource | None]:
     """Resolve retry strategy from multiple config keys.
 
     Resolves both retry_mode and max_attempts from sources and constructs
@@ -23,7 +24,7 @@ def resolve_retry_strategy(
         are resolved. Returns (None, None) if both values are missing.
 
         For mixed sources, the source name includes both component sources:
-        "retry_mode=environment, max_attempts=config_file"
+        {"retry_mode": "environment", "max_attempts": "default"}
     """
 
     retry_mode, mode_source = resolver.get("retry_mode")
@@ -45,6 +46,13 @@ def resolve_retry_strategy(
     )
 
     # Construct mixed source string showing where each component came from
-    source = f"retry_mode={mode_source or 'default'}, max_attempts={attempts_source or 'default'}"
+    source = ComplexSource(
+        {
+            "retry_mode": mode_source.name if mode_source else SourceName.DEFAULT,
+            "max_attempts": attempts_source.name
+            if attempts_source
+            else SourceName.DEFAULT,
+        }
+    )
 
     return (options, source)

…-core/src/smithy_core/config/property.py …e/src/smithy_aws_core/config/property.pypackages/smithy-core/src/smithy_core/config/property.py renamed to packages/smithy-aws-core/src/smithy_aws_core/config/property.py packages/smithy-core/src/smithy_core/config/property.py renamed to packages/smithy-aws-core/src/smithy_aws_core/config/property.py

@@ -3,7 +3,8 @@
 from collections.abc import Callable
 from typing import Any
 
-from smithy_core.config.resolver import ConfigResolver
+from smithy_aws_core.config.resolver import ConfigResolver
+from smithy_aws_core.config.source_info import SimpleSource, SourceInfo, SourceName
 
 
 class ConfigProperty:
@@ -27,8 +28,9 @@ def __init__(self):
     def __init__(
         self,
         key: str,
-        validator: Callable[[Any, str | None], Any] | None = None,
-        resolver_func: Callable[[ConfigResolver], tuple[Any, str | None]] | None = None,
+        validator: Callable[[Any, SourceInfo | None], Any] | None = None,
+        resolver_func: Callable[[ConfigResolver], tuple[Any, SourceInfo | None]]
+        | None = None,
         default_value: Any = None,
     ):
         """Initialize config property descriptor.
@@ -78,7 +80,7 @@ def __get__(self, obj: Any, objtype: type | None = None) -> Any:
 
         if value is None:
             value = self.default_value
-            source = "default"
+            source = SimpleSource(SourceName.DEFAULT)
 
         if self.validator:
             value = self.validator(value, source)
@@ -99,7 +101,11 @@ def __set__(self, obj: Any, value: Any) -> None:
         # Determine source based on when the value was set
         # If cache already exists, it means it was not set during initialization
         # In that case source will be set to in-code
-        source = "in-code" if hasattr(obj, self.cache_attr) else "instance"
+        source = (
+            SimpleSource(SourceName.IN_CODE)
+            if hasattr(obj, self.cache_attr)
+            else SimpleSource(SourceName.INSTANCE)
+        )
         if self.validator:
             value = self.validator(value, source)
 

…-core/src/smithy_core/config/resolver.py …e/src/smithy_aws_core/config/resolver.pypackages/smithy-core/src/smithy_core/config/resolver.py renamed to packages/smithy-aws-core/src/smithy_aws_core/config/resolver.py packages/smithy-core/src/smithy_core/config/resolver.py renamed to packages/smithy-aws-core/src/smithy_aws_core/config/resolver.py

@@ -5,6 +5,8 @@
 
 from smithy_core.interfaces.config import ConfigSource
 
+from smithy_aws_core.config.source_info import SimpleSource
+
 
 class ConfigResolver:
     """Resolves configuration values from multiple sources.
@@ -21,7 +23,7 @@ def __init__(self, sources: Sequence[ConfigSource]) -> None:
         """
         self._sources = sources
 
-    def get(self, key: str) -> tuple[Any, str | None]:
+    def get(self, key: str) -> tuple[Any, SimpleSource | None]:
         """Resolve a configuration value from sources by iterating through them in precedence order.
 
         :param key: The configuration key to resolve (e.g., 'retry_mode')
@@ -32,5 +34,5 @@ def get(self, key: str) -> tuple[Any, str | None]:
         for source in self._sources:
             value = source.get(key)
             if value is not None:
-                return (value, source.name)
+                return (value, SimpleSource(source.name))
         return (None, None)

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

@@ -0,0 +1,48 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class SourceName(StrEnum):
+    """Known source names for config value provenance tracking."""
+
+    INSTANCE = "instance"  # value provided via Config constructor
+
+    IN_CODE = "in-code"  # value set via setter after Config construction
+
+    ENVIRONMENT = "environment"  # value resolved from environment variable
+
+    DEFAULT = "default"  # value fall back to default
+
+
+@dataclass(frozen=True)
+class SimpleSource:
+    """Source info for a config value resolved from a single source.
+
+    Examples: region from environment, max_attempts from config file.
+    """
+
+    # TODO: Currently only environment variable is implemented as a config
+    # source. Tests use raw strings (e.g., "environment", "config_file") as
+    # source names to simulate multi-source scenarios. Once additional
+    # config sources are implemented, update the `name` parameter type
+    # from `str` to `SourceName` and replace raw strings in tests with
+    # the corresponding enum values.
+    name: str
+
+
+@dataclass(frozen=True)
+class ComplexSource:
+    """Source info for a config value resolved from multiple sources.
+
+    Used when a config property is composed of multiple sources.
+    Example: retry_strategy is composed of retry_mode and max_attempts and they both
+    could be from different sources: {"retry_mode": "environment", "max_attempts": "config_file"}
+    """
+
+    components: dict[str, str]
+
+
+SourceInfo = SimpleSource | ComplexSource

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

@@ -2,6 +2,8 @@
 # SPDX-License-Identifier: Apache-2.0
 import os
 
+from smithy_aws_core.config.source_info import SourceName
+
 
 class EnvironmentSource:
     """Configuration from environment variables."""
@@ -16,7 +18,7 @@ def __init__(self, prefix: str = "AWS_"):
     @property
     def name(self) -> str:
         """Returns the source name."""
-        return "environment"
+        return SourceName.ENVIRONMENT
 
     def get(self, key: str) -> str | None:
         """Returns a configuration value from environment variables.

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

@@ -7,11 +7,15 @@
 from smithy_core.interfaces.retries import RetryStrategy
 from smithy_core.retries import RetryStrategyOptions, RetryStrategyType
 
+from smithy_aws_core.config.source_info import SourceInfo
+
 
 class ConfigValidationError(ValueError):
     """Raised when a configuration value fails validation."""
 
-    def __init__(self, key: str, value: Any, reason: str, source: str | None = None):
+    def __init__(
+        self, key: str, value: Any, reason: str, source: SourceInfo | None = None
+    ):
         self.key = key
         self.value = value
         self.reason = reason
@@ -23,7 +27,7 @@ def __init__(self, key: str, value: Any, reason: str, source: str | None = None)
         super().__init__(msg)
 
 
-def validate_region(region: str | None, source: str | None = None) -> str:
+def validate_region(region: str | None, source: SourceInfo | None = None) -> str:
     """Validate region name format.
 
     :param region: The value to validate
@@ -53,7 +57,7 @@ def validate_region(region: str | None, source: str | None = None) -> str:
     return region
 
 
-def validate_retry_mode(retry_mode: str, source: str | None = None) -> str:
+def validate_retry_mode(retry_mode: str, source: SourceInfo | None = None) -> str:
     """Validate retry mode.
 
     Valid values: 'standard'
@@ -85,7 +89,9 @@ def validate_retry_mode(retry_mode: str, source: str | None = None) -> str:
     return retry_mode
 
 
-def validate_max_attempts(max_attempts: str | int, source: str | None = None) -> int:
+def validate_max_attempts(
+    max_attempts: str | int, source: SourceInfo | None = None
+) -> int:
     """Validate and convert max_attempts to integer.
 
     :param max_attempts: The max attempts value (string or int)
@@ -117,7 +123,7 @@ def validate_max_attempts(max_attempts: str | int, source: str | None = None) ->
 
 
 def validate_retry_strategy(
-    value: Any, source: str | None = None
+    value: Any, source: SourceInfo | None = None
 ) -> RetryStrategy | RetryStrategyOptions:
     """Validate retry strategy configuration.
 

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

@@ -4,7 +4,8 @@
 from typing import Any
 
 from smithy_aws_core.config.custom_resolvers import resolve_retry_strategy
-from smithy_core.config.resolver import ConfigResolver
+from smithy_aws_core.config.resolver import ConfigResolver
+from smithy_aws_core.config.source_info import ComplexSource
 from smithy_core.retries import RetryStrategyOptions
 
 
@@ -39,7 +40,9 @@ def test_resolves_from_both_values(self) -> None:
         assert isinstance(result, RetryStrategyOptions)
         assert result.retry_mode == "standard"
         assert result.max_attempts == 3
-        assert source_name == "retry_mode=environment, max_attempts=environment"
+        assert source_name == ComplexSource(
+            {"retry_mode": "environment", "max_attempts": "environment"}
+        )
 
     def test_tracks_different_sources_for_each_component(self) -> None:
         source1 = StubSource("environment", {"retry_mode": "standard"})
@@ -51,7 +54,9 @@ def test_tracks_different_sources_for_each_component(self) -> None:
         assert isinstance(result, RetryStrategyOptions)
         assert result.retry_mode == "standard"
         assert result.max_attempts == 5
-        assert source_name == "retry_mode=environment, max_attempts=config_file"
+        assert source_name == ComplexSource(
+            {"retry_mode": "environment", "max_attempts": "config_file"}
+        )
 
     def test_converts_max_attempts_string_to_int(self) -> None:
         source = StubSource(
@@ -76,7 +81,9 @@ def test_returns_strategy_when_only_retry_mode_set(self) -> None:
         # None for max_attempts means the RetryStrategy will use its
         # own default max_attempts value for the set retry_mode
         assert result.max_attempts is None
-        assert source_name == "retry_mode=environment, max_attempts=default"
+        assert source_name == ComplexSource(
+            {"retry_mode": "environment", "max_attempts": "default"}
+        )
 
     def test_returns_strategy_when_only_max_attempts_set(self) -> None:
         source = StubSource("environment", {"max_attempts": "5"})
@@ -87,7 +94,9 @@ def test_returns_strategy_when_only_max_attempts_set(self) -> None:
         assert isinstance(result, RetryStrategyOptions)
         assert result.max_attempts == 5
         assert result.retry_mode == "standard"
-        assert source_name == "retry_mode=default, max_attempts=environment"
+        assert source_name == ComplexSource(
+            {"retry_mode": "default", "max_attempts": "environment"}
+        )
 
     def test_returns_none_when_both_values_missing(self) -> None:
         source = StubSource("environment", {})