Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/build/lib/airflow/utils/timezone.py: 50%

1# 

2# Licensed to the Apache Software Foundation (ASF) under one 

3# or more contributor license agreements. See the NOTICE file 

4# distributed with this work for additional information 

5# regarding copyright ownership. The ASF licenses this file 

6# to you under the Apache License, Version 2.0 (the 

7# "License"); you may not use this file except in compliance 

8# with the License. You may obtain a copy of the License at 

9# 

10# http://www.apache.org/licenses/LICENSE-2.0 

11# 

12# Unless required by applicable law or agreed to in writing, 

13# software distributed under the License is distributed on an 

14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

15# KIND, either express or implied. See the License for the 

16# specific language governing permissions and limitations 

17# under the License. 

18from __future__ import annotations 

19 

20import datetime as dt 

21from typing import overload 

22 

23import pendulum 

24from dateutil.relativedelta import relativedelta 

25from pendulum.datetime import DateTime 

26 

27from airflow.settings import TIMEZONE 

28 

29# UTC time zone as a tzinfo instance. 

30utc = pendulum.tz.timezone("UTC") 

31 

32 

33def is_localized(value): 

34 """ 

35 Determine if a given datetime.datetime is aware. 

36 The concept is defined in Python's docs: 

37 http://docs.python.org/library/datetime.html#datetime.tzinfo 

38 Assuming value.tzinfo is either None or a proper datetime.tzinfo, 

39 value.utcoffset() implements the appropriate logic. 

40 """ 

41 return value.utcoffset() is not None 

42 

43 

44def is_naive(value): 

45 """ 

46 Determine if a given datetime.datetime is naive. 

47 The concept is defined in Python's docs: 

48 http://docs.python.org/library/datetime.html#datetime.tzinfo 

49 Assuming value.tzinfo is either None or a proper datetime.tzinfo, 

50 value.utcoffset() implements the appropriate logic. 

51 """ 

52 return value.utcoffset() is None 

53 

54 

55def utcnow() -> dt.datetime: 

56 """ 

57 Get the current date and time in UTC 

58 

59 :return: 

60 """ 

61 # pendulum utcnow() is not used as that sets a TimezoneInfo object 

62 # instead of a Timezone. This is not picklable and also creates issues 

63 # when using replace() 

64 result = dt.datetime.utcnow() 

65 result = result.replace(tzinfo=utc) 

66 

67 return result 

68 

69 

70def utc_epoch() -> dt.datetime: 

71 """ 

72 Gets the epoch in the users timezone 

73 

74 :return: 

75 """ 

76 # pendulum utcnow() is not used as that sets a TimezoneInfo object 

77 # instead of a Timezone. This is not picklable and also creates issues 

78 # when using replace() 

79 result = dt.datetime(1970, 1, 1) 

80 result = result.replace(tzinfo=utc) 

81 

82 return result 

83 

84 

85@overload 

86def convert_to_utc(value: None) -> None: 

87 ... 

88 

89 

90@overload 

91def convert_to_utc(value: dt.datetime) -> DateTime: 

92 ... 

93 

94 

95def convert_to_utc(value: dt.datetime | None) -> DateTime | None: 

96 """ 

97 Returns the datetime with the default timezone added if timezone 

98 information was not associated 

99 

100 :param value: datetime 

101 :return: datetime with tzinfo 

102 """ 

103 if value is None: 

104 return value 

105 

106 if not is_localized(value): 

107 value = pendulum.instance(value, TIMEZONE) 

108 

109 return pendulum.instance(value.astimezone(utc)) 

110 

111 

112@overload 

113def make_aware(value: None, timezone: dt.tzinfo | None = None) -> None: 

114 ... 

115 

116 

117@overload 

118def make_aware(value: DateTime, timezone: dt.tzinfo | None = None) -> DateTime: 

119 ... 

120 

121 

122@overload 

123def make_aware(value: dt.datetime, timezone: dt.tzinfo | None = None) -> dt.datetime: 

124 ... 

125 

126 

127def make_aware(value: dt.datetime | None, timezone: dt.tzinfo | None = None) -> dt.datetime | None: 

128 """ 

129 Make a naive datetime.datetime in a given time zone aware. 

130 

131 :param value: datetime 

132 :param timezone: timezone 

133 :return: localized datetime in settings.TIMEZONE or timezone 

134 """ 

135 if timezone is None: 

136 timezone = TIMEZONE 

137 

138 if not value: 

139 return None 

140 

141 # Check that we won't overwrite the timezone of an aware datetime. 

142 if is_localized(value): 

143 raise ValueError(f"make_aware expects a naive datetime, got {value}") 

144 if hasattr(value, "fold"): 

145 # In case of python 3.6 we want to do the same that pendulum does for python3.5 

146 # i.e in case we move clock back we want to schedule the run at the time of the second 

147 # instance of the same clock time rather than the first one. 

148 # Fold parameter has no impact in other cases so we can safely set it to 1 here 

149 value = value.replace(fold=1) 

150 localized = getattr(timezone, "localize", None) 

151 if localized is not None: 

152 # This method is available for pytz time zones 

153 return localized(value) 

154 convert = getattr(timezone, "convert", None) 

155 if convert is not None: 

156 # For pendulum 

157 return convert(value) 

158 # This may be wrong around DST changes! 

159 return value.replace(tzinfo=timezone) 

160 

161 

162def make_naive(value, timezone=None): 

163 """ 

164 Make an aware datetime.datetime naive in a given time zone. 

165 

166 :param value: datetime 

167 :param timezone: timezone 

168 :return: naive datetime 

169 """ 

170 if timezone is None: 

171 timezone = TIMEZONE 

172 

173 # Emulate the behavior of astimezone() on Python < 3.6. 

174 if is_naive(value): 

175 raise ValueError("make_naive() cannot be applied to a naive datetime") 

176 

177 date = value.astimezone(timezone) 

178 

179 # cross library compatibility 

180 naive = dt.datetime( 

181 date.year, date.month, date.day, date.hour, date.minute, date.second, date.microsecond 

182 ) 

183 

184 return naive 

185 

186 

187def datetime(*args, **kwargs): 

188 """ 

189 Wrapper around datetime.datetime that adds settings.TIMEZONE if tzinfo not specified 

190 

191 :return: datetime.datetime 

192 """ 

193 if "tzinfo" not in kwargs: 

194 kwargs["tzinfo"] = TIMEZONE 

195 

196 return dt.datetime(*args, **kwargs) 

197 

198 

199def parse(string: str, timezone=None) -> DateTime: 

200 """ 

201 Parse a time string and return an aware datetime 

202 

203 :param string: time string 

204 :param timezone: the timezone 

205 """ 

206 return pendulum.parse(string, tz=timezone or TIMEZONE, strict=False) # type: ignore 

207 

208 

209@overload 

210def coerce_datetime(v: None, tz: dt.tzinfo | None = None) -> None: 

211 ... 

212 

213 

214@overload 

215def coerce_datetime(v: DateTime, tz: dt.tzinfo | None = None) -> DateTime: 

216 ... 

217 

218 

219@overload 

220def coerce_datetime(v: dt.datetime, tz: dt.tzinfo | None = None) -> DateTime: 

221 ... 

222 

223 

224def coerce_datetime(v: dt.datetime | None, tz: dt.tzinfo | None = None) -> DateTime | None: 

225 """Convert ``v`` into a timezone-aware ``pendulum.DateTime``. 

226 

227 * If ``v`` is *None*, *None* is returned. 

228 * If ``v`` is a naive datetime, it is converted to an aware Pendulum DateTime. 

229 * If ``v`` is an aware datetime, it is converted to a Pendulum DateTime. 

230 Note that ``tz`` is **not** taken into account in this case; the datetime 

231 will maintain its original tzinfo! 

232 """ 

233 if v is None: 

234 return None 

235 if isinstance(v, DateTime): 

236 return v if v.tzinfo else make_aware(v, tz) 

237 # Only dt.datetime is left here. 

238 return pendulum.instance(v if v.tzinfo else make_aware(v, tz)) 

239 

240 

241def td_format(td_object: None | dt.timedelta | float | int) -> str | None: 

242 """ 

243 Format a timedelta object or float/int into a readable string for time duration. 

244 For example timedelta(seconds=3752) would become `1h:2M:32s`. 

245 If the time is less than a second, the return will be `<1s`. 

246 """ 

247 if not td_object: 

248 return None 

249 if isinstance(td_object, dt.timedelta): 

250 delta = relativedelta() + td_object 

251 else: 

252 delta = relativedelta(seconds=int(td_object)) 

253 # relativedelta for timedelta cannot convert days to months 

254 # so calculate months by assuming 30 day months and normalize 

255 months, delta.days = divmod(delta.days, 30) 

256 delta = delta.normalized() + relativedelta(months=months) 

257 

258 def _format_part(key: str) -> str: 

259 value = int(getattr(delta, key)) 

260 if value < 1: 

261 return "" 

262 # distinguish between month/minute following strftime format 

263 # and take first char of each unit, i.e. years='y', days='d' 

264 if key == "minutes": 

265 key = key.upper() 

266 key = key[0] 

267 return f"{value}{key}" 

268 

269 parts = map(_format_part, ("years", "months", "days", "hours", "minutes", "seconds")) 

270 joined = ":".join(part for part in parts if part) 

271 if not joined: 

272 return "<1s" 

273 return joined