Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/icalendar/cal/lazy.py: 38%

1"""Components for lazy parsing of components.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Literal 

6 

7from icalendar.cal.component_factory import ComponentFactory 

8from icalendar.parser.ical.lazy import LazyCalendarIcalParser 

9 

10from .calendar import Calendar 

11 

12if TYPE_CHECKING: 

13 from collections.abc import Callable 

14 

15 from icalendar.parser.ical.component import ComponentIcalParser 

16 from icalendar.parser.ical.lazy import LazySubcomponent 

17 

18 from .component import Component 

19 

20 

21class ParsedSubcomponentsStrategy: 

22 """All the subcomponents are parsed and available as a list.""" 

23 

24 def __init__(self): 

25 self._components: list[Component] = [] 

26 

27 def get_all_components(self) -> tuple[ParsedSubcomponentsStrategy, list[Component]]: 

28 """Get the subcomponents of the calendar.""" 

29 return self, self._components 

30 

31 def set_components( 

32 self, components: list[Component] 

33 ) -> ParsedSubcomponentsStrategy: 

34 """Set the subcomponents of the calendar.""" 

35 self._components = components 

36 return self 

37 

38 def add_component(self, component: Component) -> ParsedSubcomponentsStrategy: 

39 """Add a component to the calendar.""" 

40 self._components.append(component.parse()) 

41 return self 

42 

43 def is_lazy(self) -> Literal[False]: 

44 """Returns ``False`` because subcomponents are not lazily parsed.""" 

45 return False 

46 

47 def walk(self, name: str) -> tuple[ParsedSubcomponentsStrategy, list[Component]]: 

48 """Get the subcomponents of the calendar with the given name.""" 

49 result = [] 

50 for component in self._components: 

51 result += component.walk(name) 

52 return self, result 

53 

54 def with_uid( 

55 self, name: str 

56 ) -> tuple[ParsedSubcomponentsStrategy, list[Component]]: 

57 """Get the subcomponents of the calendar with the given uid.""" 

58 result = [] 

59 for component in self._components: 

60 result += component.with_uid(name) 

61 return self, result 

62 

63 

64class LazySubcomponentsStrategy: 

65 """Parse subcomponents only when accessed.""" 

66 

67 initial_components_to_parse: tuple[str, ...] = ("VTIMEZONE",) 

68 """Parse these subcomponents before any others.""" 

69 

70 def __init__(self): 

71 self._components: list[LazySubcomponent | Component] = [] 

72 self._initial_parsed: bool = False 

73 

74 @property 

75 def as_parsed(self) -> ParsedSubcomponentsStrategy: 

76 """Return the parsed components.""" 

77 return ParsedSubcomponentsStrategy().set_components( 

78 [component.parse() for component in self._components] 

79 ) 

80 

81 def get_all_components(self) -> tuple[ParsedSubcomponentsStrategy, list[Component]]: 

82 """Get the subcomponents of the calendar. 

83 

84 Parse all subcomponents. 

85 """ 

86 self.parse_initial_components() 

87 return self.as_parsed.get_all_components() 

88 

89 def set_components( 

90 self, components: list[Component] 

91 ) -> ParsedSubcomponentsStrategy: 

92 """Set the subcomponents of the calendar.""" 

93 return ParsedSubcomponentsStrategy().set_components(components) 

94 

95 def add_component( 

96 self, component: Component | LazySubcomponent 

97 ) -> LazySubcomponentsStrategy: 

98 """Add a component to the calendar.""" 

99 self._components.append(component) 

100 return self 

101 

102 def is_lazy(self) -> bool: 

103 """Return whether the subcomponents may be lazily parsed.""" 

104 return True 

105 

106 def parse_initial_components(self) -> None: 

107 """Parse the components that are required by other components. 

108 

109 This mainly concerns the timezone components. 

110 They are required by other components that have a TZID parameter. 

111 """ 

112 if self._initial_parsed: 

113 return 

114 self._initial_parsed = True 

115 for component in self._components: 

116 if component.name in self.initial_components_to_parse: 

117 component.parse() 

118 

119 def walk( 

120 self, name: str | None 

121 ) -> tuple[LazySubcomponentsStrategy, list[Component]]: 

122 """Get the subcomponents of the calendar with the given name. 

123 

124 Parse only the minimal number of subcomponents. 

125 """ 

126 if name is None: 

127 return self.as_parsed.walk(name) 

128 self.parse_initial_components() 

129 result = [] 

130 for component in self._components: 

131 result += component.walk(name) 

132 return self, result 

133 

134 def with_uid(self, uid: str) -> tuple[LazySubcomponentsStrategy, list[Component]]: 

135 """Get the subcomponents of the calendar with the given ``uid``. 

136 

137 Parse only the minimal number of subcomponents. 

138 """ 

139 self.parse_initial_components() 

140 result = [] 

141 for component in self._components: 

142 result += component.with_uid(uid) 

143 return self, result 

144 

145 

146class InitialSubcomponentsStrategy: 

147 """Initial strategy for the calendar. 

148 

149 No subcomponents. 

150 """ 

151 

152 def set_components(self, components: list[Component]) -> LazySubcomponentsStrategy: 

153 if components: 

154 raise ValueError( 

155 "Cannot set subcomponents on an uninitialised LazyCalendar. " 

156 "Parse it first or add components via add_component()." 

157 ) 

158 return LazySubcomponentsStrategy() 

159 

160 

161class LazyCalendar(Calendar): 

162 """A calendar that can handle big files. 

163 

164 Subcomponents of this calendar are evaluated lazily, 

165 meaning that they are not parsed until they are accessed. 

166 This allows the calendar to handle large files without 

167 consuming too much memory or time. 

168 

169 All properties of the calendar component are parsed immediately. 

170 Subcomponents and their properties are parsed lazily. 

171 

172 Examples: 

173 

174 By accessing the :attr:`~icalendar.cal.calendar.Calendar.events` of the calendar, 

175 only :class:`~icalendar.cal.event.Event` and 

176 :class:`~icalendar.cal.timezone.Timezone` are immediately parsed. 

177 

178 .. code-block:: pycon 

179 

180 >>> from icalendar import LazyCalendar 

181 >>> calendar = LazyCalendar.example("issue_1050_all_components") 

182 >>> len(calendar.events) == 1 

183 True 

184 

185 The calendar's subcomponents were not parsed because they were not accessed. 

186 The calendar is still lazy. 

187 

188 >>> calendar.is_lazy() 

189 True 

190 

191 When you access all :attr:`subcomponents` of the calendar, 

192 for example by getting their count, the entire calendar is 

193 parsed and becomes not lazy. 

194 

195 >>> len(calendar.subcomponents) 

196 5 

197 >>> calendar.is_lazy() 

198 False 

199 

200 """ 

201 

202 _subcomponents: ( 

203 LazySubcomponentsStrategy 

204 | ParsedSubcomponentsStrategy 

205 | InitialSubcomponentsStrategy 

206 ) 

207 """The strategy pattern for subcomponents of the calendar.""" 

208 

209 def __init__(self, *args, **kwargs): 

210 """Initialize the calendar.""" 

211 self._subcomponents = InitialSubcomponentsStrategy() 

212 super().__init__(*args, **kwargs) 

213 

214 @property 

215 def subcomponents(self) -> list[Component]: 

216 """The subcomponents of the calendar. 

217 

218 Parse all subcomponents of the calendar and return them as a list. 

219 

220 You can manipulate this list or set it. 

221 It has the same behavior as in :class:`~icalendar.cal.calendar.Calendar`. 

222 """ 

223 self._subcomponents, result = self._subcomponents.get_all_components() 

224 return result 

225 

226 @subcomponents.setter 

227 def subcomponents(self, value: list[Component]) -> None: 

228 """Set the subcomponents of the calendar.""" 

229 self._subcomponents = self._subcomponents.set_components(value) 

230 

231 @classmethod 

232 def _get_ical_parser(cls, st: str | bytes) -> ComponentIcalParser: 

233 """Get the iCal parser for the given input string.""" 

234 return LazyCalendarIcalParser( 

235 st, cls._get_component_factory(), cls.types_factory 

236 ) 

237 

238 @classmethod 

239 def _get_component_factory(cls) -> ComponentFactory: 

240 """Get the component factory for this calendar.""" 

241 factory = ComponentFactory() 

242 factory.add_component_class(cls) 

243 return factory 

244 

245 def add_component(self, component: Component) -> None: 

246 """Add a component to the calendar. 

247 

248 Use this instead of appending to 

249 :attr:`~icalendar.cal.lazy.LazyCalendar.subcomponents`, 

250 as the latter does not parse the whole calendar. 

251 """ 

252 self._subcomponents = self._subcomponents.add_component(component) 

253 

254 def is_lazy(self) -> bool: 

255 """Whether the subcomponents will be parsed lazily. 

256 

257 .. note:: If you believe that the calendar parses more than it should, 

258 please `open an issue <https://github.com/collective/icalendar/issues/new?template=bug_report.md>`_. 

259 

260 Returns: 

261 ``True`` if subcomponents are deferred and not yet parsed. 

262 ``False`` if all subcomponents have been parsed. 

263 """ 

264 return self._subcomponents.is_lazy() 

265 

266 def _walk( 

267 self, name: str | None, select: Callable[[Component], bool] 

268 ) -> list[Component]: 

269 self._subcomponents, result = self._subcomponents.walk(name) 

270 result = [component for component in result if select(component)] 

271 if (name is None or self.name == name) and select(self): 

272 result.insert(0, self) 

273 return result 

274 

275 def with_uid(self, uid: str) -> list[Component]: 

276 self._subcomponents, result = self._subcomponents.with_uid(uid) 

277 if self.uid == uid: 

278 result.insert(0, self) 

279 return result 

280 

281 with_uid.__doc__ = Calendar.with_uid.__doc__ 

282 

283 

284__all__ = ["LazyCalendar"] 

285 

286if __name__ == "__main__": 

287 import timeit 

288 

289 calendar = Calendar.example("issue_1050_all_components") 

290 COUNT = 10000 

291 calendar.subcomponents *= COUNT 

292 ics = calendar.to_ical() 

293 

294 def _benchmark(cal: type[Calendar]): 

295 """Check out how fast this is.""" 

296 cal = cal.from_ical(ics) 

297 assert len(cal.events) == COUNT 

298 

299 for cal in [Calendar, LazyCalendar]: 

300 print("Benchmarking:", cal.__name__) # noqa: T201 

301 print(timeit.timeit("_benchmark(cal)", globals=locals(), number=1)) # noqa: T201 

302 

303 # Benchmarking: Calendar 

304 # 12.277852076000272 

305 # Benchmarking: LazyCalendar 

306 # 5.738950790999297