Merge pull request #10730: [BEAM-7746] Remove use of @functools.total_ordering

Author: udim

sdks/python/apache_beam/utils/timestamp.py

@@ -26,7 +26,6 @@
 from __future__ import division
 
 import datetime
-import functools
 import time
 from builtins import object
 from typing import Any
@@ -45,9 +44,9 @@
 TimestampTypes = Union[int, float, 'Timestamp']
 # types compatible with Duration.of()
 DurationTypes = Union[int, float, 'Duration']
+TimestampDurationTypes = Union[int, float, 'Duration', 'Timestamp']
 
 
-@functools.total_ordering
 class Timestamp(object):
   """Represents a Unix second timestamp with microsecond granularity.
 
@@ -191,7 +190,7 @@ def __int__(self):
     return self.micros // 1000000
 
   def __eq__(self, other):
-    # type: (Union[int, float, Timestamp, Duration]) -> bool
+    # type: (TimestampDurationTypes) -> bool
     # Allow comparisons between Duration and Timestamp values.
     if not isinstance(other, Duration):
       try:
@@ -206,12 +205,24 @@ def __ne__(self, other):
     return not self == other
 
   def __lt__(self, other):
-    # type: (Union[int, float, Timestamp, Duration]) -> bool
+    # type: (TimestampDurationTypes) -> bool
     # Allow comparisons between Duration and Timestamp values.
     if not isinstance(other, Duration):
       other = Timestamp.of(other)
     return self.micros < other.micros
 
+  def __gt__(self, other):
+    # type: (TimestampDurationTypes) -> bool
+    return not (self < other or self == other)
+
+  def __le__(self, other):
+    # type: (TimestampDurationTypes) -> bool
+    return self < other or self == other
+
+  def __ge__(self, other):
+    # type: (TimestampDurationTypes) -> bool
+    return not self < other
+
   def __hash__(self):
     return hash(self.micros)
 
@@ -252,7 +263,6 @@ def __mod__(self, other):
     common_urns.constants.MAX_TIMESTAMP_MILLIS.constant)*1000)
 
 
-@functools.total_ordering
 class Duration(object):
   """Represents a second duration with microsecond granularity.
 
@@ -344,12 +354,24 @@ def __ne__(self, other):
     return not self == other
 
   def __lt__(self, other):
-    # type: (Union[int, float, Duration, Timestamp]) -> bool
+    # type: (TimestampDurationTypes) -> bool
     # Allow comparisons between Duration and Timestamp values.
     if not isinstance(other, Timestamp):
       other = Duration.of(other)
     return self.micros < other.micros
 
+  def __gt__(self, other):
+    # type: (TimestampDurationTypes) -> bool
+    return not (self < other or self == other)
+
+  def __le__(self, other):
+    # type: (TimestampDurationTypes) -> bool
+    return self < other or self == other
+
+  def __ge__(self, other):
+    # type: (TimestampDurationTypes) -> bool
+    return not self < other
+
   def __hash__(self):
     return hash(self.micros)