DGS-20014 Add utilities to convert decimals from/to Protobuf (#1946)

* DGS-20014 Add utilities to convert decimals from/to Protobuf

* Add comment

Author: rayokota

src/confluent_kafka/schema_registry/confluent/types/decimal.proto

@@ -9,7 +9,7 @@ message Decimal {
   // The two's-complement representation of the unscaled integer value in big-endian byte order
   bytes value = 1;
 
-  // The precision
+  // The precision (zero indicates unlimited precision)
   uint32 precision = 2;
 
   // The scale

src/confluent_kafka/schema_registry/protobuf.py

@@ -21,6 +21,7 @@
 import struct
 import warnings
 from collections import deque
+from decimal import Context, Decimal, MAX_PREC
 from typing import Set, List, Union, Optional, Any, Tuple
 
 from google.protobuf import descriptor_pb2, any_pb2, api_pb2, empty_pb2, \
@@ -1076,3 +1077,65 @@ def _is_builtin(name: str) -> bool:
     return name.startswith('confluent/') or \
            name.startswith('google/protobuf/') or \
            name.startswith('google/type/')
+
+
+def decimalToProtobuf(value: Decimal, scale: int) -> decimal_pb2.Decimal:
+    """
+    Converts a Decimal to a Protobuf value.
+
+    Args:
+        value (Decimal): The Decimal value to convert.
+
+    Returns:
+        The Protobuf value.
+    """
+    sign, digits, exp = value.as_tuple()
+
+    delta = exp + scale
+
+    if delta < 0:
+        raise ValueError(
+            "Scale provided does not match the decimal")
+
+    unscaled_datum = 0
+    for digit in digits:
+        unscaled_datum = (unscaled_datum * 10) + digit
+
+    unscaled_datum = 10**delta * unscaled_datum
+
+    bytes_req = (unscaled_datum.bit_length() + 8) // 8
+
+    if sign:
+        unscaled_datum = -unscaled_datum
+
+    bytes = unscaled_datum.to_bytes(bytes_req, byteorder="big", signed=True)
+
+    result = decimal_pb2.Decimal()
+    result.value = bytes
+    result.precision = 0
+    result.scale = scale
+    return result
+
+
+decimal_context = Context()
+
+
+def protobufToDecimal(value: decimal_pb2.Decimal) -> Decimal:
+    """
+    Converts a Protobuf value to Decimal.
+
+    Args:
+        value (decimal_pb2.Decimal): The Protobuf value to convert.
+
+    Returns:
+        The Decimal value.
+    """
+    unscaled_datum = int.from_bytes(value.value, byteorder="big", signed=True)
+
+    if value.precision > 0:
+        decimal_context.prec = value.precision
+    else:
+        decimal_context.prec = MAX_PREC
+    return decimal_context.create_decimal(unscaled_datum).scaleb(
+        -value.scale, decimal_context
+    )

tests/schema_registry/test_proto.py

@@ -16,13 +16,16 @@
 # limitations under the License.
 #
 import binascii
+from decimal import Decimal
 from io import BytesIO
 
 import pytest
 
 from confluent_kafka.schema_registry.protobuf import (ProtobufSerializer,
                                                       ProtobufDeserializer,
-                                                      _create_index_array)
+                                                      _create_index_array,
+                                                      decimalToProtobuf,
+                                                      protobufToDecimal)
 from tests.integration.schema_registry.data.proto import (DependencyTestProto_pb2,
                                                           metadata_proto_pb2)
 
@@ -83,3 +86,23 @@ def test_index_encoder(msg_idx, zigzag, expected_hex):
     buf.seek(0)
     decoded_msg_idx = ProtobufDeserializer._read_index_array(buf, zigzag=zigzag)
     assert decoded_msg_idx == msg_idx
+
+
+@pytest.mark.parametrize("decimal, scale", [
+    ("0", 0),
+    ("1.01", 2),
+    ("123456789123456789.56", 2),
+    ("1234", 0),
+    ("1234.5", 1),
+    ("-0", 0),
+    ("-1.01", 2),
+    ("-123456789123456789.56", 2),
+    ("-1234", 0),
+    ("-1234.5", 1),
+    ("-1234.56", 2)
+])
+def test_proto_decimal(decimal, scale):
+    input = Decimal(decimal)
+    converted = decimalToProtobuf(input, scale)
+    result = protobufToDecimal(converted)
+    assert result == input