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

@@ -72,6 +72,45 @@ def load_entry_point(group: str, name: str) -> type:
         ) from exc
 
 
+def _resolve_component(
+    config,
+    registry: dict,
+    entry_point_group: str,
+    component_type: str,
+):
+    """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``.
+
+    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,74 @@ 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"
+                )