opentelemetry-codegen-json: support decoding enums from names

.changelog/5402.fixed

@@ -0,0 +1 @@
+`opentelemetry-codegen-json`: support decoding enums from names

codegen/opentelemetry-codegen-json/src/opentelemetry/codegen/json/generator.py

@@ -325,7 +325,6 @@ def _generate_imports(
         std_imports = [
             "builtins",
             "dataclasses",
-            "functools",
             "typing",
         ]
         if include_enum:
@@ -336,12 +335,6 @@ def _generate_imports(
 
         writer.blank_line()
 
-        writer.assignment(
-            "_dataclass",
-            "functools.partial(dataclasses.dataclass, slots=True)",
-        )
-        writer.blank_line()
-
         # Collect all imports needed
         imports = self._collect_imports(proto_file)
         imports.add(f"import {self._get_codec_module_path()}")
@@ -428,7 +421,7 @@ def _generate_message_class(
             msg_desc.name,
             bases=(f"{codec}.JsonMessage",),
             decorators=("typing.final",),
-            decorator_name="_dataclass",
+            slots=True,
         ):
             if msg_desc.field or msg_desc.nested_type or msg_desc.enum_type:
                 writer.docstring(
@@ -770,12 +763,9 @@ def _generate_deserialization_statements(
             enum_type = self._resolve_enum_type(
                 field_desc.type_name, proto_file
             )
-            writer.writeln(
-                f'{codec}.validate_type({var_name}, builtins.int, "{field_desc.name}")'
-            )
             writer.assignment(
                 f'{target_dict}["{field_desc.name}"]',
-                f"{enum_type}({var_name})",
+                f'{codec}.decode_enum({var_name}, {enum_type}, "{field_desc.name}")',
             )
         elif is_hex_encoded_field(field_desc.name):
             writer.assignment(
@@ -837,7 +827,7 @@ def _get_deserialization_expr(
             enum_type = self._resolve_enum_type(
                 field_desc.type_name, proto_file
             )
-            return f"{enum_type}({var_name})"
+            return f'{codec}.decode_enum({var_name}, {enum_type}, "{field_desc.name}")'
         if is_hex_encoded_field(field_desc.name):
             return f'{codec}.decode_hex({var_name}, "{field_desc.name}")'
         if is_int64_type(field_desc.type):

codegen/opentelemetry-codegen-json/src/opentelemetry/codegen/json/runtime/json_codec.py

@@ -6,12 +6,14 @@
 import abc
 import base64
 import collections.abc
+import enum
 import json
 import math
 import typing
 
 T = typing.TypeVar("T")
 M = typing.TypeVar("M", bound="JsonMessage")
+EnumT = typing.TypeVar("EnumT", bound=enum.IntEnum)
 
 
 class JsonMessage(abc.ABC):
@@ -246,3 +248,39 @@ def validate_type(
             f"Field '{field_name}' expected {expected_types}, "
             f"got {type(value).__name__}"
         )
+
+
+def decode_enum(
+    value: typing.Any,
+    enum_type: type[EnumT],
+    field_name: str,
+) -> EnumT:
+    """
+    Decode a JSON enum value into an enum member.
+
+    Per the ProtoJSON spec, parsers must accept both enum names (str)
+    and integer values (int).
+
+    Args:
+        value: The enum name or integer value to decode.
+        enum_type: The enum class to decode into.
+        field_name: The name of the field being decoded (for error messages).
+    Returns:
+        The corresponding enum member.
+    """
+    if isinstance(value, bool):
+        raise TypeError(f"Field '{field_name}' expected int or str, got bool")
+    validate_type(value, (int, str), field_name)
+    if isinstance(value, str):
+        try:
+            return enum_type[value]
+        except KeyError:
+            raise KeyError(
+                f"Invalid enum name '{value}' for field '{field_name}'"
+            ) from None
+    try:
+        return enum_type(value)
+    except ValueError:
+        raise ValueError(
+            f"Invalid enum value {value} for field '{field_name}'"
+        ) from None

codegen/opentelemetry-codegen-json/tests/test_json_codec.py

@@ -1,12 +1,15 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+import enum
 import math
+from typing import Any
 
 import pytest  # type: ignore
 
 from opentelemetry.codegen.json.runtime.json_codec import (
     decode_base64,
+    decode_enum,
     decode_float,
     decode_hex,
     decode_int64,
@@ -20,6 +23,12 @@
 )
 
 
+class _Color(enum.IntEnum):
+    RED = 0
+    GREEN = 1
+    BLUE = 2
+
+
 @pytest.mark.parametrize(
     "value, expected",
     [
@@ -189,3 +198,41 @@ def test_validate_type() -> None:
         match=r"Field 'field' expected \(<class 'int'>, <class 'float'>\), got str",
     ):
         validate_type("s", (int, float), "field")
+
+
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        (1, _Color.GREEN),
+        ("GREEN", _Color.GREEN),
+        (0, _Color.RED),
+        ("RED", _Color.RED),
+    ],
+)
+def test_decode_enum(value: int | str, expected: _Color) -> None:
+    assert decode_enum(value, _Color, "field") is expected
+
+
+@pytest.mark.parametrize(
+    "value, expected_error",
+    [
+        ([], TypeError),
+        (True, TypeError),
+        (False, TypeError),
+        (99, ValueError),
+        ("NOT_A_COLOR", KeyError),
+    ],
+)
+def test_decode_enum_errors(
+    value: Any, expected_error: type[Exception]
+) -> None:
+    with pytest.raises(expected_error):
+        decode_enum(value, _Color, "field")
+
+
+def test_decode_enum_error_messages_include_field_name() -> None:
+    with pytest.raises(ValueError, match="field"):
+        decode_enum(99, _Color, "field")
+
+    with pytest.raises(KeyError, match="field"):
+        decode_enum("NOT_A_COLOR", _Color, "field")

codegen/opentelemetry-codegen-json/tests/test_serde.py

@@ -105,6 +105,19 @@ def test_generated_message_roundtrip(
     assert new_msg == msg
 
 
+def test_enum_field_accepts_name_and_int(
+    test_v1_types: tuple[type[Any], type[Any]],
+) -> None:
+    TestMessage, _ = test_v1_types
+
+    from_name = TestMessage.from_dict({"enumValue": "SUCCESS"})
+    from_int = TestMessage.from_dict({"enumValue": 1})
+
+    assert from_name.enum_value == TestMessage.TestEnum.SUCCESS
+    assert from_int.enum_value == TestMessage.TestEnum.SUCCESS
+    assert from_name == from_int
+
+
 def test_cross_reference(
     common_v1_types: type[Any], trace_v1_types: type[Any]
 ) -> None:
@@ -256,6 +269,25 @@ def test_nested_enum_suite(complex_v1_types: tuple[type[Any], ...]) -> None:
     assert new_msg.repeated_nested == msg.repeated_nested
 
 
+def test_nested_enum_suite_accepts_names(
+    complex_v1_types: tuple[type[Any], ...],
+) -> None:
+    NestedEnumSuite = complex_v1_types[3]
+
+    msg = NestedEnumSuite.from_dict(
+        {
+            "nested": "NESTED_FOO",
+            "repeatedNested": ["NESTED_FOO", "NESTED_BAR"],
+        }
+    )
+
+    assert msg.nested == NestedEnumSuite.NestedEnum.NESTED_FOO
+    assert msg.repeated_nested == [
+        NestedEnumSuite.NestedEnum.NESTED_FOO,
+        NestedEnumSuite.NestedEnum.NESTED_BAR,
+    ]
+
+
 def test_deeply_nested(complex_v1_types: tuple[type[Any], ...]) -> None:
     DeeplyNested = complex_v1_types[4]
 
@@ -304,7 +336,8 @@ def test_defaults_and_none(
         ({"listStrings": "not a list"}, TypeError, "expected <class 'list'>"),
         ({"name": 123}, TypeError, "expected <class 'str'>"),
         ({"subMessage": "not a dict"}, TypeError, "expected <class 'dict'>"),
-        ({"enumValue": "SUCCESS"}, TypeError, "expected <class 'int'>"),
+        ({"enumValue": []}, TypeError, "expected"),
+        ({"enumValue": "NOT_A_NAME"}, KeyError, None),
         ({"listMessages": [None]}, TypeError, "expected <class 'dict'>"),
     ],
 )

opentelemetry-proto-json/src/opentelemetry/proto_json/_json_codec.py

@@ -6,12 +6,14 @@
 import abc
 import base64
 import collections.abc
+import enum
 import json
 import math
 import typing
 
 T = typing.TypeVar("T")
 M = typing.TypeVar("M", bound="JsonMessage")
+EnumT = typing.TypeVar("EnumT", bound=enum.IntEnum)
 
 
 class JsonMessage(abc.ABC):
@@ -246,3 +248,41 @@ def validate_type(
             f"Field '{field_name}' expected {expected_types}, "
             f"got {type(value).__name__}"
         )
+
+
+def decode_enum(
+    value: typing.Any,
+    enum_type: type[EnumT],
+    field_name: str,
+) -> EnumT:
+    """
+    Decode a JSON enum value into an enum member.
+
+    Per the ProtoJSON spec, parsers must accept both enum names (str)
+    and integer values (int).
+
+    Args:
+        value: The enum name or integer value to decode.
+        enum_type: The enum class to decode into.
+        field_name: The name of the field being decoded (for error messages).
+    Returns:
+        The corresponding enum member.
+    """
+    if isinstance(value, bool):
+        raise TypeError(
+            f"Field '{field_name}' expected int or str, got bool"
+        )
+    validate_type(value, (int, str), field_name)
+    if isinstance(value, str):
+        try:
+            return enum_type[value]
+        except KeyError:
+            raise KeyError(
+                f"Invalid enum name '{value}' for field '{field_name}'"
+            ) from None
+    try:
+        return enum_type(value)
+    except ValueError:
+        raise ValueError(
+            f"Invalid enum value {value} for field '{field_name}'"
+        ) from None

opentelemetry-proto-json/src/opentelemetry/proto_json/collector/logs/v1/logs_service.py

@@ -8,17 +8,14 @@
 
 import builtins
 import dataclasses
-import functools
 import typing
 
-_dataclass = functools.partial(dataclasses.dataclass, slots=True)
-
 import opentelemetry.proto_json._json_codec
 import opentelemetry.proto_json.logs.v1.logs
 
 
 @typing.final
-@_dataclass
+@dataclasses.dataclass(slots=True)
 class ExportLogsServiceRequest(opentelemetry.proto_json._json_codec.JsonMessage):
     """
     Generated from protobuf message ExportLogsServiceRequest
@@ -59,7 +56,7 @@ def from_dict(cls, data: builtins.dict[builtins.str, typing.Any]) -> "ExportLogs
 
 
 @typing.final
-@_dataclass
+@dataclasses.dataclass(slots=True)
 class ExportLogsServiceResponse(opentelemetry.proto_json._json_codec.JsonMessage):
     """
     Generated from protobuf message ExportLogsServiceResponse
@@ -100,7 +97,7 @@ def from_dict(cls, data: builtins.dict[builtins.str, typing.Any]) -> "ExportLogs
 
 
 @typing.final
-@_dataclass
+@dataclasses.dataclass(slots=True)
 class ExportLogsPartialSuccess(opentelemetry.proto_json._json_codec.JsonMessage):
     """
     Generated from protobuf message ExportLogsPartialSuccess

opentelemetry-proto-json/src/opentelemetry/proto_json/collector/metrics/v1/metrics_service.py

@@ -8,17 +8,14 @@
 
 import builtins
 import dataclasses
-import functools
 import typing
 
-_dataclass = functools.partial(dataclasses.dataclass, slots=True)
-
 import opentelemetry.proto_json._json_codec
 import opentelemetry.proto_json.metrics.v1.metrics
 
 
 @typing.final
-@_dataclass
+@dataclasses.dataclass(slots=True)
 class ExportMetricsServiceRequest(opentelemetry.proto_json._json_codec.JsonMessage):
     """
     Generated from protobuf message ExportMetricsServiceRequest
@@ -59,7 +56,7 @@ def from_dict(cls, data: builtins.dict[builtins.str, typing.Any]) -> "ExportMetr
 
 
 @typing.final
-@_dataclass
+@dataclasses.dataclass(slots=True)
 class ExportMetricsServiceResponse(opentelemetry.proto_json._json_codec.JsonMessage):
     """
     Generated from protobuf message ExportMetricsServiceResponse
@@ -100,7 +97,7 @@ def from_dict(cls, data: builtins.dict[builtins.str, typing.Any]) -> "ExportMetr
 
 
 @typing.final
-@_dataclass
+@dataclasses.dataclass(slots=True)
 class ExportMetricsPartialSuccess(opentelemetry.proto_json._json_codec.JsonMessage):
     """
     Generated from protobuf message ExportMetricsPartialSuccess