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 
    16from datetime import timedelta 
    17from datetime import timezone 
    18 
    19from google.protobuf import duration_pb2 
    20from google.protobuf import timestamp_pb2 
    21from proto import datetime_helpers, utils 
    22 
    23 
    24class TimestampRule: 
    25    """A marshal between Python datetimes and protobuf timestamps. 
    26 
    27    Note: Python datetimes are less precise than protobuf datetimes 
    28    (microsecond vs. nanosecond level precision). If nanosecond-level 
    29    precision matters, it is recommended to interact with the internal 
    30    proto directly. 
    31    """ 
    32 
    33    def to_python( 
    34        self, value, *, absent: bool = None 
    35    ) -> datetime_helpers.DatetimeWithNanoseconds: 
    36        if isinstance(value, timestamp_pb2.Timestamp): 
    37            if absent: 
    38                return None 
    39            return datetime_helpers.DatetimeWithNanoseconds.from_timestamp_pb(value) 
    40        return value 
    41 
    42    def to_proto(self, value) -> timestamp_pb2.Timestamp: 
    43        if isinstance(value, datetime_helpers.DatetimeWithNanoseconds): 
    44            return value.timestamp_pb() 
    45        if isinstance(value, datetime): 
    46            return timestamp_pb2.Timestamp( 
    47                seconds=int(value.timestamp()), 
    48                nanos=value.microsecond * 1000, 
    49            ) 
    50        if isinstance(value, str): 
    51            timestamp_value = timestamp_pb2.Timestamp() 
    52            timestamp_value.FromJsonString(value=value) 
    53            return timestamp_value 
    54        return value 
    55 
    56 
    57class DurationRule: 
    58    """A marshal between Python timedeltas and protobuf durations. 
    59 
    60    Note: Python timedeltas are less precise than protobuf durations 
    61    (microsecond vs. nanosecond level precision). If nanosecond-level 
    62    precision matters, it is recommended to interact with the internal 
    63    proto directly. 
    64    """ 
    65 
    66    def to_python(self, value, *, absent: bool = None) -> timedelta: 
    67        if isinstance(value, duration_pb2.Duration): 
    68            return timedelta( 
    69                days=value.seconds // 86400, 
    70                seconds=value.seconds % 86400, 
    71                microseconds=value.nanos // 1000, 
    72            ) 
    73        return value 
    74 
    75    def to_proto(self, value) -> duration_pb2.Duration: 
    76        if isinstance(value, timedelta): 
    77            return duration_pb2.Duration( 
    78                seconds=value.days * 86400 + value.seconds, 
    79                nanos=value.microseconds * 1000, 
    80            ) 
    81        if isinstance(value, str): 
    82            duration_value = duration_pb2.Duration() 
    83            duration_value.FromJsonString(value=value) 
    84            return duration_value 
    85        return value