Move TimestampNanoSeconds to api_core

api_core/google/api_core/datetime_helpers.py

@@ -179,3 +179,71 @@ def to_rfc3339(value, ignore_zone=True):
  value = value.replace(tzinfo=None) - value.utcoffset()
 
  return value.strftime(_RFC3339_MICROS)
+
+
+class DatetimeWithNanoseconds(datetime.datetime):
+ """Track nanosecond in addition to normal datetime attrs.
+
+ Nanosecond can be passed only as a keyword argument.
+ """
+ __slots__ = ('_nanosecond',)
+
+ # pylint: disable=arguments-differ
+ def __new__(cls, *args, **kw):
+ nanos = kw.pop('nanosecond', 0)
+ if nanos > 0:
+ if 'microsecond' in kw:
+ raise TypeError(
+ "Specify only one of 'microsecond' or 'nanosecond'")
+ kw['microsecond'] = nanos // 1000
+ inst = datetime.datetime.__new__(cls, *args, **kw)
+ inst._nanosecond = nanos or 0
+ return inst
+ # pylint: disable=arguments-differ
+
+ @property
+ def nanosecond(self):
+ """Read-only: nanosecond precision."""
+ return self._nanosecond
+
+ def rfc3339(self):
+ """Return an RFC 3339-compliant timestamp.
+
+ Returns:
+ (str): Timestamp string according to RFC 3339 spec.
+ """
+ if self._nanosecond == 0:
+ return to_rfc3339(self)
+ nanos = str(self._nanosecond).rstrip('0')
+ return '{}.{}Z'.format(self.strftime(_RFC3339_NO_FRACTION), nanos)
+
+ @classmethod
+ def from_rfc3339(cls, stamp):
+ """Parse RFC 3339-compliant timestamp, preserving nanoseconds.
+
+ Args:
+ stamp (str): RFC 3339 stamp, with up to nanosecond precision
+
+ Returns:
+ :class:`DatetimeWithNanoseconds`:
+ an instance matching the timestamp string
+
+ Raises:
+ ValueError: if `stamp` does not match the expected format
+ """
+ with_nanos = _RFC3339_NANOS.match(stamp)
+ if with_nanos is None:
+ raise ValueError(
+ 'Timestamp: {}, does not match pattern: {}'.format(
+ stamp, _RFC3339_NANOS.pattern))
+ bare = datetime.datetime.strptime(
+ with_nanos.group('no_fraction'), _RFC3339_NO_FRACTION)
+ fraction = with_nanos.group('nanos')
+ if fraction is None:
+ nanos = 0
+ else:
+ scale = 9 - len(fraction)
+ nanos = int(fraction) * (10 ** scale)
+ return cls(bare.year, bare.month, bare.day,
+ bare.hour, bare.minute, bare.second,
+ nanosecond=nanos, tzinfo=pytz.UTC)

api_core/tests/unit/test_datetime_helpers.py

@@ -19,6 +19,7 @@
 
 from google.api_core import datetime_helpers
 
+
 ONE_MINUTE_IN_MICROSECONDS = 60 * 1e6
 
 
@@ -148,3 +149,95 @@ def test_to_rfc3339_with_non_utc_ignore_zone():
  value = datetime.datetime(2016, 4, 5, 13, 30, 0, tzinfo=zone)
  expected = '2016-04-05T13:30:00.000000Z'
  assert datetime_helpers.to_rfc3339(value, ignore_zone=True) == expected
+
+
+def test_datetimewithnanos_ctor_wo_nanos():
+ stamp = datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47, 123456)
+ assert stamp.year == 2016
+ assert stamp.month == 12
+ assert stamp.day == 20
+ assert stamp.hour == 21
+ assert stamp.minute == 13
+ assert stamp.second == 47
+ assert stamp.microsecond == 123456
+ assert stamp.nanosecond == 0
+
+
+def test_datetimewithnanos_ctor_w_nanos():
+ stamp = datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47, nanosecond=123456789)
+ assert stamp.year == 2016
+ assert stamp.month == 12
+ assert stamp.day == 20
+ assert stamp.hour == 21
+ assert stamp.minute == 13
+ assert stamp.second == 47
+ assert stamp.microsecond == 123456
+ assert stamp.nanosecond == 123456789
+
+
+def test_datetimewithnanos_ctor_w_micros_positional_and_nanos():
+ with pytest.raises(TypeError):
+ datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47, 123456, nanosecond=123456789)
+
+
+def test_datetimewithnanos_ctor_w_micros_keyword_and_nanos():
+ with pytest.raises(TypeError):
+ datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47,
+ microsecond=123456, nanosecond=123456789)
+
+
+def test_datetimewithnanos_rfc339_wo_nanos():
+ stamp = datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47, 123456)
+ assert stamp.rfc3339() == '2016-12-20T21:13:47.123456Z'
+
+
+def test_datetimewithnanos_rfc339_w_nanos():
+ stamp = datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47, nanosecond=123456789)
+ assert stamp.rfc3339() == '2016-12-20T21:13:47.123456789Z'
+
+
+def test_datetimewithnanos_rfc339_w_nanos_no_trailing_zeroes():
+ stamp = datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47, nanosecond=100000000)
+ assert stamp.rfc3339() == '2016-12-20T21:13:47.1Z'
+
+
+def test_datetimewithnanos_from_rfc3339_w_invalid():
+ stamp = '2016-12-20T21:13:47'
+ with pytest.raises(ValueError):
+ datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(stamp)
+
+
+def test_datetimewithnanos_from_rfc3339_wo_fraction():
+ timestamp = '2016-12-20T21:13:47Z'
+ expected = datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47,
+ tzinfo=pytz.UTC)
+ stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp)
+ assert (stamp == expected)
+
+
+def test_datetimewithnanos_from_rfc3339_w_partial_precision():
+ timestamp = '2016-12-20T21:13:47.1Z'
+ expected = datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47,
+ microsecond=100000,
+ tzinfo=pytz.UTC)
+ stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp)
+ assert stamp == expected
+
+
+def test_datetimewithnanos_from_rfc3339_w_full_precision():
+ timestamp = '2016-12-20T21:13:47.123456789Z'
+ expected = datetime_helpers.DatetimeWithNanoseconds(
+ 2016, 12, 20, 21, 13, 47,
+ nanosecond=123456789,
+ tzinfo=pytz.UTC)
+ stamp = datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(timestamp)
+ assert stamp == expected