Api_core: Convert 'DatetimeWithNanos' to / from 'google.protobuf.timestamp_pb2.Timestamp' (#6919)

Toward #6547.

1 files changed, 38 insertions, 0 deletions

diff --git a/google/api_core/datetime_helpers.py b/google/api_core/datetime_helpers.py
index 3f3523b..b0d9105 100644
--- a/google/api_core/datetime_helpers.py
+++ b/google/api_core/datetime_helpers.py
@@ -20,6 +20,8 @@ import re
 
 import pytz
 
+from google.protobuf import timestamp_pb2
+
 
 _UTC_EPOCH = datetime.datetime.utcfromtimestamp(0).replace(tzinfo=pytz.utc)
 _RFC3339_MICROS = "%Y-%m-%dT%H:%M:%S.%fZ"
@@ -263,3 +265,39 @@ class DatetimeWithNanoseconds(datetime.datetime):
             nanosecond=nanos,
             tzinfo=pytz.UTC,
         )
+
+    def timestamp_pb(self):
+        """Return a timestamp message.
+
+        Returns:
+            (:class:`~google.protobuf.timestamp_pb2.Timestamp`): Timestamp message
+        """
+        inst = self if self.tzinfo is not None else self.replace(tzinfo=pytz.UTC)
+        delta = inst - _UTC_EPOCH
+        seconds = int(delta.total_seconds())
+        nanos = self._nanosecond or self.microsecond * 1000
+        return timestamp_pb2.Timestamp(seconds=seconds, nanos=nanos)
+
+    @classmethod
+    def from_timestamp_pb(cls, stamp):
+        """Parse RFC 3339-compliant timestamp, preserving nanoseconds.
+
+        Args:
+            stamp (:class:`~google.protobuf.timestamp_pb2.Timestamp`): timestamp message
+
+        Returns:
+            :class:`DatetimeWithNanoseconds`:
+                an instance matching the timestamp message
+        """
+        microseconds = int(stamp.seconds * 1e6)
+        bare = from_microseconds(microseconds)
+        return cls(
+            bare.year,
+            bare.month,
+            bare.day,
+            bare.hour,
+            bare.minute,
+            bare.second,
+            nanosecond=stamp.nanos,
+            tzinfo=pytz.UTC,
+        )