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"]