refactor(config): add shared _resolve_component utility for plugin loading

.changelog/5215.added

@@ -0,0 +1 @@
+`opentelemetry-sdk`: add `_resolve_component` shared utility for declarative config plugin loading, reducing boilerplate in exporter factory functions

opentelemetry-sdk/src/opentelemetry/sdk/_configuration/_common.py

@@ -6,6 +6,8 @@
 import dataclasses
 import inspect
 import logging
+from collections.abc import Callable
+from typing import Any, Protocol
 
 from opentelemetry.sdk._configuration._exceptions import ConfigurationError
 from opentelemetry.util._importlib_metadata import entry_points
@@ -72,6 +74,66 @@ def load_entry_point(group: str, name: str) -> type:
         ) from exc
 
 
+class _ComponentConfig(Protocol):
+    """Protocol for config dataclasses decorated with @_additional_properties.
+
+    Values in ``additional_properties`` are nested config dicts (suitable
+    for ``**kwargs`` splatting to the user-defined component class) or
+    ``None`` (when the YAML uses ``my_plugin:`` or ``my_plugin: null``).
+
+    Note: the generated models declare ``additional_properties`` as a
+    ``ClassVar`` even though the decorator assigns it as an instance
+    attribute at runtime. This is tolerated by pyright in ``standard``
+    mode but flagged in ``strict`` mode. See #5268.
+    """
+
+    additional_properties: dict[str, dict[str, Any] | None]
+
+
+def _resolve_component(
+    config: _ComponentConfig,
+    registry: dict[str, Callable[[Any], Any]],
+    entry_point_group: str,
+    component_type: str,
+) -> Any:
+    """Resolve a config dataclass to a component instance.
+
+    Checks built-in factories in ``registry`` first (by matching typed
+    field names on ``config``), then falls back to entry point loading
+    for plugin components found in ``config.additional_properties``.
+
+    The JSON schema enforces exactly one component per config block
+    (``minProperties: 1, maxProperties: 1``). If multiple typed fields or
+    ``additional_properties`` entries are set (e.g. when schema validation
+    is bypassed), the first registry match wins.
+
+    Args:
+        config: A dataclass with ``additional_properties`` (from the
+            ``@_additional_properties`` decorator).
+        registry: Mapping of built-in component names to factory
+            callables. Each factory receives the field value from config.
+        entry_point_group: The entry point group name for plugin loading.
+        component_type: Human-readable name for error messages
+            (e.g. "span exporter").
+
+    Returns:
+        The resolved component instance.
+
+    Raises:
+        ConfigurationError: If no component type is specified in config.
+    """
+    for name, factory in registry.items():
+        value = getattr(config, name, None)
+        if value is not None:
+            return factory(value)
+    if config.additional_properties:
+        name, plugin_config = next(iter(config.additional_properties.items()))
+        return load_entry_point(entry_point_group, name)(
+            **(plugin_config or {})
+        )
+    raise ConfigurationError(f"No {component_type} type specified in config.")
+
+
 def _parse_headers(
     headers: list | None,
     headers_list: str | None,

opentelemetry-sdk/tests/_configuration/test_common.py

@@ -12,6 +12,7 @@
     _additional_properties,
     _map_compression,
     _parse_headers,
+    _resolve_component,
     load_entry_point,
 )
 from opentelemetry.sdk._configuration._exceptions import ConfigurationError
@@ -289,3 +290,84 @@ def test_log_record_exporter(self):
 
     def test_push_metric_exporter(self):
         self._assert_supports_additional_properties(PushMetricExporter)
+
+
+class TestResolveComponent(unittest.TestCase):
+    def setUp(self):
+        @_additional_properties
+        @dataclass
+        class _Config:
+            builtin_a: dict | None = None
+            builtin_b: str | None = None
+            additional_properties: ClassVar[dict[str, Any]]
+
+        self.cls = _Config
+        self.registry = {
+            "builtin_a": lambda v: ("resolved_a", v),
+            "builtin_b": lambda v: ("resolved_b", v),
+        }
+
+    def test_resolves_builtin_from_registry(self):
+        config = self.cls(builtin_a={"key": "val"})
+        result = _resolve_component(
+            config, self.registry, "test_group", "test component"
+        )
+        self.assertEqual(result, ("resolved_a", {"key": "val"}))
+
+    def test_resolves_plugin_via_entry_point(self):
+        mock_instance = MagicMock()
+        mock_class = MagicMock(return_value=mock_instance)
+        with patch(
+            "opentelemetry.sdk._configuration._common.entry_points",
+            return_value=[MagicMock(**{"load.return_value": mock_class})],
+        ):
+            # pylint: disable=unexpected-keyword-arg
+            config = self.cls(my_plugin={"opt": "val"})
+            result = _resolve_component(
+                config, self.registry, "test_group", "test component"
+            )
+        self.assertIs(result, mock_instance)
+        mock_class.assert_called_once_with(opt="val")
+
+    def test_plugin_with_empty_config(self):
+        mock_instance = MagicMock()
+        mock_class = MagicMock(return_value=mock_instance)
+        with patch(
+            "opentelemetry.sdk._configuration._common.entry_points",
+            return_value=[MagicMock(**{"load.return_value": mock_class})],
+        ):
+            # pylint: disable=unexpected-keyword-arg
+            config = self.cls(my_plugin={})
+            _resolve_component(
+                config, self.registry, "test_group", "test component"
+            )
+        mock_class.assert_called_once_with()
+
+    def test_no_component_raises_configuration_error(self):
+        config = self.cls()
+        with self.assertRaises(ConfigurationError):
+            _resolve_component(
+                config, self.registry, "test_group", "test component"
+            )
+
+    def test_plugin_not_found_raises_configuration_error(self):
+        with patch(
+            "opentelemetry.sdk._configuration._common.entry_points",
+            return_value=[],
+        ):
+            # pylint: disable=unexpected-keyword-arg
+            config = self.cls(missing_plugin={})
+            with self.assertRaises(ConfigurationError):
+                _resolve_component(
+                    config, self.registry, "test_group", "test component"
+                )
+
+    def test_first_registry_match_wins_when_multiple_set(self):
+        """When multiple built-in fields are set (which the schema should
+        prevent), the first registry match wins."""
+        config = self.cls(builtin_a={"a": 1}, builtin_b="b")
+        result = _resolve_component(
+            config, self.registry, "test_group", "test component"
+        )
+        # builtin_a comes first in the registry dict
+        self.assertEqual(result, ("resolved_a", {"a": 1}))