Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/icalendar/timezone/zoneinfo.py: 77%

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

100 statements  

1"""Use zoneinfo timezones""" 

2 

3from __future__ import annotations 

4 

5import copy 

6import copyreg 

7import functools 

8import threading 

9import zoneinfo 

10from datetime import datetime, tzinfo 

11from io import StringIO 

12from typing import TYPE_CHECKING 

13 

14from dateutil.rrule import rrule, rruleset 

15from dateutil.tz import tzical 

16from dateutil.tz.tz import _tzicalvtz 

17 

18from icalendar.tools import is_date, to_datetime 

19 

20from .provider import TZProvider 

21 

22if TYPE_CHECKING: 

23 from icalendar import prop 

24 from icalendar.cal import Timezone 

25 from icalendar.prop import vDDDTypes 

26 

27 

28class ZONEINFO(TZProvider): 

29 """Provide icalendar with timezones from zoneinfo.""" 

30 

31 name = "zoneinfo" 

32 # Use lazy initialization via properties to avoid import-time failures 

33 # in environments without timezone data (e.g., Pyodide/WebAssembly). 

34 # See https://github.com/collective/icalendar/issues/1073 

35 _utc: zoneinfo.ZoneInfo | None = None 

36 _available_timezones_cache: set | None = None 

37 # Class-level lock is intentional: caches are shared by all instances. 

38 _init_lock = threading.Lock() 

39 

40 @property 

41 def utc(self) -> zoneinfo.ZoneInfo: 

42 """Return the UTC timezone, initializing lazily on first access.""" 

43 if self._utc is None: 

44 with self._init_lock: 

45 # Double-check after acquiring lock 

46 if self._utc is None: 

47 ZONEINFO._utc = zoneinfo.ZoneInfo("UTC") 

48 return self._utc # type: ignore[return-value] 

49 

50 @property 

51 def _available_timezones(self) -> set: 

52 """Return available timezones, initializing lazily on first access.""" 

53 if self._available_timezones_cache is None: 

54 with self._init_lock: 

55 # Double-check after acquiring lock 

56 if self._available_timezones_cache is None: 

57 ZONEINFO._available_timezones_cache = zoneinfo.available_timezones() 

58 return self._available_timezones_cache # type: ignore[return-value] 

59 

60 def localize(self, dt: datetime, tz: zoneinfo.ZoneInfo) -> datetime: 

61 """Localize a datetime to a timezone.""" 

62 return dt.replace(tzinfo=tz) 

63 

64 def localize_utc(self, dt: datetime) -> datetime: 

65 """Return the datetime in UTC.""" 

66 if getattr(dt, "tzinfo", False) and dt.tzinfo is not None: 

67 return dt.astimezone(self.utc) 

68 return self.localize(dt, self.utc) 

69 

70 def timezone(self, name: str) -> tzinfo | None: 

71 """Return a timezone with a name or None if we cannot find it.""" 

72 try: 

73 return zoneinfo.ZoneInfo(name) 

74 except zoneinfo.ZoneInfoNotFoundError: 

75 pass 

76 except ValueError: 

77 # ValueError: ZoneInfo keys may not be absolute paths, got: /Europe/CUSTOM 

78 pass 

79 

80 def knows_timezone_id(self, tzid: str) -> bool: 

81 """Whether the timezone is already cached by the implementation.""" 

82 return tzid in self._available_timezones 

83 

84 def fix_rrule_until(self, rrule: rrule, ical_rrule: prop.vRecur) -> None: 

85 """Make sure the until value works for the rrule generated from the ical_rrule.""" # noqa: E501 

86 if not {"UNTIL", "COUNT"}.intersection(ical_rrule.keys()): 

87 # zoninfo does not know any transition dates after 2038 

88 rrule._until = datetime(2038, 12, 31, tzinfo=self.utc) # noqa: SLF001 

89 

90 def create_timezone(self, tz: Timezone.Timezone) -> tzinfo: 

91 """Create a timezone from the given information.""" 

92 try: 

93 return self._create_timezone(tz) 

94 except ValueError: 

95 # We might have a custom component in there. 

96 # see https://github.com/python/cpython/issues/120217 

97 tz = copy.deepcopy(tz) 

98 for sub in tz.walk(): 

99 for attr in list(sub.keys()): 

100 if attr.lower().startswith("x-"): 

101 sub.pop(attr) 

102 for sub in tz.subcomponents: 

103 start: vDDDTypes = sub.get("DTSTART") 

104 if start and hasattr(start, "dt") and is_date(start.dt): 

105 # ValueError: Unsupported DTSTART param in VTIMEZONE: VALUE=DATE 

106 sub.DTSTART = to_datetime(start.dt) 

107 return self._create_timezone(tz) 

108 

109 def _create_timezone(self, tz: Timezone.Timezone) -> tzinfo: 

110 """Create a timezone and maybe fail""" 

111 file = StringIO(tz.to_ical().decode("UTF-8", "replace")) 

112 return tzical(file).get() 

113 

114 def uses_pytz(self) -> bool: 

115 """Whether we use pytz.""" 

116 return False 

117 

118 def uses_zoneinfo(self) -> bool: 

119 """Whether we use zoneinfo.""" 

120 return True 

121 

122 

123def pickle_tzicalvtz(tzicalvtz: _tzicalvtz): 

124 """Because we use dateutil.tzical, we need to make it pickle-able.""" 

125 return _tzicalvtz, (tzicalvtz._tzid, tzicalvtz._comps) # noqa: SLF001 

126 

127 

128copyreg.pickle(_tzicalvtz, pickle_tzicalvtz) 

129 

130 

131def pickle_rrule_with_cache(self: rrule): 

132 """Make sure we can also pickle rrules that cache. 

133 

134 This is mainly copied from rrule.replace. 

135 """ 

136 new_kwargs = { 

137 "interval": self._interval, 

138 "count": self._count, 

139 "dtstart": self._dtstart, 

140 "freq": self._freq, 

141 "until": self._until, 

142 "wkst": self._wkst, 

143 "cache": self._cache is not None, 

144 } 

145 new_kwargs.update(self._original_rule) 

146 # from https://stackoverflow.com/a/64915638/1320237 

147 return functools.partial(rrule, new_kwargs.pop("freq"), **new_kwargs), () 

148 

149 

150copyreg.pickle(rrule, pickle_rrule_with_cache) 

151 

152 

153def pickle_rruleset_with_cache(rs: rruleset): 

154 """Pickle an rruleset.""" 

155 return unpickle_rruleset_with_cache, ( 

156 rs._rrule, # noqa: SLF001 

157 rs._rdate, # noqa: SLF001 

158 rs._exrule, # noqa: SLF001 

159 rs._exdate, # noqa: SLF001 

160 rs._cache is not None, # noqa: SLF001 

161 ) 

162 

163 

164def unpickle_rruleset_with_cache(rrule, rdate, exrule, exdate, cache): 

165 """unpickling the rruleset.""" 

166 rs = rruleset(cache) 

167 for o in rrule: 

168 rs.rrule(o) 

169 for o in rdate: 

170 rs.rdate(o) 

171 for o in exrule: 

172 rs.exrule(o) 

173 for o in exdate: 

174 rs.exdate(o) 

175 return rs 

176 

177 

178copyreg.pickle(rruleset, pickle_rruleset_with_cache) 

179 

180__all__ = ["ZONEINFO"]