Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions packages/smithy-core/src/smithy_core/config/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0

from .resolver import ConfigResolver

__all__ = [
"ConfigResolver",
]
37 changes: 37 additions & 0 deletions packages/smithy-core/src/smithy_core/config/resolver.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
from collections.abc import Sequence
from typing import Any

from smithy_core.interfaces.config import ConfigSource


class ConfigResolver:
"""Resolves configuration values from multiple sources.

The resolver iterates through sources in precedence order, returning
the first non-None value found for a given configuration key.
"""

def __init__(self, sources: Sequence[ConfigSource]) -> None:
"""Initialize the resolver with sources in precedence order.

:param sources: List of configuration sources in precedence order. The first
source in the list has the highest priority. The list is copied to
prevent external modification.
"""
self._sources = list(sources)

def get(self, key: str) -> tuple[Any, Any]:
"""Resolve a configuration value from sources by iterating through them in precedence order.

:param key: The configuration key to resolve (e.g., 'retry_mode')

:returns: A tuple of (value, source_name). If no source provides a value,
returns (None, None).
"""
for source in self._sources:
value = source.get(key)
if value is not None:
return (value, source.name)
return (None, None)
2 changes: 2 additions & 0 deletions packages/smithy-core/tests/unit/config/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
127 changes: 127 additions & 0 deletions packages/smithy-core/tests/unit/config/test_resolver.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0
from typing import Any

from smithy_core.config.resolver import ConfigResolver


class StubSource:
"""A simple ConfigSource implementation for testing.

Returns values from a provided dictionary, or None if the key
is not present.
"""

def __init__(self, source_name: str, data: dict[str, Any] | 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 TestConfigResolver:
def test_returns_value_from_single_source(self):
source = StubSource("environment", {"region": "us-west-2"})
resolver = ConfigResolver(sources=[source])

result = resolver.get("region")

assert result == ("us-west-2", "environment")

def test_returns_None_when_source_has_no_value(self):
source = StubSource("environment", {})
resolver = ConfigResolver(sources=[source])

result = resolver.get("region")

assert result == (None, None)

def test_returns_None_with_empty_source_list(self):
resolver = ConfigResolver(sources=[])

result = resolver.get("region")

assert result == (None, None)

def test_first_source_takes_precedence(self):
first_priority_source = StubSource("source_one", {"region": "us-east-1"})
second_priority_source = StubSource("source_two", {"region": "eu-west-1"})
resolver = ConfigResolver(
sources=[first_priority_source, second_priority_source]
)

result = resolver.get("region")

assert result == ("us-east-1", "source_one")

def test_skips_source_returning_none_and_uses_next(self):
empty_source = StubSource("source_one", {})
fallback_source = StubSource("source_two", {"region": "ap-south-1"})
resolver = ConfigResolver(sources=[empty_source, fallback_source])

result = resolver.get("region")

assert result == ("ap-south-1", "source_two")

def test_resolves_different_keys_from_different_sources(self):
instance = StubSource("source_one", {"region": "us-west-2"})
environment = StubSource("source_two", {"retry_mode": "adaptive"})
resolver = ConfigResolver(sources=[instance, environment])

region = resolver.get("region")
retry_mode = resolver.get("retry_mode")

assert region == ("us-west-2", "source_one")
assert retry_mode == ("adaptive", "source_two")

def test_returns_non_string_values(self):
source = StubSource(
"default",
{
"max_retries": 3,
"use_ssl": True,
},
)
resolver = ConfigResolver(sources=[source])

assert resolver.get("max_retries") == (3, "default")
assert resolver.get("use_ssl") == (True, "default")

def test_get_is_idempotent(self):
source = StubSource("environment", {"region": "us-west-2"})
resolver = ConfigResolver(sources=[source])

result1 = resolver.get("region")
result2 = resolver.get("region")
result3 = resolver.get("region")

assert result1 == result2 == result3 == ("us-west-2", "environment")

def test_treats_empty_string_as_valid_value(self):
source = StubSource("test", {"region": ""})
resolver = ConfigResolver(sources=[source])

value, source_name = resolver.get("region")

assert value == ""
assert source_name == "test"

def test_external_list_modifications_do_not_affect_resolver(self):
source1 = StubSource("environment", {"region": "us-west-2"})
source2 = StubSource("config", {"region": "eu-west-1"})
sources = [source1]

resolver = ConfigResolver(sources=sources)

# Modify the original list after resolver construction
sources.append(source2)
sources.clear()

# Resolver should use the original source
result = resolver.get("region")
assert result == ("us-west-2", "environment")
Loading