type annotations - sync

sync_ics2gcal/gcal.py

@@ -16,7 +16,7 @@ class GoogleCalendarService:
     """
 
     @staticmethod
-    def default() -> discovery.Resource:
+    def default():
         """make service Resource from default credentials (authorize)
         ( https://developers.google.com/identity/protocols/application-default-credentials )
         ( https://googleapis.dev/python/google-auth/latest/reference/google.auth.html#google.auth.default )
@@ -29,7 +29,7 @@ class GoogleCalendarService:
         return service
 
     @staticmethod
-    def from_srv_acc_file(service_account_file: str) -> discovery.Resource:
+    def from_srv_acc_file(service_account_file: str):
         """make service Resource from service account filename (authorize)
         """
 
@@ -43,7 +43,7 @@ class GoogleCalendarService:
         return service
 
     @staticmethod
-    def from_config(config: Optional[Dict[str, Optional[str]]] = None) -> discovery.Resource:
+    def from_config(config: Optional[Dict[str, Optional[str]]] = None):
         """make service Resource from config dict
 
         Arguments:
@@ -119,7 +119,7 @@ class GoogleCalendar:
                                  action, key, event.get(key))
         return callback
 
-    def list_events_from(self, start: datetime) -> List[Any]:
+    def list_events_from(self, start: datetime) -> List[Dict[str, Any]]:
         """ list events from calendar, where start date >= start
         """
         fields = 'nextPageToken,items(id,iCalUID,updated)'
@@ -141,7 +141,7 @@ class GoogleCalendar:
         self.logger.info('%d events listed', len(events))
         return events
 
-    def find_exists(self, events: List) -> Tuple[List[Tuple[Any, Any]], List[Any]]:
+    def find_exists(self, events: List) -> Tuple[List[Tuple[Dict[str, Any], Dict[str, Any]]], List[Dict[str, Any]]]:
         """ find existing events from list, by 'iCalUID' field
 
         Arguments:
@@ -189,7 +189,7 @@ class GoogleCalendar:
                          len(exists), len(not_found))
         return exists, not_found
 
-    def insert_events(self, events: List[Any]):
+    def insert_events(self, events: List[Dict[str, Any]]):
         """ insert list of events
 
         Arguments:
@@ -211,7 +211,7 @@ class GoogleCalendar:
             i += 1
         batch.execute()
 
-    def patch_events(self, event_tuples: List[Tuple[Any, Any]]):
+    def patch_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]):
         """ patch (update) events
 
         Arguments:
@@ -234,7 +234,7 @@ class GoogleCalendar:
             i += 1
         batch.execute()
 
-    def update_events(self, event_tuples: List[Tuple[Any, Any]]):
+    def update_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]):
         """ update events
 
         Arguments:
@@ -257,7 +257,7 @@ class GoogleCalendar:
             i += 1
         batch.execute()
 
-    def delete_events(self, events: List[Any]):
+    def delete_events(self, events: List[Dict[str, Any]]):
         """ delete events
 
         Arguments:

sync_ics2gcal/sync.py

@@ -1,22 +1,33 @@
 import datetime
-import dateutil.parser
 import logging
 import operator
+from typing import List, Any, Dict, Set, Tuple, Union, Callable
+
+import dateutil.parser
 from pytz import utc
 
+from .gcal import GoogleCalendar
+from .ical import CalendarConverter
 
-class CalendarSync():
+
+class CalendarSync:
     """class for syncronize calendar with google
     """
 
     logger = logging.getLogger('CalendarSync')
 
-    def __init__(self, gcalendar, converter):
+    def __init__(self, gcalendar: GoogleCalendar, converter: CalendarConverter):
-        self.gcalendar = gcalendar
+        self.gcalendar: GoogleCalendar = gcalendar
-        self.converter = converter
+        self.converter: CalendarConverter = converter
+        self.to_insert: List[Dict[str, Any]] = []
+        self.to_update: List[Tuple[Dict[str, Any], Dict[str, Any]]] = []
+        self.to_delete: List[Dict[str, Any]] = []
 
     @staticmethod
-    def _events_list_compare(items_src, items_dst, key='iCalUID'):
+    def _events_list_compare(items_src: List[Dict[str, Any]],
+                             items_dst: List[Dict[str, Any]],
+                             key: str = 'iCalUID') \
+            -> Tuple[List[Dict[str, Any]], List[Tuple[Dict[str, Any], Dict[str, Any]]], List[Dict[str, Any]]]:
         """ compare list of events by key
 
         Arguments:
@@ -30,16 +41,18 @@ class CalendarSync():
                       items_to_delete)
         """
 
-        def get_key(item): return item[key]
+        def get_key(item: Dict[str, Any]) -> str: return item[key]
 
-        keys_src = set(map(get_key, items_src))
+        keys_src: Set[str] = set(map(get_key, items_src))
-        keys_dst = set(map(get_key, items_dst))
+        keys_dst: Set[str] = set(map(get_key, items_dst))
 
         keys_to_insert = keys_src - keys_dst
         keys_to_update = keys_src & keys_dst
         keys_to_delete = keys_dst - keys_src
 
-        def items_by_keys(items, key_name, keys):
+        def items_by_keys(items: List[Dict[str, Any]],
+                          key_name: str,
+                          keys: Set[str]) -> List[Dict[str, Any]]:
             return list(filter(lambda item: item[key_name] in keys, items))
 
         items_to_insert = items_by_keys(items_src, key, keys_to_insert)
@@ -57,7 +70,7 @@ class CalendarSync():
         """ filter 'to_update' events by 'updated' datetime
         """
 
-        def filter_updated(event_tuple):
+        def filter_updated(event_tuple: Tuple[Dict[str, Any], Dict[str, Any]]) -> bool:
             new, old = event_tuple
             new_date = dateutil.parser.parse(new['updated'])
             old_date = dateutil.parser.parse(old['updated'])
@@ -66,7 +79,10 @@ class CalendarSync():
         self.to_update = list(filter(filter_updated, self.to_update))
 
     @staticmethod
-    def _filter_events_by_date(events, date, op):
+    def _filter_events_by_date(events: List[Dict[str, Any]],
+                               date: Union[datetime.date, datetime.datetime],
+                               op: Callable[[Union[datetime.date, datetime.datetime],
+                                             Union[datetime.date, datetime.datetime]], bool]) -> List[Dict[str, Any]]:
         """ filter events by start datetime
 
         Arguments:
@@ -78,10 +94,10 @@ class CalendarSync():
             list of filtred events
         """
 
-        def filter_by_date(event):
+        def filter_by_date(event: Dict[str, Any]) -> bool:
             date_cmp = date
-            event_start = event['start']
+            event_start: Dict[str, str] = event['start']
-            event_date = None
+            event_date: Union[datetime.date, datetime.datetime, str, None] = None
             compare_dates = False
 
             if 'date' in event_start:
@@ -101,7 +117,8 @@ class CalendarSync():
         return list(filter(filter_by_date, events))
 
     @staticmethod
-    def _tz_aware_datetime(date):
+    def _tz_aware_datetime(date: Union[datetime.date, datetime.datetime]) \
+            -> Union[datetime.date, datetime.datetime]:
         """make tz aware datetime from datetime/date (utc if no tzinfo)
 
         Arguments:
@@ -117,7 +134,7 @@ class CalendarSync():
             date = date.replace(tzinfo=utc)
         return date
 
-    def prepare_sync(self, start_date):
+    def prepare_sync(self, start_date: Union[datetime.date, datetime.datetime]) -> None:
         """prepare sync lists by comparsion of events
 
         Arguments:
@@ -135,28 +152,20 @@ class CalendarSync():
         events_src_past = CalendarSync._filter_events_by_date(
             events_src, start_date, operator.lt)
 
-        events_src = None
-
         # first events comparsion
         self.to_insert, self.to_update, self.to_delete = CalendarSync._events_list_compare(
             events_src_pending, events_dst)
 
-        events_src_pending, events_dst = None, None
-
         # find in events 'to_delete' past events from source, for update (move to past)
         _, add_to_update, self.to_delete = CalendarSync._events_list_compare(
             events_src_past, self.to_delete)
         self.to_update.extend(add_to_update)
 
-        events_src_past = None
-
         # find if events 'to_insert' exists in gcalendar, for update them
         add_to_update, self.to_insert = self.gcalendar.find_exists(
             self.to_insert)
         self.to_update.extend(add_to_update)
 
-        add_to_update = None
-
         # exclude outdated events from 'to_update' list, by 'updated' field
         self._filter_events_to_update()
 
@@ -167,7 +176,14 @@ class CalendarSync():
             len(self.to_delete)
         )
 
-    def apply(self):
+    def clear(self) -> None:
+        """ clear prepared sync lists (insert, update, delete)
+        """
+        self.to_insert.clear()
+        self.to_update.clear()
+        self.to_delete.clear()
+
+    def apply(self) -> None:
         """ apply sync (insert, update, delete), using prepared lists of events
         """
 
@@ -175,6 +191,6 @@ class CalendarSync():
         self.gcalendar.update_events(self.to_update)
         self.gcalendar.delete_events(self.to_delete)
 
-        self.logger.info('sync done')
+        self.clear()
 
-        self.to_insert, self.to_update, self.to_delete = [], [], []
+        self.logger.info('sync done')