from typing import Optional from caldav import DAVClient, Calendar from caldav.objects import SynchronizableCalendarObjectCollection # from caldav.lib.error import ResponseError from pydantic import BaseModel, Field from datetime import datetime from icalendar import Event as iEvent from uuid import uuid1 from masto_bridges.config import Config class Event(BaseModel): summary: str # the title of the event, use for username dtstart: datetime dtend: datetime uid: str = Field(default_factory=uuid1) url: str = '' location:str = '' description:str = '' # post body def to_ical(self) -> str: evt = iEvent() evt.add('dtstart', self.dtstart) evt.add('dtend', self.dtend) evt.add('uid', self.uid) evt.add('location', self.location) evt.add('summary', self.summary) evt.add('description', self.description) evt.add('url', self.url) return evt.to_ical() class CalDAV: def __init__(self, config:Config): self.config = config for value in ('CALDAV_URL', 'CALDAV_USER', 'CALDAV_PASSWORD', 'CALDAV_CALENDAR_NAME'): if getattr(self.config, value) is None: raise ValueError('Need all CALDAV fields to be present in config! Check your .env file!') self._sync_token = None self._sync = None # type: Optional[SynchronizableCalendarObjectCollection] def get_calendar(self, name:str) -> Optional[Calendar]: """ Get a calendar by its name, creating one if none exists. """ with self.client as client: cals = client.principal().calendars() # try and find cal = [cal for cal in cals if cal.name == name] if len(cal) == 0: cal = client.principal().make_calendar(name=name) else: cal = cal[0] return cal def post(self, event:Event) -> bool: with self.client as client: calendar = self.get_calendar(self.config.CALDAV_CALENDAR_NAME) calendar.save_event( **event.dict() ) return True def poll(self) -> SynchronizableCalendarObjectCollection: cal = self.get_calendar(self.config.CALDAV_CALENDAR_NAME) if self._sync_token is None: # we haven't polled yet. get and ignore previous entries. # we don't want to post a bunch of junk every time lol self._sync = cal.objects(load_objects=False) self._sync_token = self._sync.sync_token try: self._sync = cal.objects(sync_token=self._sync_token, load_objects=True) self._sync_token = self._sync.sync_token except: # deleting calendar entries makes us choke! self._sync = cal.objects(sync_token=self._sync_token, load_objects=False) self._sync_token = self._sync.sync_token return self._sync @property def client(self) -> DAVClient: return DAVClient( url=self.config.CALDAV_URL, username=self.config.CALDAV_USER, password=self.config.CALDAV_PASSWORD )