import logging from datetime import datetime, timedelta from typing import cast from asyncify import asyncify from cache import AsyncTTL 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 _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. """ 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} ...") dt_start = datetime.combine( datetime.utcnow().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)