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
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
20import datetime as dt
21from importlib import metadata
22from typing import TYPE_CHECKING, overload
24import pendulum
25from dateutil.relativedelta import relativedelta
26from packaging import version
27from pendulum.datetime import DateTime
29if TYPE_CHECKING:
30 from pendulum.tz.timezone import FixedTimezone, Timezone
33_PENDULUM3 = version.parse(metadata.version("pendulum")).major == 3
34# UTC Timezone as a tzinfo instance. Actual value depends on pendulum version:
35# - Timezone("UTC") in pendulum 3
36# - FixedTimezone(0, "UTC") in pendulum 2
37utc = pendulum.UTC
40def is_localized(value: dt.datetime) -> bool:
41 """
42 Determine if a given datetime.datetime is aware.
44 The concept is defined in Python documentation. Assuming the tzinfo is
45 either None or a proper ``datetime.tzinfo`` instance, ``value.utcoffset()``
46 implements the appropriate logic.
48 .. seealso:: http://docs.python.org/library/datetime.html#datetime.tzinfo
49 """
50 return value.utcoffset() is not None
53def is_naive(value):
54 """
55 Determine if a given datetime.datetime is naive.
57 The concept is defined in Python documentation. Assuming the tzinfo is
58 either None or a proper ``datetime.tzinfo`` instance, ``value.utcoffset()``
59 implements the appropriate logic.
61 .. seealso:: http://docs.python.org/library/datetime.html#datetime.tzinfo
62 """
63 return value.utcoffset() is None
66def utcnow() -> dt.datetime:
67 """Get the current date and time in UTC."""
68 return dt.datetime.now(tz=utc)
71@overload
72def convert_to_utc(value: None) -> None: ...
75@overload
76def convert_to_utc(value: dt.datetime) -> DateTime: ...
79def convert_to_utc(value: dt.datetime | None) -> DateTime | None:
80 """
81 Create a datetime with the default timezone added if none is associated.
83 :param value: datetime
84 :return: datetime with tzinfo
85 """
86 if value is None:
87 return value
89 if not is_localized(value):
90 value = pendulum.instance(value, TIMEZONE)
92 return pendulum.instance(value.astimezone(utc))
95@overload
96def make_aware(value: None, timezone: dt.tzinfo | None = None) -> None: ...
99@overload
100def make_aware(value: DateTime, timezone: dt.tzinfo | None = None) -> DateTime: ...
103@overload
104def make_aware(value: dt.datetime, timezone: dt.tzinfo | None = None) -> dt.datetime: ...
107def make_aware(value: dt.datetime | None, timezone: dt.tzinfo | None = None) -> dt.datetime | None:
108 """
109 Make a naive datetime.datetime in a given time zone aware.
111 :param value: datetime
112 :param timezone: timezone
113 :return: localized datetime in settings.TIMEZONE or timezone
114 """
115 if timezone is None:
116 timezone = TIMEZONE
118 if not value:
119 return None
121 # Check that we won't overwrite the timezone of an aware datetime.
122 if is_localized(value):
123 raise ValueError(f"make_aware expects a naive datetime, got {value}")
124 # In case we move clock back we want to schedule the run at the time of the second
125 # instance of the same clock time rather than the first one.
126 # Fold parameter has no impact in other cases, so we can safely set it to 1 here
127 value = value.replace(fold=1)
128 localized = getattr(timezone, "localize", None)
129 if localized is not None:
130 # This method is available for pytz time zones
131 return localized(value)
132 convert = getattr(timezone, "convert", None)
133 if convert is not None:
134 # For pendulum
135 return convert(value)
136 # This may be wrong around DST changes!
137 return value.replace(tzinfo=timezone)
140def make_naive(value, timezone=None):
141 """
142 Make an aware datetime.datetime naive in a given time zone.
144 :param value: datetime
145 :param timezone: timezone
146 :return: naive datetime
147 """
148 if timezone is None:
149 timezone = TIMEZONE
151 # Emulate the behavior of astimezone() on Python < 3.6.
152 if is_naive(value):
153 raise ValueError("make_naive() cannot be applied to a naive datetime")
155 date = value.astimezone(timezone)
157 # cross library compatibility
158 naive = dt.datetime(
159 date.year, date.month, date.day, date.hour, date.minute, date.second, date.microsecond
160 )
162 return naive
165def datetime(*args, **kwargs):
166 """
167 Wrap around datetime.datetime to add settings.TIMEZONE if tzinfo not specified.
169 :return: datetime.datetime
170 """
171 if "tzinfo" not in kwargs:
172 kwargs["tzinfo"] = TIMEZONE
174 return dt.datetime(*args, **kwargs)
177def parse(string: str, timezone=None, *, strict=False) -> DateTime:
178 """
179 Parse a time string and return an aware datetime.
181 :param string: time string
182 :param timezone: the timezone
183 :param strict: if False, it will fall back on the dateutil parser if unable to parse with pendulum
184 """
185 return pendulum.parse(string, tz=timezone or TIMEZONE, strict=strict) # type: ignore
188@overload
189def coerce_datetime(v: None, tz: dt.tzinfo | None = None) -> None: ...
192@overload
193def coerce_datetime(v: DateTime, tz: dt.tzinfo | None = None) -> DateTime: ...
196@overload
197def coerce_datetime(v: dt.datetime, tz: dt.tzinfo | None = None) -> DateTime: ...
200def coerce_datetime(v: dt.datetime | None, tz: dt.tzinfo | None = None) -> DateTime | None:
201 """
202 Convert ``v`` into a timezone-aware ``pendulum.DateTime``.
204 * If ``v`` is *None*, *None* is returned.
205 * If ``v`` is a naive datetime, it is converted to an aware Pendulum DateTime.
206 * If ``v`` is an aware datetime, it is converted to a Pendulum DateTime.
207 Note that ``tz`` is **not** taken into account in this case; the datetime
208 will maintain its original tzinfo!
209 """
210 if v is None:
211 return None
212 if isinstance(v, DateTime):
213 return v if v.tzinfo else make_aware(v, tz)
214 # Only dt.datetime is left here.
215 return pendulum.instance(v if v.tzinfo else make_aware(v, tz))
218def td_format(td_object: None | dt.timedelta | float | int) -> str | None:
219 """
220 Format a timedelta object or float/int into a readable string for time duration.
222 For example timedelta(seconds=3752) would become `1h:2M:32s`.
223 If the time is less than a second, the return will be `<1s`.
224 """
225 if not td_object:
226 return None
227 if isinstance(td_object, dt.timedelta):
228 delta = relativedelta() + td_object
229 else:
230 delta = relativedelta(seconds=int(td_object))
231 # relativedelta for timedelta cannot convert days to months
232 # so calculate months by assuming 30 day months and normalize
233 months, delta.days = divmod(delta.days, 30)
234 delta = delta.normalized() + relativedelta(months=months)
236 def _format_part(key: str) -> str:
237 value = int(getattr(delta, key))
238 if value < 1:
239 return ""
240 # distinguish between month/minute following strftime format
241 # and take first char of each unit, i.e. years='y', days='d'
242 if key == "minutes":
243 key = key.upper()
244 key = key[0]
245 return f"{value}{key}"
247 parts = map(_format_part, ("years", "months", "days", "hours", "minutes", "seconds"))
248 joined = ":".join(part for part in parts if part)
249 if not joined:
250 return "<1s"
251 return joined
254def parse_timezone(name: str | int) -> FixedTimezone | Timezone:
255 """
256 Parse timezone and return one of the pendulum Timezone.
258 Provide the same interface as ``pendulum.timezone(name)``
260 :param name: Either IANA timezone or offset to UTC in seconds.
262 :meta private:
263 """
264 if _PENDULUM3:
265 # This only presented in pendulum 3 and code do not reached into the pendulum 2
266 return pendulum.timezone(name) # type: ignore[operator]
267 # In pendulum 2 this refers to the function, in pendulum 3 refers to the module
268 return pendulum.tz.timezone(name) # type: ignore[operator]
271def local_timezone() -> FixedTimezone | Timezone:
272 """
273 Return local timezone.
275 Provide the same interface as ``pendulum.tz.local_timezone()``
277 :meta private:
278 """
279 return pendulum.tz.local_timezone()
282TIMEZONE: FixedTimezone | Timezone = utc
285def initialize(default_timezone: str) -> None:
286 """
287 Initialize the default timezone for the timezone library.
289 Automatically called by airflow-core and task-sdk during their initialization.
290 """
291 global TIMEZONE
292 if default_timezone == "system":
293 TIMEZONE = local_timezone()
294 else:
295 TIMEZONE = parse_timezone(default_timezone)
298def from_timestamp(timestamp: int | float, tz: str | FixedTimezone | Timezone = utc) -> DateTime:
299 """
300 Parse timestamp and return DateTime in a given time zone.
302 :param timestamp: epoch time in seconds.
303 :param tz: In which timezone should return a resulting object.
304 Could be either one of pendulum timezone, IANA timezone or `local` literal.
306 :meta private:
307 """
308 result = coerce_datetime(dt.datetime.fromtimestamp(timestamp, tz=utc))
309 if tz != utc or tz != "UTC":
310 if isinstance(tz, str) and tz.lower() == "local":
311 tz = local_timezone()
312 result = result.in_timezone(tz)
313 return result