Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/serialization/serializers/timezone.py: 30%

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 

21import sys 

22from typing import TYPE_CHECKING, Any, cast 

23 

24from airflow.utils.module_loading import qualname 

25 

26if TYPE_CHECKING: 

27 from airflow.serialization.serde import U 

28 

29 

30serializers = [ 

31 "pendulum.tz.timezone.FixedTimezone", 

32 "pendulum.tz.timezone.Timezone", 

33] 

34 

35PY39 = sys.version_info >= (3, 9) 

36 

37if PY39: 

38 serializers.append("zoneinfo.ZoneInfo") 

39else: 

40 serializers.append("backports.zoneinfo.ZoneInfo") 

41 

42deserializers = serializers 

43 

44__version__ = 1 

45 

46 

47def serialize(o: object) -> tuple[U, str, int, bool]: 

48 """Encode a Pendulum Timezone for serialization. 

49 

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 

58 

59 name = qualname(o) 

60 

61 if isinstance(o, FixedTimezone): 

62 if o.offset == 0: 

63 return "UTC", name, __version__, True 

64 return o.offset, name, __version__, True 

65 

66 tz_name = _get_tzinfo_name(cast(datetime.tzinfo, o)) 

67 if tz_name is not None: 

68 return tz_name, name, __version__, True 

69 

70 if cast(datetime.tzinfo, o).utcoffset(None) == datetime.timedelta(0): 

71 return "UTC", qualname(FixedTimezone), __version__, True 

72 

73 return "", "", 0, False 

74 

75 

76def deserialize(classname: str, version: int, data: object) -> Any: 

77 from airflow.utils.timezone import parse_timezone 

78 

79 if not isinstance(data, (str, int)): 

80 raise TypeError(f"{data} is not of type int or str but of {type(data)}") 

81 

82 if version > __version__: 

83 raise TypeError(f"serialized {version} of {classname} > {__version__}") 

84 

85 if "zoneinfo.ZoneInfo" in classname: 

86 try: 

87 from zoneinfo import ZoneInfo 

88 except ImportError: 

89 from backports.zoneinfo import ZoneInfo 

90 

91 return ZoneInfo(data) 

92 

93 return parse_timezone(data) 

94 

95 

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 

100 

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] 

110 

111 return None