Event: sync events using generators

inbox/events/abstract.py

@@ -1,6 +1,6 @@
 import abc
 import datetime
-from typing import Dict, List, Optional
+from typing import Dict, Iterable, List, Optional
 
 from inbox.events.util import CalendarSyncResponse
 from inbox.logging import get_logger
@@ -39,7 +39,7 @@ def sync_calendars(self) -> CalendarSyncResponse:
     @abc.abstractmethod
     def sync_events(
         self, calendar_uid: str, sync_from_time: Optional[datetime.datetime] = None
-    ) -> List[Event]:
+    ) -> Iterable[Event]:
         """
         Fetch event data for an individual calendar.
 
@@ -49,7 +49,7 @@ def sync_events(
                     changed since this time.
 
         Returns:
-            A list of uncommited Event instances
+            An iterable of uncommited Event instances
         """
         raise NotImplementedError()
 

inbox/events/google.py

@@ -6,7 +6,7 @@
 import random
 import urllib.parse
 import uuid
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Iterable, List, Optional
 
 import arrow
 import attrs
@@ -73,7 +73,7 @@ def sync_calendars(self) -> CalendarSyncResponse:
 
     def sync_events(
         self, calendar_uid: str, sync_from_time: Optional[datetime.datetime] = None
-    ) -> List[Event]:
+    ) -> Iterable[Event]:
         """
         Fetch event data for an individual calendar.
 
@@ -88,29 +88,25 @@ def sync_events(
                 all event data.
 
         Returns:
-            A list of uncommited Event instances
+            An iterable of uncommited Event instances
         """
-        updates = []
         raw_events = self._get_raw_events(calendar_uid, sync_from_time)
         read_only_calendar = self.calendars_table.get(calendar_uid, True)
         for raw_event in iterate_and_periodically_switch_to_gevent(raw_events):
             try:
-                parsed = parse_event_response(raw_event, read_only_calendar)
-                updates.append(parsed)
+                yield parse_event_response(raw_event, read_only_calendar)
             except (arrow.parser.ParserError, ValueError):
                 self.log.warning(
                     "Skipping unparseable event", exc_info=True, raw=raw_event
                 )
 
-        return updates
-
-    def _get_raw_calendars(self) -> List[Dict[str, Any]]:
+    def _get_raw_calendars(self) -> Iterable[Dict[str, Any]]:
         """Gets raw data for the user's calendars."""
         return self._get_resource_list(CALENDARS_URL)
 
     def _get_raw_events(
         self, calendar_uid: str, sync_from_time: Optional[datetime.datetime] = None
-    ) -> List[Dict[str, Any]]:
+    ) -> Iterable[Dict[str, Any]]:
         """Gets raw event data for the given calendar.
 
         Parameters
@@ -123,7 +119,7 @@ def _get_raw_events(
 
         Returns
         -------
-        list of dictionaries representing JSON.
+        iterable of dictionaries representing JSON.
         """
         if sync_from_time is not None:
             # Note explicit offset is required by Google calendar API.
@@ -135,7 +131,7 @@ def _get_raw_events(
             urllib.parse.quote(calendar_uid)
         )
         try:
-            return self._get_resource_list(
+            yield from self._get_resource_list(
                 url, updatedMin=sync_from_time_str, eventTypes="default"
             )
         except requests.exceptions.HTTPError as exc:
@@ -144,14 +140,13 @@ def _get_raw_events(
                 # The calendar API may return 410 if you pass a value for
                 # updatedMin that's too far in the past. In that case, refetch
                 # all events.
-                return self._get_resource_list(url)
+                yield from self._get_resource_list(url)
             else:
                 raise
 
-    def _get_resource_list(self, url: str, **params) -> List[Dict[str, Any]]:
+    def _get_resource_list(self, url: str, **params) -> Iterable[Dict[str, Any]]:
         """Handles response pagination."""
         token = self._get_access_token()
-        items = []
         next_page_token: Optional[str] = None
         params["showDeleted"] = True
         while True:
@@ -161,10 +156,10 @@ def _get_resource_list(self, url: str, **params) -> List[Dict[str, Any]]:
                 r = requests.get(url, params=params, auth=OAuthRequestsWrapper(token))
                 r.raise_for_status()
                 data = r.json()
-                items += data["items"]
+                yield from data["items"]
                 next_page_token = data.get("nextPageToken")
                 if next_page_token is None:
-                    return items
+                    return
 
             except requests.exceptions.SSLError:
                 self.log.warning(

inbox/events/microsoft/events_provider.py

@@ -110,7 +110,7 @@ def sync_calendars(self) -> CalendarSyncResponse:
 
     def sync_events(
         self, calendar_uid: str, sync_from_time: Optional[datetime.datetime] = None
-    ) -> List[Event]:
+    ) -> Iterable[Event]:
         """
         Fetch event data for an individual calendar.
 
@@ -120,15 +120,14 @@ def sync_events(
                     changed since this time.
 
         Returns:
-            A list of uncommited Event instances
+            An iteratable of uncommited Event instances
         """
         if sync_from_time:
             # this got here from the database, we store them as naive
             # UTC in the database. The code downstream is timezone aware so
             # we attach timezone here.
             sync_from_time = sync_from_time.replace(tzinfo=pytz.UTC)
 
-        updates = []
         raw_events = cast(
             Iterable[MsGraphEvent],
             self.client.iter_events(
@@ -142,16 +141,14 @@ def sync_events(
                 continue
 
             event = parse_event(raw_event, read_only=read_only)
-            updates.append(event)
+            yield event
 
             if isinstance(event, RecurringEvent):
                 exceptions, cancellations = self._get_event_overrides(
                     raw_event, event, read_only=read_only
                 )
-                updates.extend(exceptions)
-                updates.extend(cancellations)
-
-        return updates
+                yield from exceptions
+                yield from cancellations
 
     def _get_event_overrides(
         self, raw_master_event: MsGraphEvent, master_event: RecurringEvent, *, read_only

inbox/events/remote_sync.py

@@ -1,5 +1,5 @@
 from datetime import datetime, timedelta
-from typing import Any, List, Tuple, Type
+from typing import Any, Iterable, List, Tuple, Type
 
 import more_itertools
 from requests.exceptions import HTTPError
@@ -167,7 +167,11 @@ def handle_calendar_updates(
 
 
 def handle_event_updates(
-    namespace_id: int, calendar_id: int, events: List[Event], log: Any, db_session: Any
+    namespace_id: int,
+    calendar_id: int,
+    events: Iterable[Event],
+    log: Any,
+    db_session: Any,
 ) -> None:
     """Persists new or updated Event objects to the database."""
     added_count = 0