Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/proto/datetime_helpers.py: 31%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

74 statements  

1# Copyright 2017 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# http://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 

15"""Helpers for :mod:`datetime`.""" 

16 

17import calendar 

18import datetime 

19import re 

20 

21from google.protobuf import timestamp_pb2 

22 

23 

24_UTC_EPOCH = datetime.datetime.fromtimestamp(0, datetime.timezone.utc) 

25 

26_RFC3339_MICROS = "%Y-%m-%dT%H:%M:%S.%fZ" 

27_RFC3339_NO_FRACTION = "%Y-%m-%dT%H:%M:%S" 

28# datetime.strptime cannot handle nanosecond precision: parse w/ regex 

29_RFC3339_NANOS = re.compile( 

30 r""" 

31 (?P<no_fraction> 

32 \d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2} # YYYY-MM-DDTHH:MM:SS 

33 ) 

34 ( # Optional decimal part 

35 \. # decimal point 

36 (?P<nanos>\d{1,9}) # nanoseconds, maybe truncated 

37 )? 

38 Z # Zulu 

39""", 

40 re.VERBOSE, 

41) 

42 

43 

44def _from_microseconds(value): 

45 """Convert timestamp in microseconds since the unix epoch to datetime. 

46 

47 Args: 

48 value (float): The timestamp to convert, in microseconds. 

49 

50 Returns: 

51 datetime.datetime: The datetime object equivalent to the timestamp in 

52 UTC. 

53 """ 

54 return _UTC_EPOCH + datetime.timedelta(microseconds=value) 

55 

56 

57def _to_rfc3339(value, ignore_zone=True): 

58 """Convert a datetime to an RFC3339 timestamp string. 

59 

60 Args: 

61 value (datetime.datetime): 

62 The datetime object to be converted to a string. 

63 ignore_zone (bool): If True, then the timezone (if any) of the 

64 datetime object is ignored and the datetime is treated as UTC. 

65 

66 Returns: 

67 str: The RFC3339 formatted string representing the datetime. 

68 """ 

69 if not ignore_zone and value.tzinfo is not None: 

70 # Convert to UTC and remove the time zone info. 

71 value = value.replace(tzinfo=None) - value.utcoffset() 

72 

73 return value.strftime(_RFC3339_MICROS) 

74 

75 

76class DatetimeWithNanoseconds(datetime.datetime): 

77 """Track nanosecond in addition to normal datetime attrs. 

78 

79 Nanosecond can be passed only as a keyword argument. 

80 """ 

81 

82 __slots__ = ("_nanosecond",) 

83 

84 # pylint: disable=arguments-differ 

85 def __new__(cls, *args, **kw): 

86 nanos = kw.pop("nanosecond", 0) 

87 if nanos > 0: 

88 if "microsecond" in kw: 

89 raise TypeError("Specify only one of 'microsecond' or 'nanosecond'") 

90 kw["microsecond"] = nanos // 1000 

91 inst = datetime.datetime.__new__(cls, *args, **kw) 

92 inst._nanosecond = nanos or 0 

93 return inst 

94 

95 # pylint: disable=arguments-differ 

96 def replace(self, *args, **kw): 

97 """Return a date with the same value, except for those parameters given 

98 new values by whichever keyword arguments are specified. For example, 

99 if d == date(2002, 12, 31), then 

100 d.replace(day=26) == date(2002, 12, 26). 

101 NOTE: nanosecond and microsecond are mutually exclusive arguments. 

102 """ 

103 

104 ms_provided = "microsecond" in kw 

105 ns_provided = "nanosecond" in kw 

106 provided_ns = kw.pop("nanosecond", 0) 

107 

108 prev_nanos = self.nanosecond 

109 

110 if ms_provided and ns_provided: 

111 raise TypeError("Specify only one of 'microsecond' or 'nanosecond'") 

112 

113 if ns_provided: 

114 # if nanos were provided, manipulate microsecond kw arg to super 

115 kw["microsecond"] = provided_ns // 1000 

116 inst = super().replace(*args, **kw) 

117 

118 if ms_provided: 

119 # ms were provided, nanos are invalid, build from ms 

120 inst._nanosecond = inst.microsecond * 1000 

121 elif ns_provided: 

122 # ns were provided, replace nanoseconds to match after calling super 

123 inst._nanosecond = provided_ns 

124 else: 

125 # if neither ms or ns were provided, passthru previous nanos. 

126 inst._nanosecond = prev_nanos 

127 

128 return inst 

129 

130 @property 

131 def nanosecond(self): 

132 """Read-only: nanosecond precision.""" 

133 return self._nanosecond or self.microsecond * 1000 

134 

135 def rfc3339(self): 

136 """Return an RFC3339-compliant timestamp. 

137 

138 Returns: 

139 (str): Timestamp string according to RFC3339 spec. 

140 """ 

141 if self._nanosecond == 0: 

142 return _to_rfc3339(self) 

143 nanos = str(self._nanosecond).rjust(9, "0").rstrip("0") 

144 return "{}.{}Z".format(self.strftime(_RFC3339_NO_FRACTION), nanos) 

145 

146 @classmethod 

147 def from_rfc3339(cls, stamp): 

148 """Parse RFC3339-compliant timestamp, preserving nanoseconds. 

149 

150 Args: 

151 stamp (str): RFC3339 stamp, with up to nanosecond precision 

152 

153 Returns: 

154 :class:`DatetimeWithNanoseconds`: 

155 an instance matching the timestamp string 

156 

157 Raises: 

158 ValueError: if `stamp` does not match the expected format 

159 """ 

160 with_nanos = _RFC3339_NANOS.match(stamp) 

161 if with_nanos is None: 

162 raise ValueError( 

163 "Timestamp: {}, does not match pattern: {}".format( 

164 stamp, _RFC3339_NANOS.pattern 

165 ) 

166 ) 

167 bare = datetime.datetime.strptime( 

168 with_nanos.group("no_fraction"), _RFC3339_NO_FRACTION 

169 ) 

170 fraction = with_nanos.group("nanos") 

171 if fraction is None: 

172 nanos = 0 

173 else: 

174 scale = 9 - len(fraction) 

175 nanos = int(fraction) * (10**scale) 

176 return cls( 

177 bare.year, 

178 bare.month, 

179 bare.day, 

180 bare.hour, 

181 bare.minute, 

182 bare.second, 

183 nanosecond=nanos, 

184 tzinfo=datetime.timezone.utc, 

185 ) 

186 

187 def timestamp_pb(self): 

188 """Return a timestamp message. 

189 

190 Returns: 

191 (:class:`~google.protobuf.timestamp_pb2.Timestamp`): Timestamp message 

192 """ 

193 inst = ( 

194 self 

195 if self.tzinfo is not None 

196 else self.replace(tzinfo=datetime.timezone.utc) 

197 ) 

198 delta = inst - _UTC_EPOCH 

199 seconds = int(delta.total_seconds()) 

200 nanos = self._nanosecond or self.microsecond * 1000 

201 return timestamp_pb2.Timestamp(seconds=seconds, nanos=nanos) 

202 

203 @classmethod 

204 def from_timestamp_pb(cls, stamp): 

205 """Parse RFC3339-compliant timestamp, preserving nanoseconds. 

206 

207 Args: 

208 stamp (:class:`~google.protobuf.timestamp_pb2.Timestamp`): timestamp message 

209 

210 Returns: 

211 :class:`DatetimeWithNanoseconds`: 

212 an instance matching the timestamp message 

213 """ 

214 microseconds = int(stamp.seconds * 1e6) 

215 bare = _from_microseconds(microseconds) 

216 return cls( 

217 bare.year, 

218 bare.month, 

219 bare.day, 

220 bare.hour, 

221 bare.minute, 

222 bare.second, 

223 nanosecond=stamp.nanos, 

224 tzinfo=datetime.timezone.utc, 

225 )