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

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 

27# UTC time zone as a tzinfo instance. 

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

29 

30 

31def is_localized(value): 

32 """Determine if a given datetime.datetime is aware. 

33 

34 The concept is defined in Python documentation. Assuming the tzinfo is 

35 either None or a proper ``datetime.tzinfo`` instance, ``value.utcoffset()`` 

36 implements the appropriate logic. 

37 

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

39 """ 

40 return value.utcoffset() is not None 

41 

42 

43def is_naive(value): 

44 """Determine if a given datetime.datetime is naive. 

45 

46 The concept is defined in Python documentation. Assuming the tzinfo is 

47 either None or a proper ``datetime.tzinfo`` instance, ``value.utcoffset()`` 

48 implements the appropriate logic. 

49 

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

51 """ 

52 return value.utcoffset() is None 

53 

54 

55def utcnow() -> dt.datetime: 

56 """Get the current date and time in UTC.""" 

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

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

59 # when using replace() 

60 result = dt.datetime.utcnow() 

61 result = result.replace(tzinfo=utc) 

62 

63 return result 

64 

65 

66def utc_epoch() -> dt.datetime: 

67 """Gets the epoch in the users timezone.""" 

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

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

70 # when using replace() 

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

72 result = result.replace(tzinfo=utc) 

73 

74 return result 

75 

76 

77@overload 

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

79 ... 

80 

81 

82@overload 

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

84 ... 

85 

86 

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

88 """Creates a datetime with the default timezone added if none is associated. 

89 

90 :param value: datetime 

91 :return: datetime with tzinfo 

92 """ 

93 if value is None: 

94 return value 

95 

96 if not is_localized(value): 

97 from airflow.settings import TIMEZONE 

98 

99 value = pendulum.instance(value, TIMEZONE) 

100 

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

102 

103 

104@overload 

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

106 ... 

107 

108 

109@overload 

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

111 ... 

112 

113 

114@overload 

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

116 ... 

117 

118 

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

120 """ 

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

122 

123 :param value: datetime 

124 :param timezone: timezone 

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

126 """ 

127 if timezone is None: 

128 from airflow.settings import TIMEZONE 

129 

130 timezone = TIMEZONE 

131 

132 if not value: 

133 return None 

134 

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

136 if is_localized(value): 

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

138 if hasattr(value, "fold"): 

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

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

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

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

143 value = value.replace(fold=1) 

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

145 if localized is not None: 

146 # This method is available for pytz time zones 

147 return localized(value) 

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

149 if convert is not None: 

150 # For pendulum 

151 return convert(value) 

152 # This may be wrong around DST changes! 

153 return value.replace(tzinfo=timezone) 

154 

155 

156def make_naive(value, timezone=None): 

157 """ 

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

159 

160 :param value: datetime 

161 :param timezone: timezone 

162 :return: naive datetime 

163 """ 

164 if timezone is None: 

165 from airflow.settings import TIMEZONE 

166 

167 timezone = TIMEZONE 

168 

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

170 if is_naive(value): 

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

172 

173 date = value.astimezone(timezone) 

174 

175 # cross library compatibility 

176 naive = dt.datetime( 

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

178 ) 

179 

180 return naive 

181 

182 

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

184 """ 

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

186 

187 :return: datetime.datetime 

188 """ 

189 if "tzinfo" not in kwargs: 

190 from airflow.settings import TIMEZONE 

191 

192 kwargs["tzinfo"] = TIMEZONE 

193 

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

195 

196 

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

198 """ 

199 Parse a time string and return an aware datetime. 

200 

201 :param string: time string 

202 :param timezone: the timezone 

203 """ 

204 from airflow.settings import 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