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, TodoStatus
 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.model_dump(
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 (
142        isinstance(item, Todo)
143        and isinstance(store_item, Todo)
144        and item.status
145        and not item.completed
146    ):
147        if (
148            store_item.status != TodoStatus.COMPLETED
149            and item.status == TodoStatus.COMPLETED
150        ):
151            update["completed"] = item.dtstamp
152        if store_item.completed and item.status != TodoStatus.COMPLETED:
153            update["completed"] = None
154    if rrule := update.get("rrule"):
155        update["rrule"] = Recur.model_validate(rrule)
156    if recurrence_id and store_item.rrule:
157        # Forking a new event off the old event preserves the original uid and
158        # recurrence_id.
159        update.update(
160            {
161                "uid": store_item.uid,
162                "recurrence_id": recurrence_id,
163            }
164        )
165        if recurrence_range == Range.NONE:
166            # The new event copied from the original is a single instance,
167            # which is not recurring.
168            update["rrule"] = None
169        else:
170            # Overwriting with a new recurring event
171            update["created"] = item.dtstamp
172
173            # Adjust start and end time of the event
174            dtstart: datetime.datetime | datetime.date = RecurrenceId.to_value(
175                recurrence_id
176            )
177            if item.dtstart:
178                dtstart = item.dtstart
179            update["dtstart"] = dtstart
180            # Event either has a duration (which should already be set) or has
181            # an explicit end which needs to be realigned to new start time.
182            if isinstance(store_item, Event) and store_item.dtend:
183                update["dtend"] = dtstart + store_item.computed_duration
184    return update
185
186
187class GenericStore(Generic[_T]):
188    """A a store manages the lifecycle of items on a Calendar."""
189
190    def __init__(
191        self,
192        items: list[_T],
193        timezones: list[Timezone],
194        exc: type[StoreError],
195        dtstamp_fn: Callable[[], datetime.datetime] = lambda: dtstamp_factory(),
196        tzinfo: datetime.tzinfo | None = None,
197    ):
198        """Initialize the EventStore."""
199        self._items = items
200        self._timezones = timezones
201        self._exc = exc
202        self._dtstamp_fn = dtstamp_fn
203        self._tzinfo = tzinfo or local_timezone()
204
205    def add(self, item: _T) -> _T:
206        """Add the specified item to the calendar.
207
208        This will handle assigning modification dates, sequence numbers, etc
209        if those fields are unset.
210
211        The store will ensure the `ical.calendar.Calendar` has the necessary
212        `ical.timezone.Timezone` needed to fully specify the time information
213        when encoded.
214        """
215        update: dict[str, Any] = {}
216        if not item.created:
217            update["created"] = item.dtstamp
218        if item.sequence is None:
219            update["sequence"] = 0
220        if isinstance(item, Todo) and not item.dtstart:
221            if item.due:
222                update["dtstart"] = item.due - datetime.timedelta(days=1)
223            else:
224                update["dtstart"] = datetime.datetime.now(tz=self._tzinfo)
225        if (
226            isinstance(item, Todo)
227            and not item.completed
228            and item.status == TodoStatus.COMPLETED
229        ):
230            update["completed"] = item.dtstamp
231        new_item = cast(_T, item.copy_and_validate(update=update))
232
233        # The store can only manage cascading deletes for some relationship types
234        for relation in new_item.related_to or ():
235            if relation.reltype != RelationshipType.PARENT:
236                raise self._exc(f"Unsupported relationship type {relation.reltype}")
237
238        _LOGGER.debug("Adding item: %s", new_item)
239        self._ensure_timezone(item.dtstart)
240        if isinstance(item, Event) and item.dtend:
241            self._ensure_timezone(item.dtend)
242        self._items.append(new_item)
243        return new_item
244
245    def delete(
246        self,
247        uid: str,
248        recurrence_id: str | None = None,
249        recurrence_range: Range = Range.NONE,
250    ) -> None:
251        """Delete the item from the calendar.
252
253        This method is used to delete an existing item. For a recurring item
254        either the whole item or instances of an item may be deleted. To
255        delete the complete range of a recurring item, the `uid` property
256        for the item must be specified and the `recurrence_id` should not
257        be specified. To delete an individual instance of the item the
258        `recurrence_id` must be specified.
259
260        When deleting individual instances, the range property may specify
261        if deletion of just a specific instance, or a range of instances.
262        """
263        items_to_delete: list[_T] = [
264            item for _, item in _match_items(self._items, uid, recurrence_id)
265        ]
266        if not items_to_delete:
267            raise self._exc(
268                f"No existing item with uid/recurrence_id: {uid}/{recurrence_id}"
269            )
270
271        for store_item in items_to_delete:
272            self._apply_delete(store_item, recurrence_id, recurrence_range)
273
274    def _apply_delete(
275        self,
276        store_item: _T,
277        recurrence_id: str | None = None,
278        recurrence_range: Range = Range.NONE,
279    ) -> None:
280        if (
281            recurrence_id
282            and recurrence_range == Range.THIS_AND_FUTURE
283            and RecurrenceId.to_value(recurrence_id) == store_item.dtstart
284        ):
285            # Editing the first instance and all forward is the same as editing the
286            # entire series so don't bother forking a new event
287            recurrence_id = None
288
289        children = []
290        for event in self._items:
291            for relation in event.related_to or ():
292                if (
293                    relation.reltype == RelationshipType.PARENT
294                    and relation.uid == store_item.uid
295                ):
296                    children.append(event)
297        for child in children:
298            self._items.remove(child)
299
300        # Deleting all instances in the series
301        if not recurrence_id or not store_item.rrule:
302            self._items.remove(store_item)
303            return
304
305        exdate = RecurrenceId.to_value(recurrence_id)
306        if recurrence_range == Range.NONE:
307            # A single recurrence instance is removed. Add an exclusion to
308            # to the event.
309            # RecurrenceId does not support timezone information. The exclusion
310            # must have the same timezone as the item to compare.
311            if (
312                isinstance(exdate, datetime.datetime)
313                and isinstance(store_item.dtstart, datetime.datetime)
314                and store_item.dtstart.tzinfo
315            ):
316                exdate = exdate.replace(tzinfo=store_item.dtstart.tzinfo)
317            store_item.exdate.append(exdate)
318            return
319
320        # Assumes any recurrence deletion is valid, and that overwriting
321        # the "until" value will not produce more instances. UNTIL is
322        # inclusive so it can't include the specified exdate. FREQ=DAILY
323        # is the lowest frequency supported so subtracting one day is
324        # safe and works for both dates and datetimes.
325        store_item.rrule.count = None
326        if (
327            isinstance(exdate, datetime.datetime)
328            and isinstance(store_item.dtstart, datetime.datetime)
329            and store_item.dtstart.tzinfo
330        ):
331            exdate = exdate.astimezone(datetime.timezone.utc)
332        store_item.rrule.until = exdate - datetime.timedelta(days=1)
333        now = self._dtstamp_fn()
334        store_item.dtstamp = now
335        store_item.last_modified = now
336
337    def edit(
338        self,
339        uid: str,
340        item: _T,
341        recurrence_id: str | None = None,
342        recurrence_range: Range = Range.NONE,
343    ) -> None:
344        """Update the item with the specified uid.
345
346        The specified item should be created with minimal fields, just
347        including the fields that should be updated. The default fields such
348        as `uid` and `dtstamp` may be used to set the uid for a new created item
349        when updating a recurring item, or for any modification times.
350
351        For a recurring item, either the whole item or individual instances
352        of the item may be edited. To edit the complete range of a recurring
353        item the `uid` property must be specified and the `recurrence_id` should
354        not be specified. To edit an individual instances of the item the
355        `recurrence_id` must be specified. The `recurrence_range` determines if
356        just that individual instance is updated or all items following as well.
357
358        The store will ensure the `ical.calendar.Calendar` has the necessary
359        `ical.timezone.Timezone` needed to fully specify the item time information
360        when encoded.
361        """
362        items_to_edit: list[tuple[int, _T]] = [
363            (index, item)
364            for index, item in _match_items(self._items, uid, recurrence_id)
365        ]
366        if not items_to_edit:
367            raise self._exc(
368                f"No existing item with uid/recurrence_id: {uid}/{recurrence_id}"
369            )
370
371        for store_index, store_item in items_to_edit:
372            self._apply_edit(
373                store_index, store_item, item, recurrence_id, recurrence_range
374            )
375
376    def _apply_edit(
377        self,
378        store_index: int,
379        store_item: _T,
380        item: _T,
381        recurrence_id: str | None = None,
382        recurrence_range: Range = Range.NONE,
383    ) -> None:
384        if (
385            recurrence_id
386            and recurrence_range == Range.THIS_AND_FUTURE
387            and RecurrenceId.to_value(recurrence_id) == store_item.dtstart
388        ):
389            # Editing the first instance and all forward is the same as editing the
390            # entire series so don't bother forking a new item
391            recurrence_id = None
392
393        update = _prepare_update(store_item, item, recurrence_id, recurrence_range)
394        if recurrence_range == Range.NONE:
395            # Changing the recurrence rule of a single item in the middle of the series
396            # is not allowed. It is allowed to convert a single instance item to recurring.
397            if item.rrule and store_item.rrule:
398                if item.rrule.as_rrule_str() != store_item.rrule.as_rrule_str():
399                    raise self._exc(
400                        f"Can't update single instance with rrule (rrule={item.rrule})"
401                    )
402                item.rrule = None
403
404        # Make a deep copy since deletion may update this objects recurrence rules
405        new_item = cast(_T, store_item.copy_and_validate(update=update))
406        if (
407            recurrence_id
408            and new_item.rrule
409            and new_item.rrule.count
410            and store_item.dtstart
411        ):
412            # The recurring item count needs to skip any items that
413            # come before the start of the new item. Use a RulesetIterable
414            # to handle workarounds for dateutil.rrule limitations.
415            dtstart: datetime.date | datetime.datetime = update["dtstart"]
416            ruleset = RulesetIterable(
417                store_item.dtstart,
418                [new_item.rrule.as_rrule(store_item.dtstart)],
419                [],
420                [],
421            )
422            for dtvalue in ruleset:
423                if dtvalue >= dtstart:
424                    break
425                new_item.rrule.count = new_item.rrule.count - 1
426
427        # The store can only manage cascading deletes for some relationship types
428        for relation in new_item.related_to or ():
429            if relation.reltype != RelationshipType.PARENT:
430                raise self._exc(f"Unsupported relationship type {relation.reltype}")
431
432        self._ensure_timezone(new_item.dtstart)
433        if isinstance(new_item, Event) and new_item.dtend:
434            self._ensure_timezone(new_item.dtend)
435
436        # Editing a single instance of a recurring item is like deleting that instance
437        # then adding a new instance on the specified date. If recurrence id is not
438        # specified then the entire item is replaced.
439        self.delete(
440            store_item.uid,
441            recurrence_id=recurrence_id,
442            recurrence_range=recurrence_range,
443        )
444        self._items.insert(store_index, new_item)
445
446    def _ensure_timezone(
447        self, dtvalue: datetime.datetime | datetime.date | None
448    ) -> None:
449        if (new_timezone := _ensure_timezone(dtvalue, self._timezones)) is not None:
450            self._timezones.append(new_timezone)
451
452
453class EventStore(GenericStore[Event]):
454    """An event store manages the lifecycle of events on a Calendar.
455
456    An `ical.calendar.Calendar` is a lower level object that can be directly
457    manipulated to add/remove an `ical.event.Event`. That is, it does not
458    handle updating timestamps, incrementing sequence numbers, or managing
459    lifecycle of a recurring event during an update.
460
461
462    Here is an example for setting up an `EventStore`:
463
464    ```python
465    import datetime
466    from ical.calendar import Calendar
467    from ical.event import Event
468    from ical.store import EventStore
469    from ical.types import Recur
470
471    calendar = Calendar()
472    store = EventStore(calendar)
473
474    event = Event(
475        summary="Event summary",
476        start="2022-07-03",
477        end="2022-07-04",
478        rrule=Recur.from_rrule("FREQ=WEEKLY;COUNT=3"),
479    )
480    store.add(event)
481    ```
482
483    This will add events to the calendar:
484    ```python3
485    for event in calendar.timeline:
486        print(event.summary, event.uid, event.recurrence_id, event.dtstart)
487    ```
488    With output like this:
489    ```
490    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220703 2022-07-03
491    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220710 2022-07-10
492    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220717 2022-07-17
493    ```
494
495    You may also delete an event, or a specific instance of a recurring event:
496    ```python
497    # Delete a single instance of the recurring event
498    store.delete(uid=event.uid, recurrence_id="20220710")
499    ```
500
501    Then viewing the store using the `print` example removes the individual
502    instance in the event:
503    ```
504    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220703 2022-07-03
505    Event summary a521cf45-2c02-11ed-9e5c-066a07ffbaf5 20220717 2022-07-17
506    ```
507
508    Editing an event is also supported:
509    ```python
510    store.edit("event-uid-1", Event(summary="New Summary"))
511    ```
512    """
513
514    def __init__(
515        self,
516        calendar: Calendar,
517        dtstamp_fn: Callable[[], datetime.datetime] = lambda: dtstamp_factory(),
518    ):
519        """Initialize the EventStore."""
520        super().__init__(
521            calendar.events,
522            calendar.timezones,
523            EventStoreError,
524            dtstamp_fn,
525            tzinfo=None,
526        )
527
528
529class TodoStore(GenericStore[Todo]):
530    """A To-do store manages the lifecycle of to-dos on a Calendar."""
531
532    def __init__(
533        self,
534        calendar: Calendar,
535        tzinfo: datetime.tzinfo | None = None,
536        dtstamp_fn: Callable[[], datetime.datetime] = lambda: dtstamp_factory(),
537    ):
538        """Initialize the TodoStore."""
539        super().__init__(
540            calendar.todos,
541            calendar.timezones,
542            TodoStoreError,
543            dtstamp_fn,
544            tzinfo=tzinfo,
545        )
546        self._calendar = calendar
547
548    def todo_list(self, dtstart: datetime.datetime | None = None) -> Iterable[Todo]:
549        """Return a list of all todos on the calendar.
550
551        This view accounts for recurring todos.
552        """
553        return todo_list_view(self._calendar.todos, dtstart)