Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/build/lib/airflow/utils/dates.py: 17%

102 statements  

« prev     ^ index     » next       coverage.py v7.0.1, created at 2022-12-25 06:11 +0000

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 warnings 

21from datetime import datetime, timedelta 

22 

23from croniter import croniter 

24from dateutil.relativedelta import relativedelta # for doctest 

25 

26from airflow.exceptions import RemovedInAirflow3Warning 

27from airflow.utils import timezone 

28 

29cron_presets: dict[str, str] = { 

30 "@hourly": "0 * * * *", 

31 "@daily": "0 0 * * *", 

32 "@weekly": "0 0 * * 0", 

33 "@monthly": "0 0 1 * *", 

34 "@quarterly": "0 0 1 */3 *", 

35 "@yearly": "0 0 1 1 *", 

36} 

37 

38 

39def date_range( 

40 start_date: datetime, 

41 end_date: datetime | None = None, 

42 num: int | None = None, 

43 delta: str | timedelta | relativedelta | None = None, 

44) -> list[datetime]: 

45 """ 

46 Get a set of dates as a list based on a start, end and delta, delta 

47 can be something that can be added to `datetime.datetime` 

48 or a cron expression as a `str` 

49 

50 .. code-block:: pycon 

51 >>> from airflow.utils.dates import date_range 

52 >>> from datetime import datetime, timedelta 

53 >>> date_range(datetime(2016, 1, 1), datetime(2016, 1, 3), delta=timedelta(1)) 

54 [datetime.datetime(2016, 1, 1, 0, 0, tzinfo=Timezone('UTC')), 

55 datetime.datetime(2016, 1, 2, 0, 0, tzinfo=Timezone('UTC')), 

56 datetime.datetime(2016, 1, 3, 0, 0, tzinfo=Timezone('UTC'))] 

57 >>> date_range(datetime(2016, 1, 1), datetime(2016, 1, 3), delta="0 0 * * *") 

58 [datetime.datetime(2016, 1, 1, 0, 0, tzinfo=Timezone('UTC')), 

59 datetime.datetime(2016, 1, 2, 0, 0, tzinfo=Timezone('UTC')), 

60 datetime.datetime(2016, 1, 3, 0, 0, tzinfo=Timezone('UTC'))] 

61 >>> date_range(datetime(2016, 1, 1), datetime(2016, 3, 3), delta="0 0 0 * *") 

62 [datetime.datetime(2016, 1, 1, 0, 0, tzinfo=Timezone('UTC')), 

63 datetime.datetime(2016, 2, 1, 0, 0, tzinfo=Timezone('UTC')), 

64 datetime.datetime(2016, 3, 1, 0, 0, tzinfo=Timezone('UTC'))] 

65 

66 :param start_date: anchor date to start the series from 

67 :param end_date: right boundary for the date range 

68 :param num: alternatively to end_date, you can specify the number of 

69 number of entries you want in the range. This number can be negative, 

70 output will always be sorted regardless 

71 :param delta: step length. It can be datetime.timedelta or cron expression as string 

72 """ 

73 warnings.warn( 

74 "`airflow.utils.dates.date_range()` is deprecated. Please use `airflow.timetables`.", 

75 category=RemovedInAirflow3Warning, 

76 stacklevel=2, 

77 ) 

78 

79 if not delta: 

80 return [] 

81 if end_date: 

82 if start_date > end_date: 

83 raise Exception("Wait. start_date needs to be before end_date") 

84 if num: 

85 raise Exception("Wait. Either specify end_date OR num") 

86 if not end_date and not num: 

87 end_date = timezone.utcnow() 

88 

89 delta_iscron = False 

90 time_zone = start_date.tzinfo 

91 

92 abs_delta: timedelta | relativedelta 

93 if isinstance(delta, str): 

94 delta_iscron = True 

95 if timezone.is_localized(start_date): 

96 start_date = timezone.make_naive(start_date, time_zone) 

97 cron = croniter(cron_presets.get(delta, delta), start_date) 

98 elif isinstance(delta, timedelta): 

99 abs_delta = abs(delta) 

100 elif isinstance(delta, relativedelta): 

101 abs_delta = abs(delta) 

102 else: 

103 raise Exception("Wait. delta must be either datetime.timedelta or cron expression as str") 

104 

105 dates = [] 

106 if end_date: 

107 if timezone.is_naive(start_date) and not timezone.is_naive(end_date): 

108 end_date = timezone.make_naive(end_date, time_zone) 

109 while start_date <= end_date: # type: ignore 

110 if timezone.is_naive(start_date): 

111 dates.append(timezone.make_aware(start_date, time_zone)) 

112 else: 

113 dates.append(start_date) 

114 

115 if delta_iscron: 

116 start_date = cron.get_next(datetime) 

117 else: 

118 start_date += abs_delta 

119 else: 

120 num_entries: int = num # type: ignore 

121 for _ in range(abs(num_entries)): 

122 if timezone.is_naive(start_date): 

123 dates.append(timezone.make_aware(start_date, time_zone)) 

124 else: 

125 dates.append(start_date) 

126 

127 if delta_iscron and num_entries > 0: 

128 start_date = cron.get_next(datetime) 

129 elif delta_iscron: 

130 start_date = cron.get_prev(datetime) 

131 elif num_entries > 0: 

132 start_date += abs_delta 

133 else: 

134 start_date -= abs_delta 

135 

136 return sorted(dates) 

137 

138 

139def round_time(dt, delta, start_date=timezone.make_aware(datetime.min)): 

140 """ 

141 Returns the datetime of the form start_date + i * delta 

142 which is closest to dt for any non-negative integer i. 

143 Note that delta may be a datetime.timedelta or a dateutil.relativedelta 

144 

145 .. code-block:: pycon 

146 

147 >>> round_time(datetime(2015, 1, 1, 6), timedelta(days=1)) 

148 datetime.datetime(2015, 1, 1, 0, 0) 

149 >>> round_time(datetime(2015, 1, 2), relativedelta(months=1)) 

150 datetime.datetime(2015, 1, 1, 0, 0) 

151 >>> round_time(datetime(2015, 9, 16, 0, 0), timedelta(1), datetime(2015, 9, 14, 0, 0)) 

152 datetime.datetime(2015, 9, 16, 0, 0) 

153 >>> round_time(datetime(2015, 9, 15, 0, 0), timedelta(1), datetime(2015, 9, 14, 0, 0)) 

154 datetime.datetime(2015, 9, 15, 0, 0) 

155 >>> round_time(datetime(2015, 9, 14, 0, 0), timedelta(1), datetime(2015, 9, 14, 0, 0)) 

156 datetime.datetime(2015, 9, 14, 0, 0) 

157 >>> round_time(datetime(2015, 9, 13, 0, 0), timedelta(1), datetime(2015, 9, 14, 0, 0)) 

158 datetime.datetime(2015, 9, 14, 0, 0) 

159 """ 

160 if isinstance(delta, str): 

161 # It's cron based, so it's easy 

162 time_zone = start_date.tzinfo 

163 start_date = timezone.make_naive(start_date, time_zone) 

164 cron = croniter(delta, start_date) 

165 prev = cron.get_prev(datetime) 

166 if prev == start_date: 

167 return timezone.make_aware(start_date, time_zone) 

168 else: 

169 return timezone.make_aware(prev, time_zone) 

170 

171 # Ignore the microseconds of dt 

172 dt -= timedelta(microseconds=dt.microsecond) 

173 

174 # We are looking for a datetime in the form start_date + i * delta 

175 # which is as close as possible to dt. Since delta could be a relative 

176 # delta we don't know its exact length in seconds so we cannot rely on 

177 # division to find i. Instead we employ a binary search algorithm, first 

178 # finding an upper and lower limit and then dissecting the interval until 

179 # we have found the closest match. 

180 

181 # We first search an upper limit for i for which start_date + upper * delta 

182 # exceeds dt. 

183 upper = 1 

184 while start_date + upper * delta < dt: 

185 # To speed up finding an upper limit we grow this exponentially by a 

186 # factor of 2 

187 upper *= 2 

188 

189 # Since upper is the first value for which start_date + upper * delta 

190 # exceeds dt, upper // 2 is below dt and therefore forms a lower limited 

191 # for the i we are looking for 

192 lower = upper // 2 

193 

194 # We now continue to intersect the interval between 

195 # start_date + lower * delta and start_date + upper * delta 

196 # until we find the closest value 

197 while True: 

198 # Invariant: start + lower * delta < dt <= start + upper * delta 

199 # If start_date + (lower + 1)*delta exceeds dt, then either lower or 

200 # lower+1 has to be the solution we are searching for 

201 if start_date + (lower + 1) * delta >= dt: 

202 # Check if start_date + (lower + 1)*delta or 

203 # start_date + lower*delta is closer to dt and return the solution 

204 if (start_date + (lower + 1) * delta) - dt <= dt - (start_date + lower * delta): 

205 return start_date + (lower + 1) * delta 

206 else: 

207 return start_date + lower * delta 

208 

209 # We intersect the interval and either replace the lower or upper 

210 # limit with the candidate 

211 candidate = lower + (upper - lower) // 2 

212 if start_date + candidate * delta >= dt: 

213 upper = candidate 

214 else: 

215 lower = candidate 

216 

217 # in the special case when start_date > dt the search for upper will 

218 # immediately stop for upper == 1 which results in lower = upper // 2 = 0 

219 # and this function returns start_date. 

220 

221 

222def infer_time_unit(time_seconds_arr): 

223 """ 

224 Determine the most appropriate time unit for an array of time durations 

225 specified in seconds. 

226 e.g. 5400 seconds => 'minutes', 36000 seconds => 'hours' 

227 """ 

228 if len(time_seconds_arr) == 0: 

229 return "hours" 

230 max_time_seconds = max(time_seconds_arr) 

231 if max_time_seconds <= 60 * 2: 

232 return "seconds" 

233 elif max_time_seconds <= 60 * 60 * 2: 

234 return "minutes" 

235 elif max_time_seconds <= 24 * 60 * 60 * 2: 

236 return "hours" 

237 else: 

238 return "days" 

239 

240 

241def scale_time_units(time_seconds_arr, unit): 

242 """Convert an array of time durations in seconds to the specified time unit.""" 

243 if unit == "minutes": 

244 return list(map(lambda x: x / 60, time_seconds_arr)) 

245 elif unit == "hours": 

246 return list(map(lambda x: x / (60 * 60), time_seconds_arr)) 

247 elif unit == "days": 

248 return list(map(lambda x: x / (24 * 60 * 60), time_seconds_arr)) 

249 return time_seconds_arr 

250 

251 

252def days_ago(n, hour=0, minute=0, second=0, microsecond=0): 

253 """ 

254 Get a datetime object representing `n` days ago. By default the time is 

255 set to midnight. 

256 """ 

257 warnings.warn( 

258 "Function `days_ago` is deprecated and will be removed in Airflow 3.0. " 

259 "You can achieve equivalent behavior with `pendulum.today('UTC').add(days=-N, ...)`", 

260 RemovedInAirflow3Warning, 

261 stacklevel=2, 

262 ) 

263 

264 today = timezone.utcnow().replace(hour=hour, minute=minute, second=second, microsecond=microsecond) 

265 return today - timedelta(days=n) 

266 

267 

268def parse_execution_date(execution_date_str): 

269 """Parse execution date string to datetime object.""" 

270 return timezone.parse(execution_date_str)