From 052ba440d05e44fb4977f34f081af526e20bd5df Mon Sep 17 00:00:00 2001 From: Dmitry Date: Wed, 19 May 2021 12:55:54 +0300 Subject: [PATCH] type aliases --- sync_ics2gcal/__init__.py | 8 ++++++-- sync_ics2gcal/gcal.py | 25 +++++++++++++---------- sync_ics2gcal/ical.py | 18 ++++++++++------- sync_ics2gcal/sync.py | 42 +++++++++++++++++++-------------------- 4 files changed, 53 insertions(+), 40 deletions(-) diff --git a/sync_ics2gcal/__init__.py b/sync_ics2gcal/__init__.py index 24b73f5..a97f100 100644 --- a/sync_ics2gcal/__init__.py +++ b/sync_ics2gcal/__init__.py @@ -1,12 +1,16 @@ from .ical import ( CalendarConverter, - EventConverter + EventConverter, + DateDateTime ) from .gcal import ( GoogleCalendarService, - GoogleCalendar + GoogleCalendar, + EventData, + EventList, + EventTuple ) from .sync import ( diff --git a/sync_ics2gcal/gcal.py b/sync_ics2gcal/gcal.py index 6c96cbd..c63b12a 100644 --- a/sync_ics2gcal/gcal.py +++ b/sync_ics2gcal/gcal.py @@ -1,11 +1,15 @@ import logging +from datetime import datetime +from typing import List, Dict, Any, Callable, Tuple, Optional, Union import google.auth from google.oauth2 import service_account from googleapiclient import discovery from pytz import utc -from datetime import datetime -from typing import List, Dict, Any, Callable, Tuple, Optional + +EventData = Dict[str, Union[str, 'EventData', None]] +EventList = List[EventData] +EventTuple = Tuple[EventData, EventData] class GoogleCalendarService: @@ -62,7 +66,7 @@ class GoogleCalendarService: return service -def select_event_key(event: Dict[str, Any]) -> Optional[str]: +def select_event_key(event: EventData) -> Optional[str]: """select event key for logging Arguments: @@ -90,7 +94,7 @@ class GoogleCalendar: self.service: discovery.Resource = service self.calendarId: str = calendarId - def _make_request_callback(self, action: str, events_by_req: List[Dict[str, Any]]) -> Callable: + def _make_request_callback(self, action: str, events_by_req: EventList) -> Callable: """make callback for log result of batch request Arguments: @@ -117,9 +121,10 @@ class GoogleCalendar: key = resp_key self.logger.info('event %s ok, %s: %s', action, key, event.get(key)) + return callback - def list_events_from(self, start: datetime) -> List[Dict[str, Any]]: + def list_events_from(self, start: datetime) -> EventList: """ list events from calendar, where start date >= start """ fields = 'nextPageToken,items(id,iCalUID,updated)' @@ -141,7 +146,7 @@ class GoogleCalendar: self.logger.info('%d events listed', len(events)) return events - def find_exists(self, events: List) -> Tuple[List[Tuple[Dict[str, Any], Dict[str, Any]]], List[Dict[str, Any]]]: + def find_exists(self, events: List) -> Tuple[List[EventTuple], EventList]: """ find existing events from list, by 'iCalUID' field Arguments: @@ -189,7 +194,7 @@ class GoogleCalendar: len(exists), len(not_found)) return exists, not_found - def insert_events(self, events: List[Dict[str, Any]]): + def insert_events(self, events: EventList): """ insert list of events Arguments: @@ -211,7 +216,7 @@ class GoogleCalendar: i += 1 batch.execute() - def patch_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]): + def patch_events(self, event_tuples: List[EventTuple]): """ patch (update) events Arguments: @@ -234,7 +239,7 @@ class GoogleCalendar: i += 1 batch.execute() - def update_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]): + def update_events(self, event_tuples: List[EventTuple]): """ update events Arguments: @@ -257,7 +262,7 @@ class GoogleCalendar: i += 1 batch.execute() - def delete_events(self, events: List[Dict[str, Any]]): + def delete_events(self, events: EventList): """ delete events Arguments: diff --git a/sync_ics2gcal/ical.py b/sync_ics2gcal/ical.py index 678c471..9d0502c 100644 --- a/sync_ics2gcal/ical.py +++ b/sync_ics2gcal/ical.py @@ -1,12 +1,16 @@ import datetime import logging -from typing import Union, Dict, Any, Callable, Optional, List +from typing import Union, Dict, Callable, Optional from icalendar import Calendar, Event from pytz import utc +from .gcal import EventData, EventList -def format_datetime_utc(value: Union[datetime.date, datetime.datetime]) -> str: +DateDateTime = Union[datetime.date, datetime.datetime] + + +def format_datetime_utc(value: DateDateTime) -> str: """utc datetime as string from date or datetime value Arguments: @@ -25,8 +29,8 @@ def format_datetime_utc(value: Union[datetime.date, datetime.datetime]) -> str: ).replace(tzinfo=None).isoformat() + 'Z' -def gcal_date_or_dateTime(value: Union[datetime.date, datetime.datetime], - check_value: Union[datetime.date, datetime.datetime, None] = None)\ +def gcal_date_or_dateTime(value: DateDateTime, + check_value: Optional[DateDateTime] = None) \ -> Dict[str, str]: """date or dateTime to gcal (start or end dict) @@ -117,7 +121,7 @@ class EventConverter(Event): raise ValueError('no DTEND or DURATION') return result - def _put_to_gcal(self, gcal_event: Dict[str, Any], + def _put_to_gcal(self, gcal_event: EventData, prop: str, func: Callable[[str], str], ics_prop: Optional[str] = None): """get property from ical event if exist, and put to gcal event @@ -134,7 +138,7 @@ class EventConverter(Event): if ics_prop in self: gcal_event[prop] = func(ics_prop) - def to_gcal(self) -> Dict[str, Any]: + def to_gcal(self) -> EventData: """Convert Returns: @@ -182,7 +186,7 @@ class CalendarConverter: """ self.calendar = Calendar.from_ical(string) - def events_to_gcal(self) -> List[Dict[str, Any]]: + def events_to_gcal(self) -> EventList: """Convert events to google calendar resources """ diff --git a/sync_ics2gcal/sync.py b/sync_ics2gcal/sync.py index 475ad95..d37964e 100644 --- a/sync_ics2gcal/sync.py +++ b/sync_ics2gcal/sync.py @@ -1,13 +1,13 @@ import datetime import logging import operator -from typing import List, Any, Dict, Set, Tuple, Union, Callable +from typing import List, Dict, Set, Tuple, Union, Callable import dateutil.parser from pytz import utc -from .gcal import GoogleCalendar -from .ical import CalendarConverter +from .gcal import GoogleCalendar, EventData, EventList, EventTuple +from .ical import CalendarConverter, DateDateTime class CalendarSync: @@ -19,15 +19,15 @@ class CalendarSync: def __init__(self, gcalendar: GoogleCalendar, converter: CalendarConverter): self.gcalendar: GoogleCalendar = gcalendar 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]] = [] + self.to_insert: EventList = [] + self.to_update: List[EventTuple] = [] + self.to_delete: EventList = [] @staticmethod - def _events_list_compare(items_src: List[Dict[str, Any]], - items_dst: List[Dict[str, Any]], + def _events_list_compare(items_src: EventList, + items_dst: EventList, key: str = 'iCalUID') \ - -> Tuple[List[Dict[str, Any]], List[Tuple[Dict[str, Any], Dict[str, Any]]], List[Dict[str, Any]]]: + -> Tuple[EventList, List[EventTuple], EventList]: """ compare list of events by key Arguments: @@ -41,7 +41,7 @@ class CalendarSync: items_to_delete) """ - def get_key(item: Dict[str, Any]) -> str: return item[key] + def get_key(item: EventData) -> str: return item[key] keys_src: Set[str] = set(map(get_key, items_src)) keys_dst: Set[str] = set(map(get_key, items_dst)) @@ -50,9 +50,9 @@ class CalendarSync: keys_to_update = keys_src & keys_dst keys_to_delete = keys_dst - keys_src - def items_by_keys(items: List[Dict[str, Any]], + def items_by_keys(items: EventList, key_name: str, - keys: Set[str]) -> List[Dict[str, Any]]: + keys: Set[str]) -> EventList: return list(filter(lambda item: item[key_name] in keys, items)) items_to_insert = items_by_keys(items_src, key, keys_to_insert) @@ -70,7 +70,7 @@ class CalendarSync: """ filter 'to_update' events by 'updated' datetime """ - def filter_updated(event_tuple: Tuple[Dict[str, Any], Dict[str, Any]]) -> bool: + def filter_updated(event_tuple: EventTuple) -> bool: new, old = event_tuple new_date = dateutil.parser.parse(new['updated']) old_date = dateutil.parser.parse(old['updated']) @@ -79,10 +79,10 @@ class CalendarSync: self.to_update = list(filter(filter_updated, self.to_update)) @staticmethod - 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]]: + def _filter_events_by_date(events: EventList, + date: DateDateTime, + op: Callable[[DateDateTime, + DateDateTime], bool]) -> EventList: """ filter events by start datetime Arguments: @@ -94,10 +94,10 @@ class CalendarSync: list of filtred events """ - def filter_by_date(event: Dict[str, Any]) -> bool: + def filter_by_date(event: EventData) -> bool: date_cmp = date event_start: Dict[str, str] = event['start'] - event_date: Union[datetime.date, datetime.datetime, str, None] = None + event_date: Union[DateDateTime, str, None] = None compare_dates = False if 'date' in event_start: @@ -117,7 +117,7 @@ class CalendarSync: return list(filter(filter_by_date, events)) @staticmethod - def _tz_aware_datetime(date: Union[datetime.date, datetime.datetime]) -> datetime.datetime: + def _tz_aware_datetime(date: DateDateTime) -> datetime.datetime: """make tz aware datetime from datetime/date (utc if no tzinfo) Arguments: @@ -133,7 +133,7 @@ class CalendarSync: date = date.replace(tzinfo=utc) return date - def prepare_sync(self, start_date: Union[datetime.date, datetime.datetime]) -> None: + def prepare_sync(self, start_date: DateDateTime) -> None: """prepare sync lists by comparsion of events Arguments: