Merge pull request #213 from niccokunzmann/issue-211-pagination

Issue 211 pagination

Author: niccokunzmann

README.rst

@@ -282,6 +282,55 @@ The result is an iterator that returns the events in order.
     for event in recurring_ical_events.of(an_icalendar_object).after(datetime.datetime.now()):
         print(event["DTSTART"]) # The start is ordered
 
+Pagination
+**********
+
+Pagination allows you to chop the resulting components into chunks of a certain size.
+
+.. code-block: python
+
+    # we get a calendar with 10 events and 2 events per page
+    >>> ten_events = recurring_ical_events.example_calendar("event_10_times")
+    >>> pages = recurring_ical_events.of(ten_events).paginate(2)
+
+    # we can iterate over the pages with 2 events each
+    >>> for i, last_page in enumerate(pages):
+    ...     print(f"page: {i}")
+    ...     for event in last_page:
+    ...         print(f"->  {event.start}")
+    ...     if i == 1: break
+    page: 0
+    ->  2020-01-13 07:45:00+01:00
+    ->  2020-01-14 07:45:00+01:00
+    page: 1
+    ->  2020-01-15 07:45:00+01:00
+    ->  2020-01-16 07:45:00+01:00
+
+If you run a web service and you would like to continue pagination after a certain page,
+this can be done, too. Just hand someone the ``next_page_id`` and continue from there on.
+
+.. code-block: python
+
+    # resume the same query from the next page
+    >>> pages = recurring_ical_events.of(ten_events).paginate(2, next_page_id = last_page.next_page_id)
+    >>> for i, last_page in enumerate(pages):
+    ...     print(f"page: {i + 2}")
+    ...     for event in last_page:
+    ...         print(f"->  {event.start}")
+    ...     if i == 1: break
+    page: 2
+    ->  2020-01-17 07:45:00+01:00
+    ->  2020-01-18 07:45:00+01:00
+    page: 3
+    ->  2020-01-19 07:45:00+01:00
+    ->  2020-01-20 07:45:00+01:00
+
+The ``last_page.next_page_id`` is a string so that it can be used easily.
+It is tested against malicious modification and can safely be passed from a third party source.
+
+Additionally to the page size, you can also pass a ``start`` and an ``end`` to the pages so that
+all components are visible within that time.
+
 Different Components, not just Events
 *************************************
 
@@ -620,9 +669,10 @@ To release new versions,
 Changelog
 ---------
 
-- v3.4.2
+- v3.5.0
 
   - Restructure module into package with a file structure.
+  - Add pagination, see `Issue 211 <https://github.com/niccokunzmann/python-recurring-ical-events/issues/211>`_
 
 - v3.4.1
 

recurring_ical_events/examples.py

@@ -1,4 +1,5 @@
 """Functionality for smaller examples."""
+
 import icalendar
 
 from recurring_ical_events.constants import CALENDARS

recurring_ical_events/occurrence.py

@@ -1,26 +1,85 @@
 """Occurrences of events and other components."""
+
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from datetime import date, datetime, timedelta
+from typing import TYPE_CHECKING, NamedTuple, Optional
 
 from icalendar import Alarm
 
 from recurring_ical_events.adapters.component import ComponentAdapter
-from recurring_ical_events.types import ComponentID
 from recurring_ical_events.util import (
     cached_property,
     make_comparable,
     time_span_contains_event,
 )
 
 if TYPE_CHECKING:
-    import datetime
-
     from icalendar import Alarm
     from icalendar.cal import Component
 
     from recurring_ical_events.adapters.component import ComponentAdapter
-    from recurring_ical_events.types import ComponentID, RecurrenceIDs, Time
+    from recurring_ical_events.types import UID, RecurrenceIDs, Time
+
+
+class OccurrenceID(NamedTuple):
+    """The ID of a component's occurrence to identify it clearly.
+
+    Attributes:
+        name: The name of the component, e.g. "VEVENT"
+        uid: The UID of the component.
+        recurrence_id: The Recurrence-ID of the component in UTC but without tzinfo.
+        start: The start of the component
+    """
+
+    name: str
+    uid: UID
+    recurrence_id: Optional[Time]
+    start: Time
+
+    def to_string(self) -> str:
+        """Return a string representation of this id."""
+        return "#".join(
+            [
+                self.name,
+                self.recurrence_id.isoformat() if self.recurrence_id else "",
+                self.start.isoformat(),
+                self.uid,
+            ]
+        )
+
+    @staticmethod
+    def _dt_from_string(iso_string: str) -> Time:
+        """Create a datetime from the string representation."""
+        if len(iso_string) == 10:
+            return date.fromisoformat(iso_string)
+        return datetime.fromisoformat(iso_string)
+
+    @classmethod
+    def from_string(cls, string_id: str) -> OccurrenceID:
+        """Parse a string and return the component id."""
+        name, recurrence_id, start, uid = string_id.split("#", 3)
+        return cls(
+            name,
+            uid,
+            cls._dt_from_string(recurrence_id) if recurrence_id else None,
+            cls._dt_from_string(start),
+        )
+
+    @classmethod
+    def from_occurrence(
+        cls, name: str, uid: str, recurrence_ids: RecurrenceIDs, start: Time
+    ):
+        """Create a new OccurrenceID from the given values.
+
+        Args:
+            name: The component name.
+            uid: The UID string.
+            recurrence_ids: The recurrence ID tuple.
+                This is expected as UTC with tzinfo being None.
+            start: start time of the component either with or without timezone
+        """
+        return cls(name, uid, recurrence_ids[0] if recurrence_ids else None, start)
 
 
 class Occurrence:
@@ -30,7 +89,7 @@ def __init__(
         self,
         adapter: ComponentAdapter,
         start: Time | None = None,
-        end: Time | None | datetime.timedelta = None,
+        end: Time | None | timedelta = None,
     ):
         """Create an event repetition.
 
@@ -61,13 +120,13 @@ def __lt__(self, other: Occurrence) -> bool:
         return self_start < other_start
 
     @cached_property
-    def id(self) -> ComponentID:
+    def id(self) -> OccurrenceID:
         """The id of the component."""
-        recurrence_id = (*self._adapter.recurrence_ids, self.start)[0]
-        return (
+        return OccurrenceID.from_occurrence(
             self._adapter.component_name(),
             self._adapter.uid,
-            recurrence_id,
+            self._adapter.recurrence_ids,
+            self.start,
         )
 
     def __hash__(self) -> int:
@@ -102,7 +161,7 @@ class AlarmOccurrence(Occurrence):
 
     def __init__(
         self,
-        trigger: datetime.datetime,
+        trigger: datetime,
         alarm: Alarm,
         parent: ComponentAdapter | Occurrence,
     ) -> None:
@@ -122,12 +181,12 @@ def as_component(self, keep_recurrence_attributes):
         return parent
 
     @cached_property
-    def id(self) -> ComponentID:
+    def id(self) -> OccurrenceID:
         """The id of the component."""
-        return (
+        return OccurrenceID.from_occurrence(
             self.parent.component_name(),
             self.parent.uid,
-            *self.parent.recurrence_ids[:1],
+            self.parent.recurrence_ids,
             self.start,
         )
 
@@ -142,4 +201,5 @@ def __repr__(self) -> str:
 __all__ = [
     "Occurrence",
     "AlarmOccurrence",
+    "OccurrenceID",
 ]

recurring_ical_events/pages.py

@@ -0,0 +1,131 @@
+"""Pagination for recurring ical events.
+
+See https://github.com/niccokunzmann/python-recurring-ical-events/issues/211
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Iterator, Optional
+
+from recurring_ical_events.util import compare_greater
+
+if TYPE_CHECKING:
+    from icalendar import Component
+
+    from recurring_ical_events.occurrence import Occurrence
+    from recurring_ical_events.types import Time
+
+
+class Page:
+    """One page in a series of pages."""
+
+    def __init__(self, components: list[Component], next_page_id: str = ""):
+        """ "Create a new page."""
+        self._components = components
+        self._next_page_id = next_page_id
+
+    @property
+    def components(self) -> list[Component]:
+        """All the components of one page."""
+        return self._components
+
+    def has_next_page(self) -> bool:
+        """Wether there is a page following this one."""
+        return self.next_page_id != ""
+
+    @property
+    def next_page_id(self) -> str:
+        """Return the id of the next page or ''."""
+        return self._next_page_id
+
+    def __len__(self) -> int:
+        """The number of components."""
+        return len(self.components)
+
+    def is_last(self):
+        """Wether this is the last page and there is no other page following."""
+        return self._next_page_id == ""
+
+    def __iter__(self) -> Iterator[Component]:
+        """Return an iterator over the components."""
+        return iter(self.components)
+
+
+class Pages:
+    """A pagination configuration to iterate over pages."""
+
+    def __init__(
+        self,
+        occurrence_iterator: Iterator[Occurrence],
+        size: int,
+        stop: Optional[Time] = None,
+        keep_recurrence_attributes: bool = False,  # noqa: FBT001
+    ):
+        """Create a new paginated iterator over components."""
+        self._iterator = occurrence_iterator
+        self._stop = stop
+        self._size = size
+        if self._size <= 0:
+            raise ValueError(
+                f"A page must have at least one component, not {self._size}."
+            )
+        self._keep_recurrence_attributes = keep_recurrence_attributes
+        self._next_occurrence: Optional[Occurrence] = None
+        for occurrence in self._iterator:
+            if self._stop is None or compare_greater(self._stop, occurrence.start):
+                self._next_occurrence = occurrence
+            break
+
+    @property
+    def size(self) -> int:
+        """The maximum number of components per page."""
+        return self._size
+
+    def generate_next_page(self) -> Page:
+        """Generate the next page.
+
+        In contrast to next(self), this does not raise StopIteration.
+        But it works the same: the next page is generated and returned.
+        """
+        for page in self:
+            return page
+        return Page([])
+
+    def __next__(self) -> Page:
+        """Return the next page."""
+        if self._next_occurrence is None:
+            raise StopIteration
+        last_occurrence = self._next_occurrence
+        occurrences = [last_occurrence]
+        for occurrence in self._iterator:
+            if self._stop is not None and compare_greater(occurrence.start, self._stop):
+                break
+            last_occurrence = occurrence
+            if len(occurrences) < self._size:
+                occurrences.append(occurrence)
+            else:
+                break
+        if occurrences[-1] == last_occurrence:
+            self._next_occurrence = None
+        else:
+            self._next_occurrence = last_occurrence
+        return self._create_page_from_occurrences(occurrences)
+
+    def _create_page_from_occurrences(self, occurrences: list[Occurrence]) -> Page:
+        """Create a new page from the occurrences listed."""
+        return Page(
+            [
+                occurrence.as_component(self._keep_recurrence_attributes)
+                for occurrence in occurrences
+            ],
+            next_page_id=self._next_occurrence.id.to_string()
+            if self._next_occurrence is not None
+            else "",
+        )
+
+    def __iter__(self) -> Pages:
+        """Return the iterator."""
+        return self
+
+
+__all__ = ["Page", "Pages"]