Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 4 additions & 18 deletions ical/recur_adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,20 +28,6 @@
_DateOrDatetime = datetime.datetime | datetime.date


def _recurrence_id_for(dt: _DateOrDatetime) -> RecurrenceId:
"""Compute the RecurrenceId for a recurrence date.

This converts a date/datetime from a recurrence expansion into the
floating-time RecurrenceId string used to identify that instance.
"""
# Make recurrence_id floating time to avoid dealing with serializing
# TZID. This value will still be unique within the series and is in
# the context of dtstart which may have a timezone.
if isinstance(dt, datetime.datetime) and dt.tzinfo:
dt = dt.replace(tzinfo=None)
return RecurrenceId.__parse_property_value__(dt)


class FilteredRecurrenceIterable(Iterable[_DateOrDatetime]):
"""An iterable that filters out dates from a recurrence expansion.

Expand All @@ -52,7 +38,7 @@ class FilteredRecurrenceIterable(Iterable[_DateOrDatetime]):
def __init__(
self,
recur: Iterable[_DateOrDatetime],
exclude_ids: frozenset[RecurrenceId],
exclude_ids: frozenset[datetime.date | datetime.datetime],
) -> None:
"""Initialize the filtered iterable."""
self._recur = recur
Expand All @@ -61,7 +47,7 @@ def __init__(
def __iter__(self) -> Iterator[_DateOrDatetime]:
"""Iterate over recurrence dates, excluding overridden ones."""
for dt in self._recur:
if _recurrence_id_for(dt) not in self._exclude_ids:
if dt not in self._exclude_ids:
yield dt


Expand Down Expand Up @@ -89,7 +75,7 @@ def get(
"""Return a lazy sortable item."""

dtend = dtstart + self._duration if self._duration else dtstart
recurrence_id = _recurrence_id_for(dtstart)
recurrence_id = RecurrenceId.from_value(dtstart)

def build() -> ItemType:
updates = {
Expand Down Expand Up @@ -137,7 +123,7 @@ def merge_and_expand_items(
# An edited instance has a recurrence_id (identifying which
# instance it replaces) but no rrule (it's a single instance).
exclude_ids = frozenset(
item.recurrence_id
item.recurrence_id.date
for item in uid_items
if item.recurrence_id and not item.rrule
)
Expand Down
100 changes: 50 additions & 50 deletions ical/store.py
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@ def _ensure_timezone(
) from err


def _match_item(item: _T, uid: str, recurrence_id: str | None) -> bool:
def _match_item(item: _T, uid: str, recurrence_id: RecurrenceId | None) -> bool:
"""Return True if the item is an instance of a recurring event."""
if item.uid != uid:
return False
Expand All @@ -81,21 +81,23 @@ def _match_item(item: _T, uid: str, recurrence_id: str | None) -> bool:
return True
# Match a single item with the specified recurrence_id. If the item is an
# edited instance match return it
if item.recurrence_id == recurrence_id:
if isinstance(recurrence_id.date, datetime.datetime) and isinstance(
item.dtstart, datetime.datetime
):
# Add timezone info if we're missing it
if item.dtstart.tzinfo is not None:
recurrence_id = recurrence_id.from_value(
recurrence_id, timezone=item.dtstart.tzinfo
)

dtstart = recurrence_id.date
if item.recurrence_id and item.recurrence_id.date == dtstart:
_LOGGER.debug("Matched exact recurrence_id: %s", item)
return True
# Otherwise, determine if this instance is in the series
_LOGGER.debug(
"Expanding item %s %s to look for match of %s", uid, item.dtstart, recurrence_id
)
dtstart = RecurrenceId.to_value(recurrence_id)
if isinstance(dtstart, datetime.datetime) and isinstance(
item.dtstart, datetime.datetime
):
# The recurrence_id does not support timezone information, so put it in the
# same timezone as the item to compare.
if item.dtstart.tzinfo is not None:
dtstart = dtstart.replace(tzinfo=item.dtstart.tzinfo)
for dt in item.as_rrule() or ():
if isinstance(dt, datetime.datetime):
if dt.date() > _MAX_SCAN_DATE:
Expand All @@ -111,7 +113,7 @@ def _match_item(item: _T, uid: str, recurrence_id: str | None) -> bool:


def _match_items(
items: list[_T], uid: str, recurrence_id: str | None
items: list[_T], uid: str, recurrence_id: RecurrenceId | None
) -> Generator[tuple[int, _T], None, None]:
"""Return items from the list that match the uid and recurrence_id."""
for index, item in enumerate(items):
Expand All @@ -122,8 +124,7 @@ def _match_items(
def _prepare_update(
store_item: Event | Todo,
item: Event | Todo,
recurrence_id: str | None = None,
recurrence_range: Range = Range.NONE,
recurrence_id: RecurrenceId | None = None,
) -> dict[str, Any]:
"""Prepare an update to an existing event."""
partial_update = item.model_dump(
Expand Down Expand Up @@ -162,7 +163,7 @@ def _prepare_update(
"recurrence_id": recurrence_id,
}
)
if recurrence_range == Range.NONE:
if recurrence_id.range == Range.NONE:
# The new event copied from the original is a single instance,
# which is not recurring.
update["rrule"] = None
Expand All @@ -171,9 +172,7 @@ def _prepare_update(
update["created"] = item.dtstamp

# Adjust start and end time of the event
dtstart: datetime.datetime | datetime.date = RecurrenceId.to_value(
recurrence_id
)
dtstart: datetime.datetime | datetime.date = recurrence_id.date
if item.dtstart:
dtstart = item.dtstart
update["dtstart"] = dtstart
Expand Down Expand Up @@ -245,8 +244,12 @@ def add(self, item: _T) -> _T:
def delete(
self,
uid: str,
recurrence_id: str | None = None,
recurrence_range: Range = Range.NONE,
recurrence_id: RecurrenceId
| datetime.datetime
| datetime.date
| str
| None = None,
recurrence_range: Range | None = None,
) -> None:
"""Delete the item from the calendar.

Expand All @@ -260,27 +263,28 @@ def delete(
When deleting individual instances, the range property may specify
if deletion of just a specific instance, or a range of instances.
"""
recurrence = None
if recurrence_id:
recurrence = RecurrenceId.from_value(recurrence_id, recurrence_range)

items_to_delete: list[_T] = [
item for _, item in _match_items(self._items, uid, recurrence_id)
item for _, item in _match_items(self._items, uid, recurrence)
]
if not items_to_delete:
raise self._exc(
f"No existing item with uid/recurrence_id: {uid}/{recurrence_id}"
)

for store_item in items_to_delete:
self._apply_delete(store_item, recurrence_id, recurrence_range)
self._apply_delete(store_item, recurrence)

def _apply_delete(
self,
store_item: _T,
recurrence_id: str | None = None,
recurrence_range: Range = Range.NONE,
self, store_item: _T, recurrence_id: RecurrenceId | None = None
) -> None:
if (
recurrence_id
and recurrence_range == Range.THIS_AND_FUTURE
and RecurrenceId.to_value(recurrence_id) == store_item.dtstart
and recurrence_id.range == Range.THIS_AND_FUTURE
and recurrence_id.date == store_item.dtstart
):
# Editing the first instance and all forward is the same as editing the
# entire series so don't bother forking a new event
Expand All @@ -302,18 +306,10 @@ def _apply_delete(
self._items.remove(store_item)
return

exdate = RecurrenceId.to_value(recurrence_id)
if recurrence_range == Range.NONE:
exdate = recurrence_id.date
if recurrence_id.range == Range.NONE:
# A single recurrence instance is removed. Add an exclusion to
# to the event.
# RecurrenceId does not support timezone information. The exclusion
# must have the same timezone as the item to compare.
if (
isinstance(exdate, datetime.datetime)
and isinstance(store_item.dtstart, datetime.datetime)
and store_item.dtstart.tzinfo
):
exdate = exdate.replace(tzinfo=store_item.dtstart.tzinfo)
store_item.exdate.append(exdate)
return

Expand All @@ -338,7 +334,11 @@ def edit(
self,
uid: str,
item: _T,
recurrence_id: str | None = None,
recurrence_id: RecurrenceId
| datetime.datetime
| datetime.date
| str
| None = None,
recurrence_range: Range = Range.NONE,
) -> None:
"""Update the item with the specified uid.
Expand All @@ -359,39 +359,39 @@ def edit(
`ical.timezone.Timezone` needed to fully specify the item time information
when encoded.
"""
recurrence = None
if recurrence_id:
recurrence = RecurrenceId.from_value(recurrence_id, recurrence_range)

items_to_edit: list[tuple[int, _T]] = [
(index, item)
for index, item in _match_items(self._items, uid, recurrence_id)
(index, item) for index, item in _match_items(self._items, uid, recurrence)
]
if not items_to_edit:
raise self._exc(
f"No existing item with uid/recurrence_id: {uid}/{recurrence_id}"
)

for store_index, store_item in items_to_edit:
self._apply_edit(
store_index, store_item, item, recurrence_id, recurrence_range
)
self._apply_edit(store_index, store_item, item, recurrence)

def _apply_edit(
self,
store_index: int,
store_item: _T,
item: _T,
recurrence_id: str | None = None,
recurrence_range: Range = Range.NONE,
recurrence_id: RecurrenceId | None = None,
) -> None:
if (
recurrence_id
and recurrence_range == Range.THIS_AND_FUTURE
and RecurrenceId.to_value(recurrence_id) == store_item.dtstart
and recurrence_id.range == Range.THIS_AND_FUTURE
and recurrence_id.date == store_item.dtstart
):
# Editing the first instance and all forward is the same as editing the
# entire series so don't bother forking a new item
recurrence_id = None

update = _prepare_update(store_item, item, recurrence_id, recurrence_range)
if recurrence_range == Range.NONE:
update = _prepare_update(store_item, item, recurrence_id)
if recurrence_id and recurrence_id.range == Range.NONE:
# Changing the recurrence rule of a single item in the middle of the series
# is not allowed. It is allowed to convert a single instance item to recurring.
if item.rrule and store_item.rrule:
Expand Down Expand Up @@ -439,7 +439,7 @@ def _apply_edit(
self.delete(
store_item.uid,
recurrence_id=recurrence_id,
recurrence_range=recurrence_range,
recurrence_range=recurrence_id.range if recurrence_id else None,
)
self._items.insert(store_index, new_item)

Expand Down
Loading