ical.store

  1"""Library for managing the lifecycle of components in a calendar.
  2
  3A store is like a manager for events within a Calendar, updating the necessary
  4properties such as modification times, sequence numbers, and ids. This higher
  5level API is a more convenient API than working with the lower level objects
  6directly.
  7"""
  8
  9# pylint: disable=unnecessary-lambda
 10
 11from __future__ import annotations
 12
 13import datetime
 14import logging
 15from collections.abc import Callable, Iterable, Generator
 16from typing import Any, TypeVar, Generic, cast
 17
 18from .calendar import Calendar
 19from .component import validate_recurrence_dates
 20from .event import Event
 21from .exceptions import StoreError, TodoStoreError, EventStoreError
 22from .iter import RulesetIterable
 23from .list import todo_list_view
 24from .timezone import Timezone
 25from .todo import Todo
 26from .types import Range, Recur, RecurrenceId, RelationshipType
 27from .tzif.timezoneinfo import TimezoneInfoError
 28from .util import dtstamp_factory, local_timezone
 29
 30
 31_LOGGER = logging.getLogger(__name__)
 32
 33
 34__all__ = [
 35    "EventStore",
 36    "EventStoreError",
 37    "TodoStore",
 38    "TodoStoreError",
 39    "StoreError",
 40]
 41
 42_T = TypeVar("_T", bound="Event | Todo")
 43# We won't be able to edit dates more than 100 years in the future, but this
 44# should be sufficient for most use cases.
 45_MAX_SCAN_DATE = datetime.date.today() + datetime.timedelta(days=365 * 100)
 46
 47
 48def _ensure_timezone(
 49    dtvalue: datetime.datetime | datetime.date | None, timezones: list[Timezone]
 50) -> Timezone | None:
 51    """Create a timezone object for the specified date if it does not already exist."""
 52    if (
 53        not isinstance(dtvalue, datetime.datetime)
 54        or not dtvalue.utcoffset()
 55        or not dtvalue.tzinfo
 56    ):
 57        return None
 58
 59    # Verify this timezone does not already exist. The number of timezones
 60    # in a calendar is typically very small so iterate over the whole thing
 61    # to avoid any synchronization/cache issues.
 62    key = str(dtvalue.tzinfo)
 63    for timezone in timezones:
 64        if timezone.tz_id == key:
 65            return None
 66
 67    try:
 68        return Timezone.from_tzif(key)
 69    except TimezoneInfoError as err:
 70        raise EventStoreError(
 71            f"No timezone information available for event: {key}"
 72        ) from err
 73
 74
 75def _match_item(item: _T, uid: str, recurrence_id: str | None) -> bool:
 76    """Return True if the item is an instance of a recurring event."""
 77    if item.uid != uid:
 78        return False
 79    if recurrence_id is None:
 80        # Match all items with the specified uids
 81        return True
 82    # Match a single item with the specified recurrence_id. If the item is an
 83    # edited instance match return it
 84    if item.recurrence_id == recurrence_id:
 85        _LOGGER.debug("Matched exact recurrence_id: %s", item)
 86        return True
 87    # Otherwise, determine if this instance is in the series
 88    _LOGGER.debug(
 89        "Expanding item %s %s to look for match of %s", uid, item.dtstart, recurrence_id
 90    )
 91    dtstart = RecurrenceId.to_value(recurrence_id)
 92    if isinstance(dtstart, datetime.datetime) and isinstance(
 93        item.dtstart, datetime.datetime
 94    ):
 95        # The recurrence_id does not support timezone information, so put it in the
 96        # same timezone as the item to compare.
 97        if item.dtstart.tzinfo is not None:
 98            dtstart = dtstart.replace(tzinfo=item.dtstart.tzinfo)
 99    for dt in item.as_rrule() or ():
100        if isinstance(dt, datetime.datetime):
101            if dt.date() > _MAX_SCAN_DATE:
102                _LOGGER.debug("Aborting scan, date %s is beyond max scan date", dt)
103                break
104        elif dt > _MAX_SCAN_DATE:
105            _LOGGER.debug("Aborting scan, date %s is beyond max scan date", dt)
106            break
107        if dt == dtstart:
108            _LOGGER.debug("Found expanded recurrence_id: %s", dt)
109            return True
110    return False
111
112
113def _match_items(
114    items: list[_T], uid: str, recurrence_id: str | None
115) -> Generator[tuple[int, _T], None, None]:
116    """Return items from the list that match the uid and recurrence_id."""
117    for index, item in enumerate(items):
118        if _match_item(item, uid, recurrence_id):
119            yield index, item
120
121
122def _prepare_update(
123    store_item: Event | Todo,
124    item: Event | Todo,
125    recurrence_id: str | None = None,
126    recurrence_range: Range = Range.NONE,
127) -> dict[str, Any]:
128    """Prepare an update to an existing event."""
129    partial_update = item.dict(
130        exclude_unset=True,
131        exclude={"dtstamp", "uid", "sequence", "created", "last_modified"},
132    )
133    _LOGGER.debug("Preparing update update=%s", item)
134    update = {
135        "created": store_item.dtstamp,
136        "sequence": (store_item.sequence + 1) if store_item.sequence else 1,
137        "last_modified": item.dtstamp,
138        **partial_update,
139        "dtstamp": item.dtstamp,
140    }
141    if rrule := update.get("rrule"):
142        update["rrule"] = Recur.parse_obj(rrule)
143    if recurrence_id and store_item.rrule:
144        # Forking a new event off the old event preserves the original uid and
145        # recurrence_id.
146        update.update(
147            {
148                "uid": store_item.uid,
149                "recurrence_id": recurrence_id,
150            }
151        )
152        if recurrence_range == Range.NONE:
153            # The new event copied from the original is a single instance,
154            # which is not recurring.
155            update["rrule"] = None
156        else:
157            # Overwriting with a new recurring event
158            update["created"] = item.dtstamp
159
160            # Adjust start and end time of the event
161            dtstart: datetime.datetime | datetime.date = RecurrenceId.to_value(
162                recurrence_id
163            )
164            if item.dtstart:
165                dtstart = item.dtstart
166            update["dtstart"] = dtstart
167            # Event either has a duration (which should already be set) or has
168            # an explicit end which needs to be realigned to new start time.
169            if isinstance(store_item, Event) and store_item.dtend:
170                update["dtend"] = dtstart + store_item.computed_duration
171    return update
172
173
174class GenericStore(Generic[_T]):
175    """A a store manages the lifecycle of items on a Calendar."""
176
177    def __init__(
178        self,
179        items: list[_T],
180        timezones: list[Timezone],
181        exc: type[StoreError],
182        dtstamp_fn: Callable[[], datetime.datetime] = lambda: dtstamp_factory(),
183        tzinfo: datetime.tzinfo | None = None,
184    ):
185        """Initialize the EventStore."""
186        self._items = items
187        self._timezones = timezones
188        self._exc = exc
189        self._dtstamp_fn = dtstamp_fn
190        self._tzinfo = tzinfo or local_timezone()
191
192    def add(self, item: _T) -> _T:
193        """Add the specified item to the calendar.
194
195        This will handle assigning modification dates, sequence numbers, etc
196        if those fields are unset.
197
198        The store will ensure the `ical.calendar.Calendar` has the necessary
199        `ical.timezone.Timezone` needed to fully specify the time information
200        when encoded.
201        """
202        update: dict[str, Any] = {}
203        if not item.created:
204            update["created"] = item.dtstamp
205        if item.sequence is None:
206            update["sequence"] = 0
207        if isinstance(item, Todo) and not item.dtstart:
208            if item.due:
209                update["dtstart"] = item.due - datetime.timedelta(days=1)
210            else:
211                update["dtstart"] = datetime.datetime.now(tz=self._tzinfo)
212        new_item = cast(_T, item.copy_and_validate(update=update))
213
214        # The store can only manage cascading deletes for some relationship types
215        for relation in new_item.related_to or ():
216            if relation.reltype != RelationshipType.PARENT:
217                raise self._exc(f"Unsupported relationship type {relation.reltype}")
218
219        _LOGGER.debug("Adding item: %s", new_item)
220        self._ensure_timezone(item.dtstart)
221        if isinstance(item, Event) and item.dtend:
222            self._ensure_timezone(item.dtend)
223        self._items.append(new_item)
224        return new_item
225
226    def delete(
227        self,
228        uid: str,
229        recurrence_id: str | None = None,
230        recurrence_range: Range = Range.NONE,
231    ) -> None:
232        """Delete the item from the calendar.
233
234        This method is used to delete an existing item. For a recurring item
235        either the whole item or instances of an item may be deleted. To
236        delete the complete range of a recurring item, the `uid` property
237        for the item must be specified and the `recurrence_id` should not
238        be specified. To delete an individual instance of the item the
239        `recurrence_id` must be specified.
240
241        When deleting individual instances, the range property may specify
242        if deletion of just a specific instance, or a range of instances.
243        """
244        items_to_delete: list[_T] = [
245            item for _, item in _match_items(self._items, uid, recurrence_id)
246        ]
247        if not items_to_delete:
248            raise self._exc(
249                f"No existing item with uid/recurrence_id: {uid}/{recurrence_id}"
250            )
251
252        for store_item in items_to_delete:
253            self._apply_delete(store_item, recurrence_id, recurrence_range)
254
255    def _apply_delete(
256        self,
257        store_item: _T,
258        recurrence_id: str | None = None,
259        recurrence_range: Range = Range.NONE,
260    ) -> None:
261        if (
262            recurrence_id
263            and recurrence_range == Range.THIS_AND_FUTURE
264            and RecurrenceId.to_value(recurrence_id) == store_item.dtstart
265        ):
266            # Editing the first instance and all forward is the same as editing the
267            # entire series so don't bother forking a new event
268            recurrence_id = None
269
270        children = []
271        for event in self._items:
272            for relation in event.related_to or ():
273                if (
274                    relation.reltype == RelationshipType.PARENT
275                    and relation.uid == store_item.uid
276                ):
277                    children.append(event)
278        for child in children:
279            self._items.remove(child)
280
281        # Deleting all instances in the series
282        if not recurrence_id or not store_item.rrule:
283            self._items.remove(store_item)
284            return
285
286        exdate = RecurrenceId.to_value(recurrence_id)
287        if recurrence_range == Range.NONE:
288            # A single recurrence instance is removed. Add an exclusion to
289            # to the event.
290            # RecurrenceId does not support timezone information. The exclusion
291            # must have the same timezone as the item to compare.
292            if (
293                isinstance(exdate, datetime.datetime)
294                and isinstance(store_item.dtstart, datetime.datetime)
295                and store_item.dtstart.tzinfo
296            ):
297                exdate = exdate.replace(tzinfo=store_item.dtstart.tzinfo)
298            store_item.exdate.append(exdate)
299            return
300
301        # Assumes any recurrence deletion is valid, and that overwriting
302        # the "until" value will not produce more instances. UNTIL is
303        # inclusive so it can't include the specified exdate. FREQ=DAILY
304        # is the lowest frequency supported so subtracting one day is
305        # safe and works for both dates and datetimes.
306        store_item.rrule.count = None
307        if (
308            isinstance(exdate, datetime.datetime)
309            and isinstance(store_item.dtstart, datetime.datetime)
310            and store_item.dtstart.tzinfo
311        ):
312            exdate = exdate.astimezone(datetime.timezone.utc)
313        store_item.rrule.until = exdate - datetime.timedelta(days=1)
314        now = self._dtstamp_fn()
315        store_item.dtstamp = now
316        store_item.last_modified = now
317
318    def edit(
319        self,
320        uid: str,
321        item: _T,
322        recurrence_id: str | None = None,
323        recurrence_range: Range = Range.NONE,
324    ) -> None:
325        """Update the item with the specified uid.
326
327        The specified item should be created with minimal fields, just
328        including the fields that should be updated. The default fields such
329        as `uid` and `dtstamp` may be used to set the uid for a new created item
330        when updating a recurring item, or for any modification times.
331
332        For a recurring item, either the whole item or individual instances
333        of the item may be edited. To edit the complete range of a recurring
334        item the `uid` property must be specified and the `recurrence_id` should
335        not be specified. To edit an individual instances of the item the
336        `recurrence_id` must be specified. The `recurrence_range` determines if
337        just that individual instance is updated or all items following as well.
338
339        The store will ensure the `ical.calendar.Calendar` has the necessary
340        `ical.timezone.Timezone` needed to fully specify the item time information
341        when encoded.
342        """
343        items_to_edit: list[tuple[int, _T]] = [
344            (index, item)
345            for index, item in _match_items(self._items, uid, recurrence_id)
346        ]
347        if not items_to_edit:
348            raise self._exc(
349                f"No existing item with uid/recurrence_id: {uid}/{recurrence_id}"
350            )
351
352        for store_index, store_item in items_to_edit:
353            self._apply_edit(
354                store_index, store_item, item, recurrence_id, recurrence_range
355            )
356
357    def _apply_edit(
358        self,
359        store_index: int,
360        store_item: _T,
361        item: _T,
362        recurrence_id: str | None = None,
363        recurrence_range: Range = Range.NONE,
364    ) -> None:
365        if (
366            recurrence_id
367            and recurrence_range == Range.THIS_AND_FUTURE
368            and RecurrenceId.to_value(recurrence_id) == store_item.dtstart
369        ):
370            # Editing the first instance and all forward is the same as editing the
371            # entire series so don't bother forking a new item
372            recurrence_id = None
373
374        update = _prepare_update(store_item, item, recurrence_id, recurrence_range)
375        if recurrence_range == Range.NONE:
376            # Changing the recurrence rule of a single item in the middle of the series
377            # is not allowed. It is allowed to convert a single instance item to recurring.
378            if item.rrule and store_item.rrule:
379                if item.rrule.as_rrule_str() != store_item.rrule.as_rrule_str():
380                    raise self._exc(
381                        f"Can't update single instance with rrule (rrule={item.rrule})"
382                    )
383                item.rrule = None
384
385        # Make a deep copy since deletion may update this objects recurrence rules
386        new_item = cast(_T, store_item.copy_and_validate(update=update))
387        if (
388            recurrence_id
389            and new_item.rrule
390            and new_item.rrule.count
391            and store_item.dtstart
392        ):
393            # The recurring item count needs to skip any items that
394            # come before the start of the new item. Use a RulesetIterable
395            # to handle workarounds for dateutil.rrule limitations.
396            dtstart: datetime.date | datetime.datetime = update["dtstart"]
397            ruleset = RulesetIterable(
398                store_item.dtstart,
399                [new_item.rrule.as_rrule(store_item.dtstart)],
400                [],
401                [],
402            )
403            for dtvalue in ruleset:
404                if dtvalue >= dtstart:
405                    break
406                new_item.rrule.count = new_item.rrule.count - 1
407
408        # The store can only manage cascading deletes for some relationship types
409        for relation in new_item.related_to or ():
410            if relation.reltype != RelationshipType.PARENT:
411                raise self._exc(f"Unsupported relationship type {relation.reltype}")
412
413        self._ensure_timezone(new_item.dtstart)
414        if isinstance(new_item, Event) and new_item.dtend:
415            self._ensure_timezone(new_item.dtend)
416
417        # Editing a single instance of a recurring item is like deleting that instance
418        # then adding a new instance on the specified date. If recurrence id is not
419        # specified then the entire item is replaced.
420        self.delete(
421            store_item.uid,
422            recurrence_id=recurrence_id,
423            recurrence_range=recurrence_range,
424        )
425        self._items.insert(store_index, new_item)
426
427    def _ensure_timezone(
428        self, dtvalue: datetime.datetime | datetime.date | None
429    ) -> None:
430        if (new_timezone := _ensure_timezone(dtvalue, self._timezones)) is not None:
431            self._timezones.append(new_timezone)
432
433
434class EventStore(GenericStore[Event]):
435    """An event store manages the lifecycle of events on a Calendar.
436
437    An `ical.calendar.Calendar` is a lower level object that can be directly
438    manipulated to add/remove an `ical.event.Event`. That is, it does not
439    handle updating timestamps, incrementing sequence numbers, or managing
440    lifecycle of a recurring event during an update.
441
442
443    Here is an example for setting up an `EventStore`:
444
445    ```python
446    import datetime
447    from ical.calendar import Calendar
448    from ical.event import Event
449    from ical.store import EventStore
450    from ical.types import Recur
451
452    calendar = Calendar()
453    store = EventStore(calendar)
454
455    event = Event(
456        summary="Event summary",
457        start="2022-07-03",
458        end="2022-07-04",
459        rrule=Recur.from_rrule("FREQ=WEEKLY;COUNT=3"),
460    )
461    store.add(event)
462    ```
463
464    This will add events to the calendar:
465    ```python3
466    for event in calendar.timeline:
467        print(event.summary, event.uid, event.recurrence_id, event.dtstart)
468    ```
469    With output like this:
470    ```
471    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220703 2022-07-03
472    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220710 2022-07-10
473    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220717 2022-07-17
474    ```
475
476    You may also delete an event, or a specific instance of a recurring event:
477    ```python
478    # Delete a single instance of the recurring event
479    store.delete(uid=event.uid, recurrence_id="20220710")
480    ```
481
482    Then viewing the store using the `print` example removes the individual
483    instance in the event:
484    ```
485    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220703 2022-07-03
486    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220717 2022-07-17
487    ```
488
489    Editing an event is also supported:
490    ```python
491    store.edit("event-uid-1", Event(summary="New Summary"))
492    ```
493    """
494
495    def __init__(
496        self,
497        calendar: Calendar,
498        dtstamp_fn: Callable[[], datetime.datetime] = lambda: dtstamp_factory(),
499    ):
500        """Initialize the EventStore."""
501        super().__init__(
502            calendar.events,
503            calendar.timezones,
504            EventStoreError,
505            dtstamp_fn,
506            tzinfo=None,
507        )
508
509
510class TodoStore(GenericStore[Todo]):
511    """A To-do store manages the lifecycle of to-dos on a Calendar."""
512
513    def __init__(
514        self,
515        calendar: Calendar,
516        tzinfo: datetime.tzinfo | None = None,
517        dtstamp_fn: Callable[[], datetime.datetime] = lambda: dtstamp_factory(),
518    ):
519        """Initialize the TodoStore."""
520        super().__init__(
521            calendar.todos,
522            calendar.timezones,
523            TodoStoreError,
524            dtstamp_fn,
525            tzinfo=tzinfo,
526        )
527        self._calendar = calendar
528
529    def todo_list(self, dtstart: datetime.datetime | None = None) -> Iterable[Todo]:
530        """Return a list of all todos on the calendar.
531
532        This view accounts for recurring todos.
533        """
534        return todo_list_view(self._calendar.todos, dtstart)