Abacn commented on code in PR #23014:
URL: https://github.com/apache/beam/pull/23014#discussion_r984083619
##########
sdks/python/apache_beam/typehints/schemas.py:
##########
@@ -751,3 +765,87 @@ def to_representation_type(self, value):
def to_language_type(self, value):
# type: (str) -> PythonCallableWithSource
return PythonCallableWithSource(value)
+
+
+FixedPrecisionDecimalArgumentRepresentation = NamedTuple(
+ 'FixedPrecisionDecimalArgumentRepresentation', [('precision', np.int32),
+ ('scale', np.int32)])
+
+
+class DecimalLogicalType(NoArgumentLogicalType[decimal.Decimal, bytes]):
+ """A logical type for decimal objects handling values consistent with that
+ encoded by ``BigDecimalCoder`` in the Java SDK.
+ """
+ @classmethod
+ def urn(cls):
+ return common_urns.decimal.urn
+
+ @classmethod
+ def representation_type(cls):
+ # type: () -> type
+ return bytes
+
+ @classmethod
+ def language_type(cls):
+ return decimal.Decimal
+
+ def to_representation_type(self, value):
+ # type: (decimal.Decimal) -> bytes
+ return str(value).encode()
+
+ def to_language_type(self, value):
+ # type: (bytes) -> decimal.Decimal
+ return decimal.Decimal(value.decode())
+
+
[email protected]_logical_type
+class FixedPrecisionDecimalLogicalType(
+ LogicalType[decimal.Decimal,
+ DecimalLogicalType,
+ FixedPrecisionDecimalArgumentRepresentation]):
+ """A wrapper of DecimalLogicalType that contains the precision value.
+ """
+ def __init__(self, precision=-1, scale=0):
+ self.precision = precision
+ self.scale = scale
+
+ @classmethod
+ def urn(cls):
+ return "beam:logical_type:fixed_decimal:v1"
+
+ @classmethod
+ def representation_type(cls):
+ # type: () -> type
+ return DecimalLogicalType
+
+ @classmethod
+ def language_type(cls):
+ return decimal.Decimal
+
+ def to_representation_type(self, value):
+ # type: (decimal.Decimal) -> bytes
+
+ return DecimalLogicalType().to_representation_type(value)
+
+ def to_language_type(self, value):
+ # type: (bytes) -> decimal.Decimal
+
+ return DecimalLogicalType().to_language_type(value)
+
+ @classmethod
+ def argument_type(cls):
+ return FixedPrecisionDecimalArgumentRepresentation
+
+ def argument(self):
+ return FixedPrecisionDecimalArgumentRepresentation(
+ precision=self.precision, scale=self.scale)
+
+ @classmethod
+ def _from_typing(cls, typ):
+ print(typ)
Review Comment:
oops thanks for the catch, will double check
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
