Skip to content

Commit

Permalink
Api_core: Convert 'DatetimeWithNanos' to / from 'google.protobuf.time…
Browse files Browse the repository at this point in the history
…stamp_pb2.Timestamp' (#6919)

Toward #6547.
  • Loading branch information
tseaver authored Dec 13, 2018
1 parent 64ea005 commit df51357
Show file tree
Hide file tree
Showing 2 changed files with 193 additions and 87 deletions.
38 changes: 38 additions & 0 deletions api_core/google/api_core/datetime_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@

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"
Expand Down Expand Up @@ -263,3 +265,39 @@ def from_rfc3339(cls, stamp):
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,
)
242 changes: 155 additions & 87 deletions api_core/tests/unit/test_datetime_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,12 +12,14 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import calendar
import datetime

import pytest
import pytz

from google.api_core import datetime_helpers
from google.protobuf import timestamp_pb2


ONE_MINUTE_IN_MICROSECONDS = 60 * 1e6
Expand Down Expand Up @@ -154,93 +156,159 @@ def test_to_rfc3339_with_non_utc_ignore_zone():
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
class Test_DateTimeWithNanos(object):

@staticmethod
def test_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

@staticmethod
def test_ctor_w_nanos():
stamp = datetime_helpers.DatetimeWithNanoseconds(
2016, 12, 20, 21, 13, 47, 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
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

@staticmethod
def test_ctor_w_micros_positional_and_nanos():
with pytest.raises(TypeError):
datetime_helpers.DatetimeWithNanoseconds(
2016, 12, 20, 21, 13, 47, 123456, nanosecond=123456789
)

@staticmethod
def test_ctor_w_micros_keyword_and_nanos():
with pytest.raises(TypeError):
datetime_helpers.DatetimeWithNanoseconds(
2016, 12, 20, 21, 13, 47, microsecond=123456, nanosecond=123456789
)

@staticmethod
def test_rfc3339_wo_nanos():
stamp = datetime_helpers.DatetimeWithNanoseconds(2016, 12, 20, 21, 13, 47, 123456)
assert stamp.rfc3339() == "2016-12-20T21:13:47.123456Z"

@staticmethod
def test_rfc3339_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_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
@staticmethod
def test_rfc3339_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"

@staticmethod
def test_from_rfc3339_w_invalid():
stamp = "2016-12-20T21:13:47"
with pytest.raises(ValueError):
datetime_helpers.DatetimeWithNanoseconds.from_rfc3339(stamp)

@staticmethod
def test_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

@staticmethod
def test_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

@staticmethod
def test_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

@staticmethod
def test_timestamp_pb_wo_nanos_naive():
stamp = datetime_helpers.DatetimeWithNanoseconds(
2016, 12, 20, 21, 13, 47, 123456)
delta = stamp.replace(tzinfo=pytz.UTC) - datetime_helpers._UTC_EPOCH
seconds = int(delta.total_seconds())
nanos = 123456000
timestamp = timestamp_pb2.Timestamp(seconds=seconds, nanos=nanos)
assert stamp.timestamp_pb() == timestamp

@staticmethod
def test_timestamp_pb_w_nanos():
stamp = datetime_helpers.DatetimeWithNanoseconds(
2016, 12, 20, 21, 13, 47, nanosecond=123456789, tzinfo=pytz.UTC
)
delta = stamp - datetime_helpers._UTC_EPOCH
timestamp = timestamp_pb2.Timestamp(
seconds=int(delta.total_seconds()), nanos=123456789)
assert stamp.timestamp_pb() == timestamp

@staticmethod
def test_from_timestamp_pb_wo_nanos():
when = datetime.datetime(2016, 12, 20, 21, 13, 47, 123456, tzinfo=pytz.UTC)
delta = when - datetime_helpers._UTC_EPOCH
seconds = int(delta.total_seconds())
timestamp = timestamp_pb2.Timestamp(seconds=seconds)

stamp = datetime_helpers.DatetimeWithNanoseconds.from_timestamp_pb(
timestamp)

assert _to_seconds(when) == _to_seconds(stamp)
assert stamp.microsecond == 0
assert stamp.nanosecond == 0
assert stamp.tzinfo == pytz.UTC

@staticmethod
def test_from_timestamp_pb_w_nanos():
when = datetime.datetime(2016, 12, 20, 21, 13, 47, 123456, tzinfo=pytz.UTC)
delta = when - datetime_helpers._UTC_EPOCH
seconds = int(delta.total_seconds())
timestamp = timestamp_pb2.Timestamp(seconds=seconds, nanos=123456789)

stamp = datetime_helpers.DatetimeWithNanoseconds.from_timestamp_pb(
timestamp)

assert _to_seconds(when) == _to_seconds(stamp)
assert stamp.microsecond == 123456
assert stamp.nanosecond == 123456789
assert stamp.tzinfo == pytz.UTC


def _to_seconds(value):
"""Convert a datetime to seconds since the unix epoch.
Args:
value (datetime.datetime): The datetime to covert.
Returns:
int: Microseconds since the unix epoch.
"""
assert value.tzinfo is pytz.UTC
return calendar.timegm(value.timetuple())

0 comments on commit df51357

Please sign in to comment.