Add config source interface and environment variable source

Author: ubaskota

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

@@ -0,0 +1,5 @@
+from .sources import EnvironmentSource
+
+__all__ = [
+    "EnvironmentSource",
+]

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

@@ -0,0 +1,34 @@
+import os
+from typing import Any
+
+
+class EnvironmentSource:
+    """Configuration from environment variables."""
+
+    SOURCE = "environment"
+
+    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 self.SOURCE
+
+    def get(self, key: str) -> Any | None:
+        """Returns a configuration value from environment variables.
+
+        :param key: The standard configuration key (e.g., 'region', 'retry_mode').
+
+        :returns: The value from the corresponding environment variable, or None if not set or empty.
+        """
+        env_var = f"{self._prefix}{key.upper()}"
+        config_value = os.environ.get(env_var)
+        if config_value is None:
+            return None
+        stripped = config_value.strip()
+        return stripped if stripped else None

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

@@ -0,0 +1,100 @@
+import os
+from unittest.mock import patch
+
+from smithy_aws_core.config.sources import EnvironmentSource
+from smithy_core.interfaces.source import ConfigSource
+
+
+class TestEnvironmentSource:
+    def test_implements_config_source_protocol(self):
+        source = EnvironmentSource()
+        assert isinstance(source, ConfigSource)
+        assert hasattr(source, "name")
+        assert hasattr(source, "get")
+        assert callable(source.get)
+
+    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=False):
+            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=False):
+            source = EnvironmentSource()
+            value = source.get("region")
+            # Empty string should be treated as None
+            assert value is None
+
+    def test_get_handles_whitespace_env_var(self):
+        with patch.dict(os.environ, {"AWS_REGION": "  us-west-2  "}, clear=False):
+            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=False):
+            source = EnvironmentSource()
+            value = source.get("region")
+            # Whitespaces should be stripped
+            assert value is None
+
+    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=False):
+            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=False):
+            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=False):
+            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
+
+    def test_env_var_names_are_case_sensative(self):
+        with patch.dict(os.environ, {"aws_region": "us-west-2"}, clear=False):
+            source = EnvironmentSource()
+            value = source.get("region")
+            # Should not find 'aws_region' (lowercase), only 'AWS_REGION'
+            assert value is None

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

@@ -0,0 +1,25 @@
+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.
+        """
+        ...