diff --git a/.gitignore b/.gitignore index d98c726..b5c5b91 100644 --- a/.gitignore +++ b/.gitignore @@ -2,8 +2,10 @@ config.yml service-account.json *.pyc my-test*.ics -.vscode/* +.vscode/ +.idea/ /dist/ /*.egg-info/ /build/ /.eggs/ +venv/ diff --git a/sync_ics2gcal/gcal.py b/sync_ics2gcal/gcal.py index db6b831..6c96cbd 100644 --- a/sync_ics2gcal/gcal.py +++ b/sync_ics2gcal/gcal.py @@ -4,9 +4,11 @@ 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 -class GoogleCalendarService(): +class GoogleCalendarService: """class for make google calendar service Resource Returns: @@ -18,9 +20,6 @@ class GoogleCalendarService(): """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 ) - - Returns: - service Resource """ scopes = ['https://www.googleapis.com/auth/calendar'] @@ -30,11 +29,8 @@ class GoogleCalendarService(): return service @staticmethod - def from_srv_acc_file(service_account_file): + def from_srv_acc_file(service_account_file: str): """make service Resource from service account filename (authorize) - - Returns: - service Resource """ scopes = ['https://www.googleapis.com/auth/calendar'] @@ -47,18 +43,15 @@ class GoogleCalendarService(): return service @staticmethod - def from_config(config=None): + def from_config(config: Optional[Dict[str, Optional[str]]] = None): """make service Resource from config dict Arguments: - config -- dict() config with keys: + config -- config with keys: (optional) service_account: - service account filename if key not in dict then default credentials will be used ( https://developers.google.com/identity/protocols/application-default-credentials ) -- None: default credentials will be used - - Returns: - service Resource """ if config is not None and 'service_account' in config: @@ -69,7 +62,7 @@ class GoogleCalendarService(): return service -def select_event_key(event): +def select_event_key(event: Dict[str, Any]) -> Optional[str]: """select event key for logging Arguments: @@ -87,17 +80,17 @@ def select_event_key(event): return key -class GoogleCalendar(): +class GoogleCalendar: """class to interact with calendar on google """ logger = logging.getLogger('GoogleCalendar') - def __init__(self, service, calendarId): - self.service = service - self.calendarId = calendarId + def __init__(self, service: discovery.Resource, calendarId: Optional[str]): + self.service: discovery.Resource = service + self.calendarId: str = calendarId - def _make_request_callback(self, action, events_by_req): + def _make_request_callback(self, action: str, events_by_req: List[Dict[str, Any]]) -> Callable: """make callback for log result of batch request Arguments: @@ -126,7 +119,7 @@ class GoogleCalendar(): action, key, event.get(key)) return callback - def list_events_from(self, start): + 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)' @@ -148,7 +141,7 @@ class GoogleCalendar(): self.logger.info('%d events listed', len(events)) return events - def find_exists(self, events): + 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: @@ -166,16 +159,16 @@ class GoogleCalendar(): def list_callback(request_id, response, exception): found = False - event = events_by_req[int(request_id)] + cur_event = events_by_req[int(request_id)] if exception is None: found = ([] != response['items']) else: self.logger.error( 'exception %s, while listing event with UID: %s', - str(exception), event['iCalUID']) + str(exception), cur_event['iCalUID']) if found: exists.append( - (event, response['items'][0])) + (cur_event, response['items'][0])) else: not_found.append(events_by_req[int(request_id)]) @@ -196,7 +189,7 @@ class GoogleCalendar(): len(exists), len(not_found)) return exists, not_found - def insert_events(self, events): + def insert_events(self, events: List[Dict[str, Any]]): """ insert list of events Arguments: @@ -218,7 +211,7 @@ class GoogleCalendar(): i += 1 batch.execute() - def patch_events(self, event_tuples): + def patch_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]): """ patch (update) events Arguments: @@ -241,7 +234,7 @@ class GoogleCalendar(): i += 1 batch.execute() - def update_events(self, event_tuples): + def update_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]): """ update events Arguments: @@ -264,7 +257,7 @@ class GoogleCalendar(): i += 1 batch.execute() - def delete_events(self, events): + def delete_events(self, events: List[Dict[str, Any]]): """ delete events Arguments: @@ -284,7 +277,7 @@ class GoogleCalendar(): i += 1 batch.execute() - def create(self, summary, timeZone=None): + def create(self, summary: str, timeZone: Optional[str] = None) -> Any: """create calendar Arguments: @@ -328,7 +321,7 @@ class GoogleCalendar(): body=rule_public ).execute() - def add_owner(self, email): + def add_owner(self, email: str): """add calendar owner by email Arguments: diff --git a/sync_ics2gcal/ical.py b/sync_ics2gcal/ical.py index 670ad75..678c471 100644 --- a/sync_ics2gcal/ical.py +++ b/sync_ics2gcal/ical.py @@ -1,12 +1,14 @@ import datetime import logging +from typing import Union, Dict, Any, Callable, Optional, List from icalendar import Calendar, Event from pytz import utc -def format_datetime_utc(value): +def format_datetime_utc(value: Union[datetime.date, datetime.datetime]) -> str: """utc datetime as string from date or datetime value + Arguments: value -- date or datetime value @@ -23,20 +25,23 @@ def format_datetime_utc(value): ).replace(tzinfo=None).isoformat() + 'Z' -def gcal_date_or_dateTime(value, check_value=None): +def gcal_date_or_dateTime(value: Union[datetime.date, datetime.datetime], + check_value: Union[datetime.date, datetime.datetime, None] = None)\ + -> Dict[str, str]: """date or dateTime to gcal (start or end dict) + Arguments: - value -- date or datetime value - check_value - date or datetime to choise result type (if not None) + value: date or datetime + check_value: optional for choose result type Returns: - dict { 'date': ... } or { 'dateTime': ... } + { 'date': ... } or { 'dateTime': ... } """ if check_value is None: check_value = value - result = {} + result: Dict[str, str] = {} if isinstance(check_value, datetime.datetime): result['dateTime'] = format_datetime_utc(value) else: @@ -52,7 +57,7 @@ class EventConverter(Event): ( https://developers.google.com/calendar/v3/reference/events#resource-representations ) """ - def _str_prop(self, prop): + def _str_prop(self, prop: str) -> str: """decoded string property Arguments: @@ -64,7 +69,7 @@ class EventConverter(Event): return self.decoded(prop).decode(encoding='utf-8') - def _datetime_str_prop(self, prop): + def _datetime_str_prop(self, prop: str) -> str: """utc datetime as string from property Arguments: @@ -76,7 +81,7 @@ class EventConverter(Event): return format_datetime_utc(self.decoded(prop)) - def _gcal_start(self): + def _gcal_start(self) -> Dict[str, str]: """ event start dict from icalendar event Raises: @@ -89,7 +94,7 @@ class EventConverter(Event): value = self.decoded('DTSTART') return gcal_date_or_dateTime(value) - def _gcal_end(self): + def _gcal_end(self) -> Dict[str, str]: """event end dict from icalendar event Raises: @@ -112,7 +117,9 @@ class EventConverter(Event): raise ValueError('no DTEND or DURATION') return result - def _put_to_gcal(self, gcal_event, prop, func, ics_prop=None): + def _put_to_gcal(self, gcal_event: Dict[str, Any], + prop: str, func: Callable[[str], str], + ics_prop: Optional[str] = None): """get property from ical event if exist, and put to gcal event Arguments: @@ -127,7 +134,7 @@ class EventConverter(Event): if ics_prop in self: gcal_event[prop] = func(ics_prop) - def to_gcal(self): + def to_gcal(self) -> Dict[str, Any]: """Convert Returns: @@ -135,12 +142,11 @@ class EventConverter(Event): """ event = { - 'iCalUID': self._str_prop('UID') + 'iCalUID': self._str_prop('UID'), + 'start': self._gcal_start(), + 'end': self._gcal_end() } - event['start'] = self._gcal_start() - event['end'] = self._gcal_end() - self._put_to_gcal(event, 'summary', self._str_prop) self._put_to_gcal(event, 'description', self._str_prop) self._put_to_gcal(event, 'location', self._str_prop) @@ -155,28 +161,28 @@ class EventConverter(Event): return event -class CalendarConverter(): +class CalendarConverter: """Convert icalendar events to google calendar resources """ logger = logging.getLogger('CalendarConverter') - def __init__(self, calendar=None): - self.calendar = calendar + def __init__(self, calendar: Optional[Calendar] = None): + self.calendar: Optional[Calendar] = calendar - def load(self, filename): + def load(self, filename: str): """ load calendar from ics file """ with open(filename, 'r', encoding='utf-8') as f: self.calendar = Calendar.from_ical(f.read()) self.logger.info('%s loaded', filename) - def loads(self, string): + def loads(self, string: str): """ load calendar from ics string """ self.calendar = Calendar.from_ical(string) - def events_to_gcal(self): + def events_to_gcal(self) -> List[Dict[str, Any]]: """Convert events to google calendar resources """ diff --git a/sync_ics2gcal/manage_calendars.py b/sync_ics2gcal/manage_calendars.py index ed3c6ef..a2c10a5 100644 --- a/sync_ics2gcal/manage_calendars.py +++ b/sync_ics2gcal/manage_calendars.py @@ -1,5 +1,6 @@ import argparse import logging.config +from typing import Optional, Dict, Any import yaml @@ -70,7 +71,7 @@ def parse_args(): return args -def load_config(): +def load_config() -> Optional[Dict[str, Any]]: result = None try: with open('config.yml', 'r', encoding='utf-8') as f: @@ -81,7 +82,7 @@ def load_config(): return result -def list_calendars(service, show_hidden, show_deleted): +def list_calendars(service, show_hidden: bool, show_deleted: bool) -> None: fields = 'nextPageToken,items(id,summary)' calendars = [] page_token = None @@ -100,7 +101,7 @@ def list_calendars(service, show_hidden, show_deleted): print('{summary}: {id}'.format_map(calendar)) -def create_calendar(service, summary, timezone, public): +def create_calendar(service, summary: str, timezone: str, public: bool) -> None: calendar = GoogleCalendar(service, None) calendar.create(summary, timezone) if public: @@ -108,33 +109,33 @@ def create_calendar(service, summary, timezone, public): print('{}: {}'.format(summary, calendar.calendarId)) -def add_owner(service, id, owner_email): - calendar = GoogleCalendar(service, id) +def add_owner(service, calendar_id: str, owner_email: str) -> None: + calendar = GoogleCalendar(service, calendar_id) calendar.add_owner(owner_email) - print('to {} added owner: {}'.format(id, owner_email)) + print('to {} added owner: {}'.format(calendar_id, owner_email)) -def remove_calendar(service, id): - calendar = GoogleCalendar(service, id) +def remove_calendar(service, calendar_id: str) -> None: + calendar = GoogleCalendar(service, calendar_id) calendar.delete() - print('removed: {}'.format(id)) + print('removed: {}'.format(calendar_id)) -def rename_calendar(service, id, summary): +def rename_calendar(service, calendar_id: str, summary: str) -> None: calendar = {'summary': summary} - service.calendars().patch(body=calendar, calendarId=id).execute() - print('{}: {}'.format(summary, id)) + service.calendars().patch(body=calendar, calendarId=calendar_id).execute() + print('{}: {}'.format(summary, calendar_id)) -def get_calendar_property(service, id, property): - response = service.calendarList().get(calendarId=id, - fields=property).execute() - print(response.get(property)) +def get_calendar_property(service, calendar_id: str, property_name: str) -> None: + response = service.calendarList().get(calendarId=calendar_id, + fields=property_name).execute() + print(response.get(property_name)) -def set_calendar_property(service, id, property, property_value): - body = {property: property_value} - response = service.calendarList().patch(body=body, calendarId=id).execute() +def set_calendar_property(service, calendar_id: str, property_name: str, property_value: str) -> None: + body = {property_name: property_value} + response = service.calendarList().patch(body=body, calendarId=calendar_id).execute() print(response) diff --git a/sync_ics2gcal/sync.py b/sync_ics2gcal/sync.py index bdc5b2f..475ad95 100644 --- a/sync_ics2gcal/sync.py +++ b/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): - self.gcalendar = gcalendar - self.converter = converter + 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]] = [] @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_dst = set(map(get_key, items_dst)) + keys_src: Set[str] = set(map(get_key, items_src)) + 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_date = None + event_start: Dict[str, str] = event['start'] + event_date: Union[datetime.date, datetime.datetime, str, None] = None compare_dates = False if 'date' in event_start: @@ -101,7 +117,7 @@ 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]) -> datetime.datetime: """make tz aware datetime from datetime/date (utc if no tzinfo) Arguments: @@ -117,7 +133,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 +151,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,14 +175,21 @@ class CalendarSync(): len(self.to_delete) ) - def apply(self): - """apply sync (insert, update, delete), using prepared lists of events + 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 """ self.gcalendar.insert_events(self.to_insert) 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') diff --git a/sync_ics2gcal/sync_calendar.py b/sync_ics2gcal/sync_calendar.py index cf39dc4..2e90b6a 100644 --- a/sync_ics2gcal/sync_calendar.py +++ b/sync_ics2gcal/sync_calendar.py @@ -1,3 +1,5 @@ +from typing import Dict, Any + import yaml import dateutil.parser @@ -12,14 +14,13 @@ from . import ( ) -def load_config(): +def load_config() -> Dict[str, Any]: with open('config.yml', 'r', encoding='utf-8') as f: result = yaml.safe_load(f) return result -def get_start_date(date_str): - result = datetime.datetime(1, 1, 1) +def get_start_date(date_str: str) -> datetime.datetime: if 'now' == date_str: result = datetime.datetime.utcnow() else: @@ -33,8 +34,8 @@ def main(): if 'logging' in config: logging.config.dictConfig(config['logging']) - calendarId = config['calendar']['google_id'] - ics_filepath = config['calendar']['source'] + calendarId: str = config['calendar']['google_id'] + ics_filepath: str = config['calendar']['source'] start = get_start_date(config['start_from']) diff --git a/tests/test_converter.py b/tests/test_converter.py index 2aadde6..37a9f97 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -1,3 +1,5 @@ +from typing import Tuple + import pytest from sync_ics2gcal import CalendarConverter @@ -26,11 +28,11 @@ LAST-MODIFIED:20180326T120235Z """ -def ics_test_cal(content): +def ics_test_cal(content: str) -> str: return "BEGIN:VCALENDAR\r\n{}END:VCALENDAR\r\n".format(content) -def ics_test_event(content): +def ics_test_event(content: str) -> str: return ics_test_cal("BEGIN:VEVENT\r\n{}END:VEVENT\r\n".format(content)) @@ -68,7 +70,7 @@ def param_events_start_end(request): return request.param -def test_event_start_end(param_events_start_end): +def test_event_start_end(param_events_start_end: Tuple[str, str, str, str]): (date_type, ics_str, start, end) = param_events_start_end converter = CalendarConverter() converter.loads(ics_str) diff --git a/tests/test_sync.py b/tests/test_sync.py index 3201929..f480260 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -3,6 +3,7 @@ import hashlib import operator from copy import deepcopy from random import shuffle +from typing import Union, List, Dict, Optional import dateutil.parser import pytest @@ -11,7 +12,7 @@ from pytz import timezone, utc from sync_ics2gcal import CalendarSync -def sha1(string): +def sha1(string: Union[str, bytes]) -> str: if isinstance(string, str): string = string.encode('utf8') h = hashlib.sha1() @@ -19,55 +20,57 @@ def sha1(string): return h.hexdigest() -def gen_events(start, stop, start_time, no_time=False): +def gen_events(start: int, + stop: int, + start_time: Union[datetime.datetime, datetime.date], + no_time: bool = False) -> List[Dict[str, Union[str, Dict[str, str]]]]: if no_time: start_time = datetime.date( start_time.year, start_time.month, start_time.day) - duration = datetime.date(1, 1, 2) - datetime.date(1, 1, 1) - date_key = "date" - suff = '' + duration: datetime.timedelta = datetime.date(1, 1, 2) - datetime.date(1, 1, 1) + date_key: str = "date" + date_end: str = '' else: start_time = utc.normalize( start_time.astimezone(utc)).replace(tzinfo=None) - duration = datetime.datetime( - 1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1) - date_key = "dateTime" - suff = 'Z' + duration: datetime.timedelta = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1) + date_key: str = "dateTime" + date_end: str = 'Z' - result = [] + result: List[Dict[str, Union[str, Dict[str, str]]]] = [] for i in range(start, stop): event_start = start_time + (duration * i) event_end = event_start + duration - updated = event_start + updated: Union[datetime.datetime, datetime.date] = event_start if no_time: updated = datetime.datetime( updated.year, updated.month, updated.day, 0, 0, 0, 1, tzinfo=utc) - event = { + event: Dict[str, Union[str, Dict[str, str]]] = { 'summary': 'test event __ {}'.format(i), 'location': 'la la la {}'.format(i), 'description': 'test TEST -- test event {}'.format(i), "iCalUID": "{}@test.com".format(sha1("test - event {}".format(i))), "updated": updated.isoformat() + 'Z', - "created": updated.isoformat() + 'Z' + "created": updated.isoformat() + 'Z', + 'start': {date_key: event_start.isoformat() + date_end}, + 'end': {date_key: event_end.isoformat() + date_end} } - event['start'] = {date_key: event_start.isoformat() + suff} - event['end'] = {date_key: event_end.isoformat() + suff} result.append(event) return result -def gen_list_to_compare(start, stop): - result = [] +def gen_list_to_compare(start: int, stop: int) -> List[Dict[str, str]]: + result: List[Dict[str, str]] = [] for i in range(start, stop): result.append({'iCalUID': 'test{:06d}'.format(i)}) return result -def get_start_date(event): - event_start = event['start'] - start_date = None +def get_start_date(event: Dict[str, Union[str, Dict[str, str]]]) -> Union[datetime.datetime, datetime.date]: + event_start: Dict[str, str] = event['start'] + start_date: Optional[str] = None is_date = False if 'date' in event_start: start_date = event_start['date'] @@ -113,7 +116,7 @@ def test_compare(): @pytest.mark.parametrize("no_time", [True, False], ids=['date', 'dateTime']) -def test_filter_events_by_date(no_time): +def test_filter_events_by_date(no_time: bool): msk = timezone('Europe/Moscow') now = utc.localize(datetime.datetime.utcnow()) msk_now = msk.normalize(now.astimezone(msk))