Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/api_core/datetime_helpers.py: 34%

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_UTC_EPOCH = datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc) 

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

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

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

27_RFC3339_NANOS = re.compile( 

28 r""" 

29 (?P<no_fraction> 

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

31 ) 

32 ( # Optional decimal part 

33 \. # decimal point 

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

35 )? 

36 Z # Zulu 

37""", 

38 re.VERBOSE, 

39) 

40 

41 

42def utcnow(): 

43 """A :meth:`datetime.datetime.utcnow()` alias to allow mocking in tests.""" 

44 return datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None) 

45 

46 

47def to_milliseconds(value): 

48 """Convert a zone-aware datetime to milliseconds since the unix epoch. 

49 

50 Args: 

51 value (datetime.datetime): The datetime to covert. 

52 

53 Returns: 

54 int: Milliseconds since the unix epoch. 

55 """ 

56 micros = to_microseconds(value) 

57 return micros // 1000 

58 

59 

60def from_microseconds(value): 

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

62 

63 Args: 

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

65 

66 Returns: 

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

68 UTC. 

69 """ 

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

71 

72 

73def to_microseconds(value): 

74 """Convert a datetime to microseconds since the unix epoch. 

75 

76 Args: 

77 value (datetime.datetime): The datetime to covert. 

78 

79 Returns: 

80 int: Microseconds since the unix epoch. 

81 """ 

82 if not value.tzinfo: 

83 value = value.replace(tzinfo=datetime.timezone.utc) 

84 # Regardless of what timezone is on the value, convert it to UTC. 

85 value = value.astimezone(datetime.timezone.utc) 

86 # Convert the datetime to a microsecond timestamp. 

87 return int(calendar.timegm(value.timetuple()) * 1e6) + value.microsecond 

88 

89 

90def from_iso8601_date(value): 

91 """Convert a ISO8601 date string to a date. 

92 

93 Args: 

94 value (str): The ISO8601 date string. 

95 

96 Returns: 

97 datetime.date: A date equivalent to the date string. 

98 """ 

99 return datetime.datetime.strptime(value, "%Y-%m-%d").date() 

100 

101 

102def from_iso8601_time(value): 

103 """Convert a zoneless ISO8601 time string to a time. 

104 

105 Args: 

106 value (str): The ISO8601 time string. 

107 

108 Returns: 

109 datetime.time: A time equivalent to the time string. 

110 """ 

111 return datetime.datetime.strptime(value, "%H:%M:%S").time() 

112 

113 

114def from_rfc3339(value): 

115 """Convert an RFC3339-format timestamp to a native datetime. 

116 

117 Supported formats include those without fractional seconds, or with 

118 any fraction up to nanosecond precision. 

119 

120 .. note:: 

121 Python datetimes do not support nanosecond precision; this function 

122 therefore truncates such values to microseconds. 

123 

124 Args: 

125 value (str): The RFC3339 string to convert. 

126 

127 Returns: 

128 datetime.datetime: The datetime object equivalent to the timestamp 

129 in UTC. 

130 

131 Raises: 

132 ValueError: If the timestamp does not match the RFC3339 

133 regular expression. 

134 """ 

135 with_nanos = _RFC3339_NANOS.match(value) 

136 

137 if with_nanos is None: 

138 raise ValueError( 

139 "Timestamp: {!r}, does not match pattern: {!r}".format( 

140 value, _RFC3339_NANOS.pattern 

141 ) 

142 ) 

143 

144 bare_seconds = datetime.datetime.strptime( 

145 with_nanos.group("no_fraction"), _RFC3339_NO_FRACTION 

146 ) 

147 fraction = with_nanos.group("nanos") 

148 

149 if fraction is None: 

150 micros = 0 

151 else: 

152 scale = 9 - len(fraction) 

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

154 micros = nanos // 1000 

155 

156 return bare_seconds.replace(microsecond=micros, tzinfo=datetime.timezone.utc) 

157 

158 

159from_rfc3339_nanos = from_rfc3339 # from_rfc3339_nanos method was deprecated. 

160 

161 

162def to_rfc3339(value, ignore_zone=True): 

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

164 

165 Args: 

166 value (datetime.datetime): 

167 The datetime object to be converted to a string. 

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

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

170 

171 Returns: 

172 str: The RFC3339 formatted string representing the datetime. 

173 """ 

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

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

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

177 

178 return value.strftime(_RFC3339_MICROS) 

179 

180 

181class DatetimeWithNanoseconds(datetime.datetime): 

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

183 

184 Nanosecond can be passed only as a keyword argument. 

185 """ 

186 

187 __slots__ = ("_nanosecond",) 

188 

189 # pylint: disable=arguments-differ 

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

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

192 if nanos > 0: 

193 if "microsecond" in kw: 

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

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

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

197 inst._nanosecond = nanos or 0 

198 return inst 

199 

200 # pylint: disable=arguments-differ 

201 

202 @property 

203 def nanosecond(self): 

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

205 return self._nanosecond 

206 

207 def rfc3339(self): 

208 """Return an RFC3339-compliant timestamp. 

209 

210 Returns: 

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

212 """ 

213 if self._nanosecond == 0: 

214 return to_rfc3339(self) 

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

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

217 

218 @classmethod 

219 def from_rfc3339(cls, stamp): 

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

221 

222 Args: 

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

224 

225 Returns: 

226 :class:`DatetimeWithNanoseconds`: 

227 an instance matching the timestamp string 

228 

229 Raises: 

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

231 """ 

232 with_nanos = _RFC3339_NANOS.match(stamp) 

233 if with_nanos is None: 

234 raise ValueError( 

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

236 stamp, _RFC3339_NANOS.pattern 

237 ) 

238 ) 

239 bare = datetime.datetime.strptime( 

240 with_nanos.group("no_fraction"), _RFC3339_NO_FRACTION 

241 ) 

242 fraction = with_nanos.group("nanos") 

243 if fraction is None: 

244 nanos = 0 

245 else: 

246 scale = 9 - len(fraction) 

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

248 return cls( 

249 bare.year, 

250 bare.month, 

251 bare.day, 

252 bare.hour, 

253 bare.minute, 

254 bare.second, 

255 nanosecond=nanos, 

256 tzinfo=datetime.timezone.utc, 

257 ) 

258 

259 def timestamp_pb(self): 

260 """Return a timestamp message. 

261 

262 Returns: 

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

264 """ 

265 inst = ( 

266 self 

267 if self.tzinfo is not None 

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

269 ) 

270 delta = inst - _UTC_EPOCH 

271 seconds = int(delta.total_seconds()) 

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

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

274 

275 @classmethod 

276 def from_timestamp_pb(cls, stamp): 

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

278 

279 Args: 

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

281 

282 Returns: 

283 :class:`DatetimeWithNanoseconds`: 

284 an instance matching the timestamp message 

285 """ 

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

287 bare = from_microseconds(microseconds) 

288 return cls( 

289 bare.year, 

290 bare.month, 

291 bare.day, 

292 bare.hour, 

293 bare.minute, 

294 bare.second, 

295 nanosecond=stamp.nanos, 

296 tzinfo=datetime.timezone.utc, 

297 )