masto-bridges/masto_bridges/caldav.py

100 lines
3.1 KiB
Python
Raw Permalink Normal View History

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
)