import logging from datetime import datetime, timedelta from asyncify import asyncify from cache import AsyncTTL from caldav import Calendar, DAVClient, Principal from caldav.lib.error import ReportError from vobject.base import Component from .calevent import CalEvent from .config import Config from .settings import SETTINGS _logger = logging.getLogger(__name__) class CalDAV: _caldav_client = DAVClient( url=SETTINGS.caldav.url, username=SETTINGS.caldav.username, password=SETTINGS.caldav.password, ) @classmethod @property def principal(cls) -> Principal: """ Gets the `Principal` object of the main CalDAV client. """ _logger.debug("principal") return cls._caldav_client.principal() @classmethod @property @AsyncTTL( time_to_live=SETTINGS.caldav.cache_ttl, maxsize=SETTINGS.caldav.cache_size, skip_args=1, ) @asyncify 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 @AsyncTTL( time_to_live=SETTINGS.caldav.cache_ttl, maxsize=SETTINGS.caldav.cache_size, skip_args=1, ) @asyncify 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 @AsyncTTL( time_to_live=SETTINGS.caldav.cache_ttl, maxsize=SETTINGS.caldav.cache_size, skip_args=1, ) @asyncify 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} ...") search_span = timedelta(days=cfg.calendar.future_days) calendar = cls.principal.calendar(calendar_name) date_start = datetime.utcnow().date() time_min = datetime.min.time() dt_start = datetime.combine(date_start, time_min) dt_end = dt_start + search_span try: search_result = calendar.date_search( start=dt_start, end=dt_end, expand=True, verify_expand=True, ) except ReportError: _logger.warning("CalDAV server does not support expanded search") search_result = calendar.date_search( start=dt_start, end=dt_end, expand=False, ) vevents = [] for event in search_result: vobject: Component = event.vobject_instance # type: ignore vevents.extend(vobject.vevent_list) return sorted(CalEvent.from_vevent(vevent) for vevent in vevents)