Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/serialization/serializers/datetime.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
20from typing import TYPE_CHECKING
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
29if TYPE_CHECKING:
30 import datetime
32 from airflow.serialization.serde import U
34__version__ = 2
36serializers = ["datetime.date", "datetime.datetime", "datetime.timedelta", "pendulum.datetime.DateTime"]
37deserializers = serializers
39TIMESTAMP = "timestamp"
40TIMEZONE = "tz"
43def serialize(o: object) -> tuple[U, str, int, bool]:
44 from datetime import date, datetime, timedelta
46 if isinstance(o, datetime):
47 qn = qualname(o)
49 tz = serialize_timezone(o.tzinfo) if o.tzinfo else None
51 return {TIMESTAMP: o.timestamp(), TIMEZONE: tz}, qn, __version__, True
53 if isinstance(o, date):
54 return o.isoformat(), qualname(o), __version__, True
56 if isinstance(o, timedelta):
57 return o.total_seconds(), qualname(o), __version__, True
59 return "", "", 0, False
62def deserialize(classname: str, version: int, data: dict | str) -> datetime.date | datetime.timedelta:
63 import datetime
65 from pendulum import DateTime
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 )
89 if classname == qualname(datetime.datetime) and isinstance(data, dict):
90 return datetime.datetime.fromtimestamp(float(data[TIMESTAMP]), tz=tz)
92 if classname == qualname(DateTime) and isinstance(data, dict):
93 return DateTime.fromtimestamp(float(data[TIMESTAMP]), tz=tz)
95 if classname == qualname(datetime.timedelta) and isinstance(data, (str, float)):
96 return datetime.timedelta(seconds=float(data))
98 if classname == qualname(datetime.date) and isinstance(data, str):
99 return datetime.date.fromisoformat(data)
101 raise TypeError(f"unknown date/time format {classname}")