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