caldav_monkey_patch_for_google_sync_via_ctag_etag.py

"""adapters/extensions to caldav lib for incremental (old-fashioned) sync https://github.com/python-caldav/caldav/issues/69
does monkey-patching on import!!

mostly by guide http://sabre.io/dav/building-a-caldav-client/
"""

from caldav.objects import *
from caldav.objects import _fix_tz


# PATCH namespaces -- add CalendarServer
from caldav.lib.namespace import nsmap, nsmap2, ns
nsmap2["CS"] = nsmap["CS"] = "http://calendarserver.org/ns/"

from caldav.elements.base import ValuedBaseElement, BaseElement

class GetCtag(ValuedBaseElement): # by analogy to GetEtag
    tag = ns("CS", "getctag")

class CalendarMultiget(BaseElement):  # by analogy to CalendarQuery
    tag = ns("C", "calendar-multiget")


class PATCHED_CLASSES:
    class DAVObject:
        def children(self, type=None, extra_props=None):
            """
            List children, using a propfind (resourcetype) on the parent object,
            at depth = 1.
            """
            c = []

            depth = 1
            properties = {}

            props = [dav.ResourceType(), dav.DisplayName()]
            if extra_props:
                props.extend(extra_props)

            response = self._query_properties(props, depth)
            properties = self._handle_prop_response(
                response=response, props=props, type=type, what='tag')

            for path in list(properties.keys()):
                resource_type = properties[path][dav.ResourceType.tag]
                resource_name = properties[path][dav.DisplayName.tag]
                extra_data = [
                    properties[path][x.tag] for x in extra_props
                ]

                if resource_type == type or type is None:
                    # TODO: investigate the RFCs thoroughly - why does a "get
                    # members of this collection"-request also return the
                    # collection URL itself?
                    # And why is the strip_trailing_slash-method needed?
                    # The collection URL should always end with a slash according
                    # to RFC 2518, section 5.2.
                    if (self.url.strip_trailing_slash() !=
                            self.url.join(path).strip_trailing_slash()):
                        c.append((self.url.join(path), resource_type,
                                  resource_name,
                                  ) + tuple(extra_data) )

            return c


    class CalendarSet:
        def calendars(self):
            """
            List all calendar collections in this set.

            Returns:
             * [Calendar(), ...]
            """
            cals = []

            data = self.children(cdav.Calendar.tag, extra_props=[GetCtag()])
            for c_url, c_type, c_name, c_ctag in data:
                cal = Calendar4SyncByCtagEtag(self.client, c_url, parent=self, name=c_name, ctag=c_ctag)
                cals.append(cal)

            return cals

# Monkey-Patching  #TODO?  maybe move to function? -- well but would be no worth importing if not doing patch..
from caldav import DAVObject, CalendarSet
DAVObject.children = PATCHED_CLASSES.DAVObject.children
CalendarSet.calendars = PATCHED_CLASSES.CalendarSet.calendars


class EventWithEtag(Event):
    def __init__(self, *args, etag=None, **kwargs):
        super().__init__(*args, **kwargs)
        assert etag is not None, "etag should not be None: {}".format(etag)
        self.etag = etag
        # hook into vevent objects -- as they will be serialized to msgraph repr
        if hasattr(self.vobject_instance, 'vevent'):
            for vevent in self.vobject_instance.vevent_list:
                vevent._etag = etag

    @property
    def uid(self):
        _etc, uid = self.url.path[:-len('.ics')].rsplit('/', 1)
        return uid



class Calendar4SyncByCtagEtag(Calendar):
    """
    supports having ctag for itself, and fetching etag for its events
    """
    # def __init__(self, client=None, url=None, parent=None, name=None, id=None,
    #              **extra):
    def __init__(self, *args, ctag, **kwargs):
        super().__init__(*args, **kwargs)
        self.ctag = ctag

    @property
    def resource_email(self):
        """extracts resource email from canonical_url
        https://apidata.googleusercontent.com/caldav/v2/[email protected]/events/ --> [email protected]
        """
        return self.canonical_url.rstrip('/').split('/')[-2]


    def date_search(self, start, end=None, expand="maybe", get_calendar_data=True):
        """
        Search events by date in the calendar. Recurring events are
        expanded if they are occuring during the specified time frame
        and if an end timestamp is given.

        Parameters:
         * start = datetime.today().
         * end = same as above.
         * compfilter = defaults to events only.  Set to None to fetch all
           calendar components.
         * expand - should recurrent events be expanded?  (to preserve
           backward-compatibility the default "maybe" will be changed into True
           unless the date_search is open-ended)

        Returns:
         * [CalendarObjectResource(), ...]

        """
        compfilter = "VEVENT"
        matches = []

        # build the request

        # fix missing tzinfo in the start and end datetimes
        start = _fix_tz(start)
        end = end and _fix_tz(end)

        ## for backward compatibility - expand should be false
        ## in an open-ended date search, otherwise true
        if expand == 'maybe':
            expand = end

        # Some servers will raise an error if we send the expand flag
        # but don't set any end-date - expand doesn't make much sense
        # if we have one recurring event describing an indefinite
        # series of events.  I think it's appropriate to raise an error
        # in this case.
        if not end and expand:
            raise error.ReportError("an open-ended date search cannot be expanded")
        elif expand:
            data = cdav.CalendarData() + cdav.Expand(start, end)
        else:
            data = cdav.CalendarData()
        prop = dav.Prop()
        prop = prop + dav.GetEtag()  # etag
        # same: prop.children.append(dav.GetEtag())
        if get_calendar_data:
            prop = prop + data

        query = cdav.TimeRange(start, end)
        if compfilter:
            query = cdav.CompFilter(compfilter) + query
        vcalendar = cdav.CompFilter("VCALENDAR") + query
        filter = cdav.Filter() + vcalendar

        root = cdav.CalendarQuery() + [prop, filter]
        response = self._query(root, 1, 'report')
        results = self._handle_prop_response(
            response=response, props=prop.children)
        for r in sorted(results.keys()): # a workaround for UT to keep consistent sequence of hrefs in list
            event = EventWithEtag(self.client, url=self.url.join(r), # FixMe: https://github.com/python-caldav/caldav/issues/70
                      data=results[r].get(cdav.CalendarData.tag), parent=self,
                      etag=results[r][dav.GetEtag.tag].strip('"')  # inject etag info -- strip quotes
            )
            matches.append(event)

        return matches


    def events(self):
        """
        List all events from the calendar. -- TODO: seems like narrow case of date_search

        Returns:
         * [Event(), ...]
        """
        all = []

        data = cdav.CalendarData()
        prop = dav.Prop() + [dav.GetEtag(), data]
        vevent = cdav.CompFilter("VEVENT")
        vcalendar = cdav.CompFilter("VCALENDAR") + vevent
        filter = cdav.Filter() + vcalendar
        root = cdav.CalendarQuery() + [prop, filter]

        response = self._query(root, 1, query_method='report')
        results = self._handle_prop_response(
            response, props=[cdav.CalendarData()])
        for r in sorted(results.keys()):
            all.append(Event(
                self.client, url=self.url.join(r),
                data=results[r][cdav.CalendarData.tag], parent=self))

        return all


    def multiget(self, req_events: list):
        """fetches many events by their url's list"""
        data = cdav.CalendarData()
        prop = dav.Prop() + [dav.GetEtag(), data]
        #
        # vevent = cdav.CompFilter("VEVENT")
        # vcalendar = cdav.CompFilter("VCALENDAR") + vevent
        # filter = cdav.Filter() + vcalendar

        hrefs = [dav.Href(value=x.url.path)  for x in req_events]

        root = CalendarMultiget() + [prop] + hrefs # .events() diff: root = cdav.CalendarQuery() + [prop, filter]
        response = self._query(root, 1, 'report')

        # from `date search`
        results = self._handle_prop_response(response=response, props=prop.children)

        matches = []
        for r in sorted(results.keys()):
            event = EventWithEtag(self.client, url=self.url.join(r),
                      data=results[r].get(cdav.CalendarData.tag), parent=self,
                      etag=results[r][dav.GetEtag.tag].strip('"')  # inject etag info -- strip quotes
            )
            matches.append(event)

        return matches

    # def object_by_uid(self, uid, comp_filter=None): # TODO: might need etag
    #     """
    #     NOTE: does not work with Google CALDAV:  https://stackoverflow.com/questions/36749307/google-caldav-ignoring-filter-options/
    #     Get one event from the calendar -- better base on `object_by_url`.

    #     """
if __name__ == "__main__":
    import caldav
    class GoogleCalDAVClient(caldav.DAVClient):
        caldav_url_base = "https://apidata.googleusercontent.com/caldav/v2"

        def __init__(self):
            super().__init__(url=self.caldav_url_base)

        def request(self, url, method="GET", body="", headers={}):
            self.inject_google_access_token()
            return super().request(url, method, body, headers)

        def inject_google_access_token(self):
            """`impersonated_user` required before calling this"""
            self.headers = self.headers or {}
            from googleapiclient.discovery import build
            service = build('calendar', 'v3', credentials=self.google_calendar_api_credentials() 
                                #for service accounts .with_subject(self.impersonated_user)
                            )

            # hacky injection of header "authorization":"Bearer <access_token>" using google (calendar) api lib
            service._http.credentials.before_request(
                request=service._http._request, method=None, url=None,
                headers=self.headers
            )

        __CACHED_CREDS = None

        def google_calendar_api_credentials(self):
            from google.auth.transport.requests import Request
            from google.oauth2 import service_account

            creds = self.__class__.__CACHED_CREDS

            # If there are no (valid) credentials available, let the service account re-authentificate.
            if not creds or not creds.valid:
                if creds and creds.expired and creds.refresh_token:
                    creds.refresh(Request())
                else:
                    SCOPES = ['https://www.googleapis.com/auth/calendar']

                    creds = service_account.Credentials.from_service_account_file(
                        '/path/to/credentials_client_id.json',
                        scopes=SCOPES )

                    # if we get info from .env --
                    #
                    # creds = service_account.Credentials.from_service_account_info(
                    #     dict_info,
                    #     scopes=SCOPES)

                # cache credentials
                self.__class__.__CACHED_CREDS = creds

            return creds





    caldav_url_base = "https://apidata.googleusercontent.com/caldav/v2"

    # client = caldav.DAVClient(
    client = GoogleCalDAVClient(
        # caldav_uri,
        caldav_url_base,
        # username="user",
        # password="passssssssss",
    )

    # from . import Calendar4SyncByCtagEtag  # should go before calling principal

    # https://gist.github.com/e1781b3c7de74d0ec80cc027adedbd0d

    principal = client.principal()
    calendars = principal.calendars()

    cal = calendars[0]

    now = datetime.datetime.utcnow()
    start = now - datetime.timedelta(days=7)
    # end = now + datetime.timedelta(days=30*5)
    end = None

    # events = cal.date_search(start, end)
    events = Calendar4SyncByCtagEtag.date_search(cal, start, end)

    print("\nEach event")
    for event in events[:1]: # limit to one
        e = event.instance.vevent
        print(event.etag, "(" + e.dtstart.value.strftime("%H:%M") + ") " , e.summary.value, e.location.value)