Python: Replace Pydantic Settings with TypedDict + load_settings() (#3843)

* Replace Pydantic Settings with TypedDict + load_settings() - Remove pydantic-settings dependency, add python-dotenv - Delete _pydantic.py (AFBaseSettings, HTTPsUrl) - Add _settings.py with generic load_settings() function, SecretString, type coercion, and Required field validation (SettingNotFoundError) - Convert all 13 settings classes from AFBaseSettings subclasses to TypedDict definitions with load_settings() calls - Update all consumers from attribute access to dict access - Add 20 unit tests for load_settings() covering basic loading, dotenv, SecretString, type coercion, and required field validation - Update all existing tests for new settings patterns * Fix mypy type errors from settings conversion - Fix str | None attribute access in responses_client (walrus operator) - Fix SecretString | None narrowing in bedrock (type: ignore after guard) - Convert _context_provider.py attribute access to dict access (missed file) - Fix endpoint type narrowing in search_provider and context_provider - Fix purview: str | None .rstrip(), int | None defaults, urlparse bytes * Address PR review: required_fields param, type validation, fixes - Move required field validation from TypedDict annotations (Required) to a required_fields parameter on load_settings(), enabling runtime decisions about which fields are required - Remove Required imports and restore from __future__ import annotations in ollama and foundry_local - Add _check_override_type() for deterministic ServiceInitializationError on invalid override types (e.g. dict passed for str field) - Fix all multi-exception test catches back to single exception type - Fix Ollama host=None: use .get() so None is passed through to SDK default - Fix Purview processor: use explicit is-None checks instead of or operator - Remove unused BaseModel import from openai/_shared.py - Add 4 new tests (24 total): required_fields param, type validation * Fix type validation: allow int for float fields _check_override_type now permits int values for float-typed fields, matching Python's standard numeric promotion behavior. * fix: wrap urlparse arg with str() to fix mypy bytes endswith error
2026-06-16 21:04:09 +08:00 · 2026-02-12 09:51:20 +01:00
parent b488158abe
commit 8457533c69
58 changed files with 1526 additions and 1113 deletions
@@ -19,6 +19,7 @@ from agent_framework import (
    SupportsChatGetResponse,
    tool,
 )
+from agent_framework._settings import SecretString
 from agent_framework.azure import AzureOpenAIAssistantsClient
 from agent_framework.exceptions import ServiceInitializationError

@@ -556,19 +557,21 @@ def test_azure_assistants_client_entra_id_authentication() -> None:
    mock_credential = MagicMock()

    with (
-        patch("agent_framework.azure._assistants_client.AzureOpenAISettings") as mock_settings_class,
+        patch("agent_framework.azure._assistants_client.load_settings") as mock_load_settings,
+        patch("agent_framework.azure._assistants_client.get_entra_auth_token") as mock_get_token,
        patch("agent_framework.azure._assistants_client.AsyncAzureOpenAI") as mock_azure_client,
        patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
    ):
-        mock_settings = MagicMock()
-        mock_settings.chat_deployment_name = "test-deployment"
-        mock_settings.api_key = None  # No API key to trigger Entra ID path
-        mock_settings.token_endpoint = "https://login.microsoftonline.com/test"
-        mock_settings.get_azure_auth_token.return_value = "entra-token-12345"
-        mock_settings.api_version = "2024-05-01-preview"
-        mock_settings.endpoint = "https://test-endpoint.openai.azure.com"
-        mock_settings.base_url = None
-        mock_settings_class.return_value = mock_settings
+        mock_load_settings.return_value = {
+            "chat_deployment_name": "test-deployment",
+            "responses_deployment_name": None,
+            "api_key": None,
+            "token_endpoint": "https://login.microsoftonline.com/test",
+            "api_version": "2024-05-01-preview",
+            "endpoint": "https://test-endpoint.openai.azure.com",
+            "base_url": None,
+        }
+        mock_get_token.return_value = "entra-token-12345"

        client = AzureOpenAIAssistantsClient(
            deployment_name="test-deployment",
@@ -579,7 +582,7 @@ def test_azure_assistants_client_entra_id_authentication() -> None:
        )

        # Verify Entra ID token was requested
-        mock_settings.get_azure_auth_token.assert_called_once_with(mock_credential)
+        mock_get_token.assert_called_once_with(mock_credential, "https://login.microsoftonline.com/test")

        # Verify client was created with the token
        mock_azure_client.assert_called_once()
@@ -592,12 +595,16 @@ def test_azure_assistants_client_entra_id_authentication() -> None:

 def test_azure_assistants_client_no_authentication_error() -> None:
    """Test authentication validation error when no auth provided."""
-    with patch("agent_framework.azure._assistants_client.AzureOpenAISettings") as mock_settings_class:
-        mock_settings = MagicMock()
-        mock_settings.chat_deployment_name = "test-deployment"
-        mock_settings.api_key = None  # No API key
-        mock_settings.token_endpoint = None  # No token endpoint
-        mock_settings_class.return_value = mock_settings
+    with patch("agent_framework.azure._assistants_client.load_settings") as mock_load_settings:
+        mock_load_settings.return_value = {
+            "chat_deployment_name": "test-deployment",
+            "responses_deployment_name": None,
+            "api_key": None,
+            "token_endpoint": None,
+            "api_version": "2024-05-01-preview",
+            "endpoint": "https://test-endpoint.openai.azure.com",
+            "base_url": None,
+        }

        # Test missing authentication raises error
        with pytest.raises(ServiceInitializationError, match="API key, ad_token, or ad_token_provider is required"):
@@ -611,17 +618,19 @@ def test_azure_assistants_client_no_authentication_error() -> None:
 def test_azure_assistants_client_ad_token_authentication() -> None:
    """Test ad_token authentication client parameter path."""
    with (
-        patch("agent_framework.azure._assistants_client.AzureOpenAISettings") as mock_settings_class,
+        patch("agent_framework.azure._assistants_client.load_settings") as mock_load_settings,
        patch("agent_framework.azure._assistants_client.AsyncAzureOpenAI") as mock_azure_client,
        patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
    ):
-        mock_settings = MagicMock()
-        mock_settings.chat_deployment_name = "test-deployment"
-        mock_settings.api_key = None  # No API key
-        mock_settings.api_version = "2024-05-01-preview"
-        mock_settings.endpoint = "https://test-endpoint.openai.azure.com"
-        mock_settings.base_url = None
-        mock_settings_class.return_value = mock_settings
+        mock_load_settings.return_value = {
+            "chat_deployment_name": "test-deployment",
+            "responses_deployment_name": None,
+            "api_key": None,
+            "token_endpoint": None,
+            "api_version": "2024-05-01-preview",
+            "endpoint": "https://test-endpoint.openai.azure.com",
+            "base_url": None,
+        }

        client = AzureOpenAIAssistantsClient(
            deployment_name="test-deployment",
@@ -645,17 +654,19 @@ def test_azure_assistants_client_ad_token_provider_authentication() -> None:
    mock_token_provider = MagicMock(spec=AsyncAzureADTokenProvider)

    with (
-        patch("agent_framework.azure._assistants_client.AzureOpenAISettings") as mock_settings_class,
+        patch("agent_framework.azure._assistants_client.load_settings") as mock_load_settings,
        patch("agent_framework.azure._assistants_client.AsyncAzureOpenAI") as mock_azure_client,
        patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
    ):
-        mock_settings = MagicMock()
-        mock_settings.chat_deployment_name = "test-deployment"
-        mock_settings.api_key = None  # No API key
-        mock_settings.api_version = "2024-05-01-preview"
-        mock_settings.endpoint = "https://test-endpoint.openai.azure.com"
-        mock_settings.base_url = None
-        mock_settings_class.return_value = mock_settings
+        mock_load_settings.return_value = {
+            "chat_deployment_name": "test-deployment",
+            "responses_deployment_name": None,
+            "api_key": None,
+            "token_endpoint": None,
+            "api_version": "2024-05-01-preview",
+            "endpoint": "https://test-endpoint.openai.azure.com",
+            "base_url": None,
+        }

        client = AzureOpenAIAssistantsClient(
            deployment_name="test-deployment",
@@ -675,17 +686,19 @@ def test_azure_assistants_client_ad_token_provider_authentication() -> None:
 def test_azure_assistants_client_base_url_configuration() -> None:
    """Test base_url client parameter path."""
    with (
-        patch("agent_framework.azure._assistants_client.AzureOpenAISettings") as mock_settings_class,
+        patch("agent_framework.azure._assistants_client.load_settings") as mock_load_settings,
        patch("agent_framework.azure._assistants_client.AsyncAzureOpenAI") as mock_azure_client,
        patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
    ):
-        mock_settings = MagicMock()
-        mock_settings.chat_deployment_name = "test-deployment"
-        mock_settings.api_key.get_secret_value.return_value = "test-api-key"
-        mock_settings.base_url = "https://custom-base-url.com"
-        mock_settings.endpoint = None  # No endpoint, should use base_url
-        mock_settings.api_version = "2024-05-01-preview"
-        mock_settings_class.return_value = mock_settings
+        mock_load_settings.return_value = {
+            "chat_deployment_name": "test-deployment",
+            "responses_deployment_name": None,
+            "api_key": SecretString("test-api-key"),
+            "token_endpoint": None,
+            "api_version": "2024-05-01-preview",
+            "endpoint": None,
+            "base_url": "https://custom-base-url.com",
+        }

        client = AzureOpenAIAssistantsClient(
            deployment_name="test-deployment", api_key="test-api-key", base_url="https://custom-base-url.com"
@@ -704,17 +717,19 @@ def test_azure_assistants_client_base_url_configuration() -> None:
 def test_azure_assistants_client_azure_endpoint_configuration() -> None:
    """Test azure_endpoint client parameter path."""
    with (
-        patch("agent_framework.azure._assistants_client.AzureOpenAISettings") as mock_settings_class,
+        patch("agent_framework.azure._assistants_client.load_settings") as mock_load_settings,
        patch("agent_framework.azure._assistants_client.AsyncAzureOpenAI") as mock_azure_client,
        patch("agent_framework.openai.OpenAIAssistantsClient.__init__", return_value=None),
    ):
-        mock_settings = MagicMock()
-        mock_settings.chat_deployment_name = "test-deployment"
-        mock_settings.api_key.get_secret_value.return_value = "test-api-key"
-        mock_settings.base_url = None  # No base_url
-        mock_settings.endpoint = "https://test-endpoint.openai.azure.com"
-        mock_settings.api_version = "2024-05-01-preview"
-        mock_settings_class.return_value = mock_settings
+        mock_load_settings.return_value = {
+            "chat_deployment_name": "test-deployment",
+            "responses_deployment_name": None,
+            "api_key": SecretString("test-api-key"),
+            "token_endpoint": None,
+            "api_version": "2024-05-01-preview",
+            "endpoint": "https://test-endpoint.openai.azure.com",
+            "base_url": None,
+        }

        client = AzureOpenAIAssistantsClient(
            deployment_name="test-deployment",
@@ -109,8 +109,11 @@ def test_init_with_empty_endpoint_and_base_url(azure_openai_unit_test_env: dict[

@pytest.mark.parametrize("override_env_param_dict", [{"AZURE_OPENAI_ENDPOINT": "http://test.com"}], indirect=True)
 def test_init_with_invalid_endpoint(azure_openai_unit_test_env: dict[str, str]) -> None:
-    with pytest.raises(ServiceInitializationError):
-        AzureOpenAIChatClient()
+    # Note: URL scheme validation was previously handled by pydantic's HTTPsUrl type.
+    # After migrating to load_settings with TypedDict, endpoint is a plain string and no longer
+    # validated at the settings level. The Azure OpenAI SDK may reject invalid URLs at runtime.
+    client = AzureOpenAIChatClient()
+    assert client is not None


@pytest.mark.parametrize("exclude_list", [["AZURE_OPENAI_BASE_URL"]], indirect=True)
@@ -0,0 +1,238 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Tests for load_settings() function."""
+
+import os
+import tempfile
+from typing import TypedDict
+
+import pytest
+
+from agent_framework._settings import SecretString, load_settings
+
+
+class SimpleSettings(TypedDict, total=False):
+    api_key: str | None
+    timeout: int | None
+    enabled: bool | None
+    rate_limit: float | None
+
+
+class RequiredFieldSettings(TypedDict, total=False):
+    name: str | None
+    optional_field: str | None
+
+
+class SecretSettings(TypedDict, total=False):
+    api_key: SecretString | None
+    username: str | None
+
+
+class TestLoadSettingsBasic:
+    """Test basic load_settings functionality."""
+
+    def test_fields_are_none_when_unset(self) -> None:
+        settings = load_settings(SimpleSettings, env_prefix="TEST_APP_")
+
+        assert settings["api_key"] is None
+        assert settings["timeout"] is None
+        assert settings["enabled"] is None
+        assert settings["rate_limit"] is None
+
+    def test_overrides(self) -> None:
+        settings = load_settings(SimpleSettings, env_prefix="TEST_APP_", timeout=60, enabled=False)
+
+        assert settings["timeout"] == 60
+        assert settings["enabled"] is False
+
+    def test_none_overrides_are_filtered(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TEST_APP_TIMEOUT", "120")
+
+        settings = load_settings(SimpleSettings, env_prefix="TEST_APP_", timeout=None)
+
+        # timeout=None is filtered, so env var wins
+        assert settings["timeout"] == 120
+
+    def test_env_vars(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TEST_APP_API_KEY", "test-key-123")
+        monkeypatch.setenv("TEST_APP_TIMEOUT", "120")
+        monkeypatch.setenv("TEST_APP_ENABLED", "false")
+
+        settings = load_settings(SimpleSettings, env_prefix="TEST_APP_")
+
+        assert settings["api_key"] == "test-key-123"
+        assert settings["timeout"] == 120
+        assert settings["enabled"] is False
+
+    def test_overrides_beat_env_vars(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TEST_APP_TIMEOUT", "120")
+
+        settings = load_settings(SimpleSettings, env_prefix="TEST_APP_", timeout=60)
+
+        assert settings["timeout"] == 60
+
+    def test_no_prefix(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("API_KEY", "no-prefix-key")
+
+        settings = load_settings(SimpleSettings, api_key=None)
+
+        assert settings["api_key"] == "no-prefix-key"
+
+
+class TestDotenvFile:
+    """Test .env file loading."""
+
+    def test_load_from_dotenv(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("TEST_APP_API_KEY", raising=False)
+        monkeypatch.delenv("TEST_APP_TIMEOUT", raising=False)
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".env", delete=False) as f:
+            f.write("TEST_APP_API_KEY=dotenv-key\n")
+            f.write("TEST_APP_TIMEOUT=90\n")
+            f.flush()
+            env_path = f.name
+
+        try:
+            settings = load_settings(SimpleSettings, env_prefix="TEST_APP_", env_file_path=env_path)
+
+            assert settings["api_key"] == "dotenv-key"
+            assert settings["timeout"] == 90
+        finally:
+            os.unlink(env_path)
+
+    def test_env_vars_override_dotenv(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TEST_APP_API_KEY", "real-env-key")
+
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".env", delete=False) as f:
+            f.write("TEST_APP_API_KEY=dotenv-key\n")
+            f.flush()
+            env_path = f.name
+
+        try:
+            settings = load_settings(SimpleSettings, env_prefix="TEST_APP_", env_file_path=env_path)
+
+            assert settings["api_key"] == "real-env-key"
+        finally:
+            os.unlink(env_path)
+
+    def test_missing_dotenv_file(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("TEST_APP_API_KEY", raising=False)
+        settings = load_settings(SimpleSettings, env_prefix="TEST_APP_", env_file_path="/nonexistent/.env")
+
+        assert settings["api_key"] is None
+
+
+class TestSecretString:
+    """Test SecretString type handling."""
+
+    def test_secretstring_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("SECRET_API_KEY", "secret-value")
+
+        settings = load_settings(SecretSettings, env_prefix="SECRET_")
+
+        assert isinstance(settings["api_key"], SecretString)
+        assert settings["api_key"] == "secret-value"
+
+    def test_secretstring_from_override(self) -> None:
+        settings = load_settings(SecretSettings, env_prefix="SECRET_", api_key="kwarg-secret")
+
+        assert isinstance(settings["api_key"], SecretString)
+        assert settings["api_key"] == "kwarg-secret"
+
+    def test_secretstring_masked_in_repr(self) -> None:
+        s = SecretString("my-secret")
+        assert "my-secret" not in repr(s)
+        assert "**********" in repr(s)
+
+    def test_get_secret_value_compat(self) -> None:
+        s = SecretString("my-secret")
+
+        assert s.get_secret_value() == "my-secret"
+        assert isinstance(s.get_secret_value(), str)
+
+
+class TestTypeCoercion:
+    """Test type coercion from string values."""
+
+    def test_int_coercion(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TEST_APP_TIMEOUT", "42")
+
+        settings = load_settings(SimpleSettings, env_prefix="TEST_APP_")
+
+        assert settings["timeout"] == 42
+        assert isinstance(settings["timeout"], int)
+
+    def test_float_coercion(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TEST_APP_RATE_LIMIT", "2.5")
+
+        settings = load_settings(SimpleSettings, env_prefix="TEST_APP_")
+
+        assert settings["rate_limit"] == 2.5
+        assert isinstance(settings["rate_limit"], float)
+
+    def test_bool_coercion_true_values(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        for true_val in ["true", "True", "TRUE", "1", "yes", "on"]:
+            monkeypatch.setenv("TEST_APP_ENABLED", true_val)
+            settings = load_settings(SimpleSettings, env_prefix="TEST_APP_")
+            assert settings["enabled"] is True, f"Failed for {true_val}"
+
+    def test_bool_coercion_false_values(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        for false_val in ["false", "False", "FALSE", "0", "no", "off"]:
+            monkeypatch.setenv("TEST_APP_ENABLED", false_val)
+            settings = load_settings(SimpleSettings, env_prefix="TEST_APP_")
+            assert settings["enabled"] is False, f"Failed for {false_val}"
+
+
+class TestRequiredFields:
+    """Test required field validation."""
+
+    def test_required_field_provided(self) -> None:
+        settings = load_settings(
+            RequiredFieldSettings,
+            env_prefix="TEST_",
+            required_fields=["name"],
+            name="my-app",
+        )
+
+        assert settings["name"] == "my-app"
+        assert settings["optional_field"] is None
+
+    def test_required_field_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("TEST_NAME", "env-app")
+
+        settings = load_settings(RequiredFieldSettings, env_prefix="TEST_", required_fields=["name"])
+
+        assert settings["name"] == "env-app"
+
+    def test_required_field_missing_raises(self) -> None:
+        from agent_framework.exceptions import SettingNotFoundError
+
+        with pytest.raises(SettingNotFoundError, match="Required setting 'name'"):
+            load_settings(RequiredFieldSettings, env_prefix="TEST_", required_fields=["name"])
+
+    def test_without_required_fields_param_allows_none(self) -> None:
+        settings = load_settings(RequiredFieldSettings, env_prefix="TEST_")
+
+        assert settings["name"] is None
+
+
+class TestOverrideTypeValidation:
+    """Test override type validation."""
+
+    def test_invalid_type_raises(self) -> None:
+        from agent_framework.exceptions import ServiceInitializationError
+
+        with pytest.raises(ServiceInitializationError, match="Invalid type for setting 'api_key'"):
+            load_settings(SimpleSettings, env_prefix="TEST_", api_key={"bad": "type"})
+
+    def test_valid_types_accepted(self) -> None:
+        settings = load_settings(SimpleSettings, env_prefix="TEST_", timeout=42, enabled=True)
+
+        assert settings["timeout"] == 42
+        assert settings["enabled"] is True
+
+    def test_str_accepted_for_secretstring(self) -> None:
+        settings = load_settings(SecretSettings, env_prefix="TEST_", api_key="plain-string")
+
+        assert isinstance(settings["api_key"], SecretString)
+        assert settings["api_key"] == "plain-string"
@@ -131,9 +131,15 @@ class TestOpenAIAssistantProviderInit:
        """Test initialization fails without API key when settings return None."""
        from unittest.mock import patch

-        # Mock OpenAISettings to return None for api_key
-        with patch("agent_framework.openai._assistant_provider.OpenAISettings") as mock_settings:
-            mock_settings.return_value.api_key = None
+        # Mock load_settings to return a dict with None for api_key
+        with patch("agent_framework.openai._assistant_provider.load_settings") as mock_load:
+            mock_load.return_value = {
+                "api_key": None,
+                "org_id": None,
+                "base_url": None,
+                "chat_model_id": None,
+                "responses_model_id": None,
+            }

            with pytest.raises(ServiceInitializationError) as exc_info:
                OpenAIAssistantProvider()
@@ -146,7 +146,7 @@ def test_init_auto_create_client(
 def test_init_validation_fail() -> None:
    """Test OpenAIAssistantsClient initialization with validation failure."""
    with pytest.raises(ServiceInitializationError):
-        # Force failure by providing invalid model ID type - this should cause validation to fail
+        # Force failure by providing invalid model ID type
        OpenAIAssistantsClient(model_id=123, api_key="valid-key")  # type: ignore