Add support for config resolver

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

@@ -0,0 +1,176 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from typing import Any, ClassVar, Self
+
+from smithy_aws_core.config.context import FileSystem, SharedConfigContext
+from smithy_aws_core.config.exceptions import ConfigError
+from smithy_aws_core.config.resolvers import resolve_region, resolve_retry_config
+from smithy_aws_core.config.types import UNSET, ConfigSource, FieldSpec, Resolved
+from smithy_aws_core.config.validators import (
+    validate_region,
+    validate_retry_strategy_options,
+)
+from smithy_core.retries import RetryStrategyOptions
+
+
+@dataclass(kw_only=True)
+class AsyncAwsConfig:
+    """Base configuration class for all AWS services.
+
+    Fields are resolved asynchronously from multiple sources (env vars,
+    config files, defaults) through the resolve() classmethod.
+
+    Do not instantiate directly — use:
+        config = await AsyncAwsConfig.resolve()
+    """
+
+    region: str | None = None
+    retry_strategy_options: RetryStrategyOptions | None = None
+
+    _ctx: SharedConfigContext | None = field(default=None, repr=False, compare=False)
+    _sources: dict[str, ConfigSource] = field(  # type: ignore[assignment]
+        default_factory=dict,
+        repr=False,
+        compare=False,
+    )
+
+    _FIELDS: ClassVar[dict[str, FieldSpec]] = {
+        "region": FieldSpec(
+            default=None,
+            resolver=resolve_region,
+            validator=validate_region,
+        ),
+        "retry_strategy_options": FieldSpec(
+            default_factory=RetryStrategyOptions,
+            resolver=resolve_retry_config,
+            validator=validate_retry_strategy_options,
+        ),
+    }
+
+    def __post_init__(self) -> None:
+        """Block direct construction. Use resolve() instead."""
+        raise ConfigError(
+            f"{type(self).__name__} cannot be constructed directly. "
+            f"Use `await {type(self).__name__}.resolve(...)` instead."
+        )
+
+    @classmethod
+    async def resolve(
+        cls,
+        *,
+        profile: str | None = None,
+        env: Mapping[str, str] | None = None,
+        fs: FileSystem | None = None,
+        config_file_path: str | None = None,
+        credentials_file_path: str | None = None,
+        **overrides: Any,
+    ) -> Self:
+        """Resolve a config object from environment, config files, and defaults.
+
+        This is the only supported way to create a config instance.
+
+        :param profile: Override the active profile name.
+        :param env: Override the environment variable mapping.
+        :param fs: Override the filesystem abstraction.
+        :param config_file_path: Override path for config file.
+        :param credentials_file_path: Override path for credentials file.
+        :param overrides: Explicit field values that skip resolution.
+        :returns: A fully-resolved config instance.
+        """
+        ctx = SharedConfigContext(
+            profile_name=profile,
+            env=env,
+            fs=fs,
+            config_file_path=config_file_path,
+            credentials_file_path=credentials_file_path,
+        )
+
+        # Create the instance bypassing __post_init__ check
+        instance = cls._create_instance()
+        instance._ctx = ctx
+
+        # Resolve each field
+        await instance._resolve_fields(overrides)
+
+        return instance
+
+    def source_of(self, field_name: str) -> ConfigSource | None:
+        """Get the source that provided a field's value.
+
+        :param field_name: The config field name.
+        :returns: The ConfigSource, or None if not tracked.
+        """
+        return self._sources.get(field_name)
+
+    def resolution_context(self) -> SharedConfigContext | None:
+        """Get the resolution context used to create this config.
+
+        :returns: The SharedConfigContext, or None if not available.
+        """
+        return self._ctx
+
+    @classmethod
+    def _create_instance(cls) -> Self:
+        """Create an instance that bypasses construction blocking."""
+        instance = object.__new__(cls)
+        object.__setattr__(instance, "_sources", {})
+        object.__setattr__(instance, "_ctx", None)
+        for field_name in cls._FIELDS:
+            object.__setattr__(instance, field_name, UNSET)
+        return instance
+
+    async def _resolve_fields(self, overrides: dict[str, Any]) -> None:
+        """Run the resolution pipeline for all fields."""
+        unknown = set(overrides) - set(self._FIELDS)
+        if unknown:
+            raise ConfigError(
+                f"Unknown config field(s): {sorted(unknown)}. "
+                f"Valid fields are: {sorted(self._FIELDS)}"
+            )
+
+        for field_name, spec in self._FIELDS.items():
+            # check for overrides first
+            if field_name in overrides:
+                value = overrides[field_name]
+                setattr(self, field_name, value)
+                self._sources[field_name] = ConfigSource.OVERRIDE
+            # check in resolver
+            elif spec.resolver is not None:
+                result: Resolved[Any] = await spec.resolver(self._ctx)
+                if result.value is not UNSET:
+                    setattr(self, field_name, result.value)
+                    self._sources[field_name] = result.source
+                else:
+                    # If resolver returned UNSET, fall back to default
+                    self._apply_default(field_name, spec)
+
+            else:
+                # No resolver, use default directly
+                self._apply_default(field_name, spec)
+
+            # Run validator for all sources
+            if spec.validator is not None:
+                spec.validator(getattr(self, field_name))
+
+    def _apply_default(self, field_name: str, spec: FieldSpec) -> None:
+        """Apply the default value for a field."""
+        if spec.default_factory is not None:
+            value = spec.default_factory()
+        else:
+            value = spec.default
+        setattr(self, field_name, value)
+        self._sources[field_name] = ConfigSource.DEFAULT
+
+    def __setattr__(self, name: str, value: Any) -> None:
+        """Track provenance when fields are set with plugins after construction"""
+        super().__setattr__(name, value)
+        # Mark as override only if the field is in _FIELDS and was already resolved
+        if (
+            name in self.__class__._FIELDS
+            and hasattr(self, "_sources")
+            and name in self._sources
+        ):
+            self._sources[name] = ConfigSource.OVERRIDE

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

@@ -0,0 +1,139 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+import logging
+import os
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any, Protocol, runtime_checkable
+
+from smithy_aws_core.config import load_config
+from smithy_aws_core.config.merged_config import MergedConfig
+
+logger = logging.getLogger(__name__)
+
+
+@runtime_checkable
+class FileSystem(Protocol):
+    """Protocol for filesystem operations.
+
+    Abstraction over file I/O so tests can provide mock implementations
+    without touching the real filesystem.
+    """
+
+    async def read_file(self, path: str) -> str | None:
+        """Read a file's content as UTF-8.
+
+        :param path: Resolved file path.
+        :returns: File content, or None if the file is inaccessible.
+        """
+        ...
+
+
+class DefaultFileSystem:
+    """Default filesystem implementation using real disk I/O."""
+
+    async def read_file(self, path: str) -> str | None:
+        """Read a file asynchronously from disk.
+
+        Missing files and permission errors return None with a warning.
+        Encoding errors (invalid UTF-8) are raised to the caller.
+
+        :param path: Resolved file path.
+        :returns: File content, or None if the file is inaccessible.
+        """
+        try:
+            content: str = await asyncio.to_thread(
+                Path(path).read_text, encoding="utf-8"
+            )
+            return content
+        except FileNotFoundError:
+            return None
+        except (PermissionError, OSError) as e:
+            logger.warning("Unable to read config file '%s': %s", path, e)
+            return None
+
+
+class SharedConfigContext:
+    """Resolution context shared across resolvers during config construction.
+
+    Holds environment state and cached file data that resolvers use to
+    look up values. Created once per resolve() call.
+    """
+
+    def __init__(
+        self,
+        *,
+        profile_name: str | None = None,
+        env: Mapping[str, str] | None = None,
+        fs: FileSystem | None = None,
+        http_client: Any | None = None,
+        config_file_path: str | Path | None = None,
+        credentials_file_path: str | Path | None = None,
+    ) -> None:
+        """Initialize the resolution context.
+
+        :param profile_name: The active profile to use. Defaults to
+            AWS_PROFILE env var, then "default".
+        :param env: Environment variable mapping. Defaults to os.environ.
+        :param fs: Filesystem abstraction for reading config files.
+            Defaults to real disk I/O.
+        :param http_client: HTTP client for network-based resolvers
+            (e.g., IMDS).
+        :param config_file_path: Override path for config file.
+        :param credentials_file_path: Override path for credentials file.
+        """
+        self._env: Mapping[str, str] = env if env is not None else os.environ
+        self._fs: FileSystem = fs if fs is not None else DefaultFileSystem()
+        self._http_client: Any | None = http_client
+        self._profile_name: str = self._resolve_profile_name(profile_name)
+        self._config_file_path: Path | None = (
+            Path(config_file_path) if config_file_path is not None else None
+        )
+        self._credentials_file_path: Path | None = (
+            Path(credentials_file_path) if credentials_file_path is not None else None
+        )
+        self._cached_config_file: MergedConfig | None = None
+
+    @property
+    def env(self) -> Mapping[str, str]:
+        """The environment variable mapping."""
+        return self._env
+
+    @property
+    def profile_name(self) -> str:
+        """The active profile name."""
+        return self._profile_name
+
+    @property
+    def fs(self) -> FileSystem:
+        """The filesystem abstraction."""
+        return self._fs
+
+    @property
+    def http_client(self) -> Any | None:
+        """HTTP client for network-based resolvers."""
+        return self._http_client
+
+    async def parsed_profiles(self) -> MergedConfig:
+        """Get the parsed and merged config/credentials file data.
+
+        The result is cached after the first call so files are only
+        read from disk once per context.
+        """
+        if self._cached_config_file is None:
+            self._cached_config_file = await load_config(
+                config_file_path=self._config_file_path,
+                credentials_file_path=self._credentials_file_path,
+            )
+        return self._cached_config_file
+
+    def _resolve_profile_name(self, explicit_profile: str | None) -> str:
+        """Determine the active profile name.
+
+        Priority: explicit argument > AWS_PROFILE env var > "default"
+        """
+        if explicit_profile is not None:
+            return explicit_profile
+        return self._env.get("AWS_PROFILE", "default")