-
Notifications
You must be signed in to change notification settings - Fork 15.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Nextgen Proto Pythonic API: Add timestamp.py
PiperOrigin-RevId: 660474875
- Loading branch information
1 parent
737a05b
commit ee26f27
Showing
2 changed files
with
176 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,64 @@ | ||
# -*- coding: utf-8 -*- | ||
# Protocol Buffers - Google's data interchange format | ||
# Copyright 2008 Google Inc. All rights reserved. | ||
# | ||
# Use of this source code is governed by a BSD-style | ||
# license that can be found in the LICENSE file or at | ||
# https://developers.google.com/open-source/licenses/bsd | ||
|
||
"""Tests proto_time_util APIs.""" | ||
|
||
import datetime | ||
import unittest | ||
|
||
from google.protobuf import timestamp | ||
from google.protobuf.internal import well_known_types_test_pb2 | ||
|
||
from google.protobuf import timestamp_pb2 | ||
|
||
|
||
class ProtoTimeUtilTest(unittest.TestCase): | ||
|
||
def test_timestamp_integer_conversion(self): | ||
self.assertEqual(1, timestamp.to_nanoseconds(timestamp.from_nanoseconds(1))) | ||
self.assertEqual(-1, timestamp.to_seconds(timestamp.from_seconds(-1))) | ||
self.assertEqual( | ||
123, timestamp.to_milliseconds(timestamp.from_milliseconds(123)) | ||
) | ||
self.assertEqual( | ||
321, timestamp.to_microseconds(timestamp.from_microseconds(321)) | ||
) | ||
|
||
def test_timstamp_current(self): | ||
# It is not easy to check with current time. For test coverage only. | ||
self.assertNotEqual(8 * 3600, timestamp.from_current_time().seconds) | ||
|
||
def test_timestamp_json(self): | ||
|
||
def check_timestamp(ts, text): | ||
self.assertEqual(text, timestamp.to_json_string(ts)) | ||
parsed_ts = timestamp.from_json_string(text) | ||
self.assertEqual(ts, parsed_ts) | ||
|
||
message = timestamp_pb2.Timestamp() | ||
message.seconds = 0 | ||
message.nanos = 0 | ||
check_timestamp(message, '1970-01-01T00:00:00Z') | ||
message.nanos = 10000000 | ||
check_timestamp(message, '1970-01-01T00:00:00.010Z') | ||
message.nanos = 10000 | ||
check_timestamp(message, '1970-01-01T00:00:00.000010Z') | ||
|
||
def test_timestamp_datetime(self): | ||
naive_utc_epoch = datetime.datetime(1970, 1, 1) | ||
message = well_known_types_test_pb2.WKTMessage() | ||
message.optional_timestamp = naive_utc_epoch | ||
self.assertEqual(0, message.optional_timestamp.seconds) # pytype: disable=attribute-error | ||
self.assertEqual(0, message.optional_timestamp.nanos) # pytype: disable=attribute-error | ||
self.assertEqual( | ||
naive_utc_epoch, timestamp.to_datetime(message.optional_timestamp) # pytype: disable=wrong-arg-types | ||
) | ||
|
||
|
||
if __name__ == '__main__': | ||
unittest.main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,112 @@ | ||
# Protocol Buffers - Google's data interchange format | ||
# Copyright 2008 Google Inc. All rights reserved. | ||
# | ||
# Use of this source code is governed by a BSD-style | ||
# license that can be found in the LICENSE file or at | ||
# https://developers.google.com/open-source/licenses/bsd | ||
|
||
"""Contains the Timestamp helper APIs.""" | ||
|
||
import datetime | ||
from typing import Optional | ||
|
||
from google.protobuf.timestamp_pb2 import Timestamp | ||
|
||
|
||
def from_json_string(value: str) -> Timestamp: | ||
"""Parse a RFC 3339 date string format to Timestamp. | ||
Args: | ||
value: A date string. Any fractional digits (or none) and any offset are | ||
accepted as long as they fit into nano-seconds precision. Example of | ||
accepted format: '1972-01-01T10:00:20.021-05:00' | ||
Raises: | ||
ValueError: On parsing problems. | ||
""" | ||
timestamp = Timestamp() | ||
timestamp.FromJsonString(value) | ||
return timestamp | ||
|
||
|
||
def from_microseconds(micros: float) -> Timestamp: | ||
"""Converts microseconds since epoch to Timestamp.""" | ||
timestamp = Timestamp() | ||
timestamp.FromMicroseconds(micros) | ||
return timestamp | ||
|
||
|
||
def from_milliseconds(millis: float) -> Timestamp: | ||
"""Converts milliseconds since epoch to Timestamp.""" | ||
timestamp = Timestamp() | ||
timestamp.FromMilliseconds(millis) | ||
return timestamp | ||
|
||
|
||
def from_nanoseconds(nanos: float) -> Timestamp: | ||
"""Converts nanoseconds since epoch to Timestamp.""" | ||
timestamp = Timestamp() | ||
timestamp.FromNanoseconds(nanos) | ||
return timestamp | ||
|
||
|
||
def from_seconds(seconds: float) -> Timestamp: | ||
"""Converts seconds since epoch to Timestamp.""" | ||
timestamp = Timestamp() | ||
timestamp.FromSeconds(seconds) | ||
return timestamp | ||
|
||
|
||
def from_current_time() -> Timestamp: | ||
"""Converts the current UTC to Timestamp.""" | ||
timestamp = Timestamp() | ||
timestamp.FromDatetime(datetime.datetime.now(tz=datetime.timezone.utc)) | ||
return timestamp | ||
|
||
|
||
def to_json_string(ts: Timestamp) -> str: | ||
"""Converts Timestamp to RFC 3339 date string format. | ||
Returns: | ||
A string converted from timestamp. The string is always Z-normalized | ||
and uses 3, 6 or 9 fractional digits as required to represent the | ||
exact time. Example of the return format: '1972-01-01T10:00:20.021Z' | ||
""" | ||
return ts.ToJsonString() | ||
|
||
|
||
def to_microseconds(ts: Timestamp) -> int: | ||
"""Converts Timestamp to microseconds since epoch.""" | ||
return ts.ToMicroseconds() | ||
|
||
|
||
def to_milliseconds(ts: Timestamp) -> int: | ||
"""Converts Timestamp to milliseconds since epoch.""" | ||
return ts.ToMilliseconds() | ||
|
||
|
||
def to_nanoseconds(ts: Timestamp) -> int: | ||
"""Converts Timestamp to nanoseconds since epoch.""" | ||
return ts.ToNanoseconds() | ||
|
||
|
||
def to_seconds(ts: Timestamp) -> int: | ||
"""Converts Timestamp to seconds since epoch.""" | ||
return ts.ToSeconds() | ||
|
||
|
||
def to_datetime( | ||
ts: Timestamp, tz: Optional[datetime.tzinfo] = None | ||
) -> datetime.datetime: | ||
"""Converts Timestamp to a datetime. | ||
Args: | ||
tz: A datetime.tzinfo subclass; defaults to None. | ||
Returns: | ||
If tzinfo is None, returns a timezone-naive UTC datetime (with no timezone | ||
information, i.e. not aware that it's UTC). | ||
Otherwise, returns a timezone-aware datetime in the input timezone. | ||
""" | ||
return ts.ToDatetime(tzinfo=tz) |