fix(api): resolve aliases in remote settings sources (#41463)

api/configs/remote_settings_sources/apollo/__init__.py

@@ -52,5 +52,4 @@ class ApolloSettingsSource(RemoteSettingsSource):
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
         if not isinstance(self.remote_configs, dict):
             raise ValueError(f"remote configs is not dict, but {type(self.remote_configs)}")
-        field_value = self.remote_configs.get(field_name)
-        return field_value, field_name, False
+        return self.resolve_field_value(self.remote_configs, field, field_name)

api/configs/remote_settings_sources/base.py

@@ -1,6 +1,7 @@
 from collections.abc import Mapping
 from typing import Any
 
+from pydantic import AliasChoices
 from pydantic.fields import FieldInfo
 
 
@@ -11,5 +12,28 @@ class RemoteSettingsSource:
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
         raise NotImplementedError
 
+    @staticmethod
+    def resolve_field_value(
+        remote_configs: Mapping[str, Any], field: FieldInfo, field_name: str
+    ) -> tuple[Any, str, bool]:
+        field_value = remote_configs.get(field_name)
+        if field_value is not None:
+            return field_value, field_name, False
+
+        validation_alias = field.validation_alias
+        if isinstance(validation_alias, str):
+            aliases = [validation_alias]
+        elif isinstance(validation_alias, AliasChoices):
+            aliases = [alias for alias in validation_alias.choices if isinstance(alias, str)]
+        else:
+            aliases = []
+
+        for alias in aliases:
+            field_value = remote_configs.get(alias)
+            if field_value is not None:
+                return field_value, alias, False
+
+        return None, field_name, False
+
     def prepare_field_value(self, field_name: str, field: FieldInfo, value: Any, value_is_complex: bool):
         return value

api/configs/remote_settings_sources/nacos/__init__.py

@@ -43,8 +43,4 @@ class NacosSettingsSource(RemoteSettingsSource):
 
     @override
     def get_field_value(self, field: FieldInfo, field_name: str) -> tuple[Any, str, bool]:
-        field_value = self.remote_configs.get(field_name)
-        if field_value is None:
-            return None, field_name, False
-
-        return field_value, field_name, False
+        return self.resolve_field_value(self.remote_configs, field, field_name)

api/tests/unit_tests/configs/test_remote_settings_sources.py

@@ -0,0 +1,39 @@
+import pytest
+from pydantic import AliasChoices
+from pydantic.fields import FieldInfo
+
+from configs.remote_settings_sources.apollo import ApolloSettingsSource
+from configs.remote_settings_sources.nacos import NacosSettingsSource
+
+
+@pytest.mark.parametrize("source_cls", [NacosSettingsSource, ApolloSettingsSource])
+def test_get_field_value_prefers_field_name(
+    source_cls: type[NacosSettingsSource] | type[ApolloSettingsSource],
+) -> None:
+    source = object.__new__(source_cls)
+    source.remote_configs = {"SETTING": "direct", "ALIAS": "aliased"}
+    field = FieldInfo(validation_alias="ALIAS")
+
+    assert source.get_field_value(field, "SETTING") == ("direct", "SETTING", False)
+
+
+@pytest.mark.parametrize("source_cls", [NacosSettingsSource, ApolloSettingsSource])
+def test_get_field_value_falls_back_to_single_validation_alias(
+    source_cls: type[NacosSettingsSource] | type[ApolloSettingsSource],
+) -> None:
+    source = object.__new__(source_cls)
+    source.remote_configs = {"ALIAS": "aliased"}
+    field = FieldInfo(validation_alias="ALIAS")
+
+    assert source.get_field_value(field, "SETTING") == ("aliased", "ALIAS", False)
+
+
+@pytest.mark.parametrize("source_cls", [NacosSettingsSource, ApolloSettingsSource])
+def test_get_field_value_falls_back_to_alias_choices_in_order(
+    source_cls: type[NacosSettingsSource] | type[ApolloSettingsSource],
+) -> None:
+    source = object.__new__(source_cls)
+    source.remote_configs = {"SECOND_ALIAS": "second"}
+    field = FieldInfo(validation_alias=AliasChoices("FIRST_ALIAS", "SECOND_ALIAS"))
+
+    assert source.get_field_value(field, "SETTING") == ("second", "SECOND_ALIAS", False)