Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/build/lib/airflow/serialization/serializers/timezone.py: 41%

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 

22import pendulum 

23from pendulum.tz.timezone import FixedTimezone, Timezone 

24 

25from airflow.utils.module_loading import qualname 

26 

27if TYPE_CHECKING: 

28 from airflow.serialization.serde import U 

29 

30 

31serializers = [FixedTimezone, Timezone] 

32deserializers = serializers 

33 

34__version__ = 1 

35 

36 

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

38 """Encode a Pendulum Timezone for serialization. 

39 

40 Airflow only supports timezone objects that implements Pendulum's Timezone 

41 interface. We try to keep as much information as possible to make conversion 

42 round-tripping possible (see ``decode_timezone``). We need to special-case 

43 UTC; Pendulum implements it as a FixedTimezone (i.e. it gets encoded as 

44 0 without the special case), but passing 0 into ``pendulum.timezone`` does 

45 not give us UTC (but ``+00:00``). 

46 """ 

47 name = qualname(o) 

48 if isinstance(o, FixedTimezone): 

49 if o.offset == 0: 

50 return "UTC", name, __version__, True 

51 return o.offset, name, __version__, True 

52 

53 if isinstance(o, Timezone): 

54 return o.name, name, __version__, True 

55 

56 return "", "", 0, False 

57 

58 

59def deserialize(classname: str, version: int, data: object) -> Timezone: 

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

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

62 

63 if version > __version__: 

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

65 

66 if isinstance(data, int): 

67 return pendulum.tz.fixed_timezone(data) 

68 

69 return pendulum.tz.timezone(data)