1# Copyright 2018 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15from datetime import datetime, timedelta
16
17from google.protobuf import duration_pb2, timestamp_pb2
18
19from proto import datetime_helpers
20
21
22class TimestampRule:
23 """A marshal between Python datetimes and protobuf timestamps.
24
25 Note: Python datetimes are less precise than protobuf datetimes
26 (microsecond vs. nanosecond level precision). If nanosecond-level
27 precision matters, it is recommended to interact with the internal
28 proto directly.
29 """
30
31 def to_python(
32 self, value, *, absent: bool = None
33 ) -> datetime_helpers.DatetimeWithNanoseconds:
34 if isinstance(value, timestamp_pb2.Timestamp):
35 if absent:
36 return None
37 return datetime_helpers.DatetimeWithNanoseconds.from_timestamp_pb(value)
38 return value
39
40 def to_proto(self, value) -> timestamp_pb2.Timestamp:
41 if isinstance(value, datetime_helpers.DatetimeWithNanoseconds):
42 return value.timestamp_pb()
43 if isinstance(value, datetime):
44 return timestamp_pb2.Timestamp(
45 seconds=int(value.timestamp()),
46 nanos=value.microsecond * 1000,
47 )
48 if isinstance(value, str):
49 timestamp_value = timestamp_pb2.Timestamp()
50 timestamp_value.FromJsonString(value=value)
51 return timestamp_value
52 return value
53
54
55class DurationRule:
56 """A marshal between Python timedeltas and protobuf durations.
57
58 Note: Python timedeltas are less precise than protobuf durations
59 (microsecond vs. nanosecond level precision). If nanosecond-level
60 precision matters, it is recommended to interact with the internal
61 proto directly.
62 """
63
64 def to_python(self, value, *, absent: bool = None) -> timedelta:
65 if isinstance(value, duration_pb2.Duration):
66 return timedelta(
67 days=value.seconds // 86400,
68 seconds=value.seconds % 86400,
69 microseconds=value.nanos // 1000,
70 )
71 return value
72
73 def to_proto(self, value) -> duration_pb2.Duration:
74 if isinstance(value, timedelta):
75 return duration_pb2.Duration(
76 seconds=value.days * 86400 + value.seconds,
77 nanos=value.microseconds * 1000,
78 )
79 if isinstance(value, str):
80 duration_value = duration_pb2.Duration()
81 duration_value.FromJsonString(value=value)
82 return duration_value
83 return value