Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/serialization/serializers/datetime.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 

20from typing import TYPE_CHECKING 

21 

22from airflow.serialization.serializers.timezone import ( 

23 deserialize as deserialize_timezone, 

24 serialize as serialize_timezone, 

25) 

26from airflow.utils.module_loading import qualname 

27from airflow.utils.timezone import parse_timezone 

28 

29if TYPE_CHECKING: 

30 import datetime 

31 

32 from airflow.serialization.serde import U 

33 

34__version__ = 2 

35 

36serializers = ["datetime.date", "datetime.datetime", "datetime.timedelta", "pendulum.datetime.DateTime"] 

37deserializers = serializers 

38 

39TIMESTAMP = "timestamp" 

40TIMEZONE = "tz" 

41 

42 

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

44 from datetime import date, datetime, timedelta 

45 

46 if isinstance(o, datetime): 

47 qn = qualname(o) 

48 

49 tz = serialize_timezone(o.tzinfo) if o.tzinfo else None 

50 

51 return {TIMESTAMP: o.timestamp(), TIMEZONE: tz}, qn, __version__, True 

52 

53 if isinstance(o, date): 

54 return o.isoformat(), qualname(o), __version__, True 

55 

56 if isinstance(o, timedelta): 

57 return o.total_seconds(), qualname(o), __version__, True 

58 

59 return "", "", 0, False 

60 

61 

62def deserialize(classname: str, version: int, data: dict | str) -> datetime.date | datetime.timedelta: 

63 import datetime 

64 

65 from pendulum import DateTime 

66 

67 tz: datetime.tzinfo | None = None 

68 if isinstance(data, dict) and TIMEZONE in data: 

69 if version == 1: 

70 # try to deserialize unsupported timezones 

71 timezone_mapping = { 

72 "EDT": parse_timezone(-4 * 3600), 

73 "CDT": parse_timezone(-5 * 3600), 

74 "MDT": parse_timezone(-6 * 3600), 

75 "PDT": parse_timezone(-7 * 3600), 

76 "CEST": parse_timezone("CET"), 

77 } 

78 if data[TIMEZONE] in timezone_mapping: 

79 tz = timezone_mapping[data[TIMEZONE]] 

80 else: 

81 tz = parse_timezone(data[TIMEZONE]) 

82 else: 

83 tz = ( 

84 deserialize_timezone(data[TIMEZONE][1], data[TIMEZONE][2], data[TIMEZONE][0]) 

85 if data[TIMEZONE] 

86 else None 

87 ) 

88 

89 if classname == qualname(datetime.datetime) and isinstance(data, dict): 

90 return datetime.datetime.fromtimestamp(float(data[TIMESTAMP]), tz=tz) 

91 

92 if classname == qualname(DateTime) and isinstance(data, dict): 

93 return DateTime.fromtimestamp(float(data[TIMESTAMP]), tz=tz) 

94 

95 if classname == qualname(datetime.timedelta) and isinstance(data, (str, float)): 

96 return datetime.timedelta(seconds=float(data)) 

97 

98 if classname == qualname(datetime.date) and isinstance(data, str): 

99 return datetime.date.fromisoformat(data) 

100 

101 raise TypeError(f"unknown date/time format {classname}")