import functools import logging from datetime import datetime, timedelta from typing import cast from asyncify import asyncify from cachetools import TTLCache, cachedmethod from caldav import Calendar, DAVClient, Event, Principal from vobject.base import Component, toVName from .calevent import CalEvent from .config import Config from .settings import SETTINGS from .webdav import davkey _logger = logging.getLogger(__name__) class CalDAV: _caldav_client = DAVClient( url=SETTINGS.caldav.url, username=SETTINGS.caldav.username, password=SETTINGS.caldav.password, ) _cache = TTLCache( ttl=SETTINGS.caldav.cache_ttl, maxsize=SETTINGS.caldav.cache_size, ) @classmethod @property def principal(cls) -> Principal: """ Gets the `Principal` object of the main CalDAV client. """ return cls._caldav_client.principal() @classmethod @property @asyncify @cachedmethod( cache=lambda cls: cls._cache, key=functools.partial(davkey, "calendars"), ) def calendars(cls) -> list[str]: """ Asynchroneously lists all calendars using the main WebDAV client. """ _logger.debug("calendars") return [str(cal.name) for cal in cls.principal.calendars()] @classmethod @asyncify @cachedmethod( cache=lambda cls: cls._cache, key=functools.partial(davkey, "get_calendar"), ) def get_calendar(cls, calendar_name: str) -> Calendar: """ Get a calendar by name using the CalDAV principal object. """ return cls.principal.calendar(calendar_name) @classmethod @asyncify @cachedmethod( cache=lambda cls: cls._cache, key=functools.partial(davkey, "get_events"), ) def get_events(cls, calendar_name: str, cfg: Config) -> list[CalEvent]: """ Get a sorted list of events by CalDAV calendar name. """ _logger.info(f"downloading {calendar_name!r} ...") dt_start = datetime.combine( datetime.now().date(), datetime.min.time(), ) dt_end = dt_start + timedelta(days=cfg.calendar.future_days) search_result = cls.principal.calendar(calendar_name).search( start=dt_start, end=dt_end, expand=True, comp_class=Event, split_expanded=False, ) vevents = [] for event in search_result: vobject = cast(Component, event.vobject_instance) vevents.extend(vobject.contents[toVName("vevent")]) return sorted(CalEvent.from_vevent(vevent) for vevent in vevents)