Add support for ZoneInfo and generic UTC

airflow/serialization/serializers/timezone.py

@@ -17,17 +17,32 @@
 # under the License.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import datetime
+import sys
+from typing import TYPE_CHECKING, cast
 
 from airflow.utils.module_loading import qualname
 
+PY39 = sys.version_info >= (3, 9)
+
 if TYPE_CHECKING:
     from pendulum.tz.timezone import Timezone
 
     from airflow.serialization.serde import U
 
 
-serializers = ["pendulum.tz.timezone.FixedTimezone", "pendulum.tz.timezone.Timezone"]
+serializers = [
+    "pendulum.tz.timezone.FixedTimezone",
+    "pendulum.tz.timezone.Timezone",
+]
+
+if PY39:
+    serializers.append("zoneinfo.ZoneInfo")
+    from zoneinfo import ZoneInfo
+else:
+    serializers.append("backports.zoneinfo.ZoneInfo")
+    from backports.zoneinfo import ZoneInfo
+
 deserializers = serializers
 
 __version__ = 1
@@ -43,21 +58,26 @@ def serialize(o: object) -> tuple[U, str, int, bool]:
     0 without the special case), but passing 0 into ``pendulum.timezone`` does
     not give us UTC (but ``+00:00``).
     """
-    from pendulum.tz.timezone import FixedTimezone, Timezone
+    from pendulum.tz.timezone import FixedTimezone
 
     name = qualname(o)
+
     if isinstance(o, FixedTimezone):
         if o.offset == 0:
             return "UTC", name, __version__, True
         return o.offset, name, __version__, True
 
-    if isinstance(o, Timezone):
-        return o.name, name, __version__, True
+    tz_name = _get_tzinfo_name(cast(datetime.tzinfo, o))
+    if tz_name is not None:
+        return tz_name, name, __version__, True
+
+    if cast(datetime.tzinfo, o).utcoffset(None) == datetime.timedelta(0):
+        return "UTC", qualname(FixedTimezone), __version__, True
 
     return "", "", 0, False
 
 
-def deserialize(classname: str, version: int, data: object) -> Timezone:
+def deserialize(classname: str, version: int, data: object) -> Timezone | ZoneInfo:
     from pendulum.tz import fixed_timezone, timezone
 
     if not isinstance(data, (str, int)):
@@ -69,4 +89,25 @@ def deserialize(classname: str, version: int, data: object) -> Timezone:
     if isinstance(data, int):
         return fixed_timezone(data)
 
+    if "zoneinfo.ZoneInfo" in classname:  # capturing backports and stdlib
+        return ZoneInfo(data)
+
     return timezone(data)
+
+
+# ported from pendulum.tz.timezone._get_tzinfo_name
+def _get_tzinfo_name(tzinfo: datetime.tzinfo | None) -> str | None:
+    if tzinfo is None:
+        return None
+
+    if hasattr(tzinfo, "key"):
+        # zoneinfo timezone
+        return tzinfo.key
+    elif hasattr(tzinfo, "name"):
+        # Pendulum timezone
+        return tzinfo.name
+    elif hasattr(tzinfo, "zone"):
+        # pytz timezone
+        return tzinfo.zone  # type: ignore[no-any-return]
+
+    return None

setup.cfg

@@ -72,6 +72,7 @@ install_requires =
     argcomplete>=1.10
     asgiref
     attrs>=22.1.0
+    backports.zoneinfo>=0.2.1;python_version<"3.9"
     blinker
     cattrs>=22.1.0
     # Colorlog 6.x merges TTYColoredFormatter into ColoredFormatter, breaking backwards compatibility with 4.x

tests/serialization/serializers/test_serializers.py

@@ -22,11 +22,18 @@
 import numpy as np
 import pendulum.tz
 import pytest
+from dateutil.tz import tzutc
 from pendulum import DateTime
 
+from airflow import PY39
 from airflow.models.param import Param, ParamsDict
 from airflow.serialization.serde import DATA, deserialize, serialize
 
+if PY39:
+    from zoneinfo import ZoneInfo
+else:
+    from backports.zoneinfo import ZoneInfo
+
 
 class TestSerializers:
     def test_datetime(self):
@@ -62,8 +69,17 @@ def test_datetime(self):
         d = deserialize(s)
         assert i.timestamp() == d.timestamp()
 
-    def test_deserialize_datetime_v1(self):
+        i = DateTime(2022, 7, 10, tzinfo=tzutc())
+        s = serialize(i)
+        d = deserialize(s)
+        assert i.timestamp() == d.timestamp()
 
+        i = DateTime(2022, 7, 10, tzinfo=ZoneInfo("Europe/Paris"))
+        s = serialize(i)
+        d = deserialize(s)
+        assert i.timestamp() == d.timestamp()
+
+    def test_deserialize_datetime_v1(self):
         s = {
             "__classname__": "pendulum.datetime.DateTime",
             "__version__": 1,