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

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.utils.module_loading import qualname 

23 

24if TYPE_CHECKING: 

25 from pendulum.tz.timezone import Timezone 

26 

27 from airflow.serialization.serde import U 

28 

29 

30serializers = ["pendulum.tz.timezone.FixedTimezone", "pendulum.tz.timezone.Timezone"] 

31deserializers = serializers 

32 

33__version__ = 1 

34 

35 

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

37 """Encode a Pendulum Timezone for serialization. 

38 

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

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

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

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

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

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

45 """ 

46 from pendulum.tz.timezone import FixedTimezone, Timezone 

47 

48 name = qualname(o) 

49 if isinstance(o, FixedTimezone): 

50 if o.offset == 0: 

51 return "UTC", name, __version__, True 

52 return o.offset, name, __version__, True 

53 

54 if isinstance(o, Timezone): 

55 return o.name, name, __version__, True 

56 

57 return "", "", 0, False 

58 

59 

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

61 from pendulum.tz import fixed_timezone, timezone 

62 

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

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

65 

66 if version > __version__: 

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

68 

69 if isinstance(data, int): 

70 return fixed_timezone(data) 

71 

72 return timezone(data)