Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/serialization/serializers/timezone.py: 30%
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
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
21import sys
22from typing import TYPE_CHECKING, Any, cast
24from airflow.utils.module_loading import qualname
26if TYPE_CHECKING:
27 from airflow.serialization.serde import U
30serializers = [
31 "pendulum.tz.timezone.FixedTimezone",
32 "pendulum.tz.timezone.Timezone",
33]
35PY39 = sys.version_info >= (3, 9)
37if PY39:
38 serializers.append("zoneinfo.ZoneInfo")
39else:
40 serializers.append("backports.zoneinfo.ZoneInfo")
42deserializers = serializers
44__version__ = 1
47def serialize(o: object) -> tuple[U, str, int, bool]:
48 """Encode a Pendulum Timezone for serialization.
50 Airflow only supports timezone objects that implements Pendulum's Timezone
51 interface. We try to keep as much information as possible to make conversion
52 round-tripping possible (see ``decode_timezone``). We need to special-case
53 UTC; Pendulum implements it as a FixedTimezone (i.e. it gets encoded as
54 0 without the special case), but passing 0 into ``pendulum.timezone`` does
55 not give us UTC (but ``+00:00``).
56 """
57 from pendulum.tz.timezone import FixedTimezone
59 name = qualname(o)
61 if isinstance(o, FixedTimezone):
62 if o.offset == 0:
63 return "UTC", name, __version__, True
64 return o.offset, name, __version__, True
66 tz_name = _get_tzinfo_name(cast(datetime.tzinfo, o))
67 if tz_name is not None:
68 return tz_name, name, __version__, True
70 if cast(datetime.tzinfo, o).utcoffset(None) == datetime.timedelta(0):
71 return "UTC", qualname(FixedTimezone), __version__, True
73 return "", "", 0, False
76def deserialize(classname: str, version: int, data: object) -> Any:
77 from airflow.utils.timezone import parse_timezone
79 if not isinstance(data, (str, int)):
80 raise TypeError(f"{data} is not of type int or str but of {type(data)}")
82 if version > __version__:
83 raise TypeError(f"serialized {version} of {classname} > {__version__}")
85 if "zoneinfo.ZoneInfo" in classname:
86 try:
87 from zoneinfo import ZoneInfo
88 except ImportError:
89 from backports.zoneinfo import ZoneInfo
91 return ZoneInfo(data)
93 return parse_timezone(data)
96# ported from pendulum.tz.timezone._get_tzinfo_name
97def _get_tzinfo_name(tzinfo: datetime.tzinfo | None) -> str | None:
98 if tzinfo is None:
99 return None
101 if hasattr(tzinfo, "key"):
102 # zoneinfo timezone
103 return tzinfo.key
104 elif hasattr(tzinfo, "name"):
105 # Pendulum timezone
106 return tzinfo.name
107 elif hasattr(tzinfo, "zone"):
108 # pytz timezone
109 return tzinfo.zone # type: ignore[no-any-return]
111 return None