b4tman
/
sync_ics2gcal
mirror of https://github.com/b4tman/sync_ics2gcal
1
0
Fork 0
Code Issues Releases Activity

type aliases

This commit is contained in:
Dmitry Belyaev 2021-05-19 12:55:54 +03:00
parent 5e01b6dd01
commit 052ba440d0
Signed by: b4tman
GPG Key ID: 41A00BF15EA7E5F3
4 changed files with 53 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
sync_ics2gcal/__init__.py
View File

@ -1,12 +1,16 @@
from .ical import (
CalendarConverter,
EventConverter
EventConverter,
DateDateTime
)
from .gcal import (
GoogleCalendarService,
GoogleCalendar
GoogleCalendar,
EventData,
EventList,
EventTuple
)
from .sync import (

25
sync_ics2gcal/gcal.py
View File

@ -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:

18
sync_ics2gcal/ical.py
View File

@ -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
"""

42
sync_ics2gcal/sync.py
View File

@ -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: