Add config source interface and environment variable source for config resolution (#640)

* Add config source interface and environment variable source

Author: ubaskota

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

@@ -0,0 +1,7 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from .sources import EnvironmentSource
+
+__all__ = [
+    "EnvironmentSource",
+]

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

@@ -0,0 +1,35 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+import os
+
+
+class EnvironmentSource:
+    """Configuration from environment variables."""
+
+    def __init__(self, prefix: str = "AWS_"):
+        """Initialize the EnvironmentSource with environment variable prefix.
+
+        :param prefix: Prefix for environment variables (default: 'AWS_')
+        """
+        self._prefix = prefix
+
+    @property
+    def name(self) -> str:
+        """Returns the source name."""
+        return "environment"
+
+    def get(self, key: str) -> str | None:
+        """Returns a configuration value from environment variables.
+
+        The key is transformed to uppercase and prefixed (e.g., 'region' → 'AWS_REGION').
+
+        :param key: The standard configuration key (e.g., 'region', 'retry_mode').
+
+        :returns: The value from the environment variable (or empty string if set to empty),
+              or None if the variable is not set.
+        """
+        env_var = f"{self._prefix}{key.upper()}"
+        config_value = os.environ.get(env_var)
+        if config_value is None:
+            return None
+        return config_value.strip()

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

@@ -0,0 +1,86 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+import os
+from unittest.mock import patch
+
+from smithy_aws_core.config.sources import EnvironmentSource
+
+
+class TestEnvironmentSource:
+    def test_source_name(self):
+        source = EnvironmentSource()
+        assert source.name == "environment"
+
+    def test_get_region_from_aws_region(self):
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=True):
+            source = EnvironmentSource()
+            value = source.get("region")
+            assert value == "us-west-2"
+
+    def test_get_returns_none_when_env_var_not_set(self):
+        with patch.dict(os.environ, {}, clear=True):
+            source = EnvironmentSource()
+            value = source.get("region")
+            assert value is None
+
+    def test_get_returns_none_for_unknown_key(self):
+        source = EnvironmentSource()
+        value = source.get("unknown_config_key")
+        assert value is None
+
+    def test_get_handles_empty_string_env_var(self):
+        with patch.dict(os.environ, {"AWS_REGION": ""}, clear=True):
+            source = EnvironmentSource()
+            value = source.get("region")
+            assert value == ""
+
+    def test_get_handles_whitespace_env_var(self):
+        with patch.dict(os.environ, {"AWS_REGION": "  us-west-2  "}, clear=True):
+            source = EnvironmentSource()
+            value = source.get("region")
+            # Whitespaces should be stripped
+            assert value == "us-west-2"
+
+    def test_get_handles_whole_whitespace_env_var(self):
+        with patch.dict(os.environ, {"AWS_REGION": "  "}, clear=True):
+            source = EnvironmentSource()
+            value = source.get("region")
+            # Whitespaces should be stripped
+            assert value == ""
+
+    def test_multiple_keys_with_different_env_vars(self):
+        env_vars = {"AWS_REGION": "eu-west-1", "AWS_RETRY_MODE": "standard"}
+        with patch.dict(os.environ, env_vars, clear=True):
+            source = EnvironmentSource()
+
+            region = source.get("region")
+            retry_mode = source.get("retry_mode")
+
+            assert region == "eu-west-1"
+            assert retry_mode == "standard"
+
+    def test_get_is_idempotent(self):
+        with patch.dict(os.environ, {"AWS_REGION": "ap-south-1"}, clear=True):
+            source = EnvironmentSource()
+            # Calling get on source multiple times should return the same value
+            value1 = source.get("region")
+            value2 = source.get("region")
+            value3 = source.get("region")
+
+            assert value1 == value2 == value3 == "ap-south-1"
+
+    def test_source_does_not_cache_env_vars(self):
+        source = EnvironmentSource()
+
+        # First read
+        with patch.dict(os.environ, {"AWS_REGION": "us-east-1"}, clear=True):
+            value1 = source.get("region")
+            assert value1 == "us-east-1"
+
+        # Environment changes
+        with patch.dict(os.environ, {"AWS_REGION": "us-west-2"}, clear=False):
+            value2 = source.get("region")
+            assert value2 == "us-west-2"
+
+        # Source reads from os.environ and not from cache
+        assert value1 != value2

packages/smithy-core/src/smithy_core/interfaces/config.py

@@ -0,0 +1,27 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+from typing import Any, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class ConfigSource(Protocol):
+    """Protocol for configuration sources that provide values from various locations
+    like environment variables and configuration files.
+    """
+
+    @property
+    def name(self) -> str:
+        """Returns a string identifying the source.
+
+        :returns: A string identifier for this source.
+        """
+        ...
+
+    def get(self, key: str) -> Any | None:
+        """Returns a configuration value from the source.
+
+        :param key: The configuration key to retrieve (e.g., 'region')
+
+        :returns: The value associated with the key, or None if not found.
+        """
+        ...