AnandInguva commented on code in PR #29849:
URL: https://github.com/apache/beam/pull/29849#discussion_r1434273087
##########
sdks/python/apache_beam/typehints/schemas.py:
##########
@@ -1143,3 +1177,95 @@ def argument_type(cls):
def argument(self):
return self.max_length
+
+
+class EnumerationTypeValue:
+ def __init__(self, enumeration_type, value):
+ self.typ = enumeration_type
+ self.value = value
+
+ def get_typ(self):
+ return self.typ
+
+ def get_value(self):
+ return self.value
+
+
+class EnumerationTypeType:
+ def __init__(self, values: dict):
+ self.values = frozenset(values.items())
+
+ def __call__(self):
+ """Makes this typing.Callable to pass runtime typechecks in <Python3.11
+ """
+ pass
+
+
+ def values(self) -> dict:
+ return self.values
+
+ def __hash__(self):
+ return hash((EnumerationTypeType, self.values))
+
+ def __eq__(self, other):
+ if not (isinstance(other, EnumerationTypeType)):
+ return False
+ return self.values == other.values
+
+ def __ne__(self, other):
+ return not (self == other)
+
+ def __str__(self):
+ return f"EnumerationTypeType({str(self.values)})"
+
+
+class EnumerationTypeValue(EnumerationTypeType):
+ def __init__(self, values, value):
+ super.__init__(values)
+ self.value = value
+
+ def value(self):
+ return self.value
+
+
[email protected]_logical_type
+class EnumerationType(LogicalType[EnumerationTypeValue, np.int32, dict]):
+ def __init__(self, type_map):
+ self.to_rep_map = {v: k for k, v in type_map.items()}
+ self.to_value_map = type_map
+
+ @classmethod
+ def urn(cls):
+ return "beam:logical_type:pythonenum"
+
+ @classmethod
+ def language_type(cls):
+ return EnumerationTypeType
+
+ def _language_type(cls):
+ type_with_argument = EnumerationTypeType(cls.argument())
+ return type_with_argument
+
+ @classmethod
+ def argument_type(cls):
+ # type: () -> type
+ return dict[np.int32, str]
Review Comment:
This will fail with python 3.8 right? https://peps.python.org/pep-0585/
--
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]
