import logging from dataclasses import dataclass from datetime import datetime, timedelta from functools import total_ordering from typing import Iterator from caldav import Calendar from caldav.lib.error import ReportError from pydantic import BaseModel, validator from vobject.icalendar import VEvent from .async_helpers import get_ttl_hash, run_in_executor, timed_alru_cache from .dav_common import caldav_principal from .settings import SETTINGS _logger = logging.getLogger(__name__) def _string_strip(in_str: str) -> str: return in_str.strip() @total_ordering class CalEvent(BaseModel): summary: str = "" description: str = "" dtstart: datetime = datetime.utcnow() dtend: datetime = datetime.utcnow() class Config: frozen = True def __lt__(self, other: "CalEvent") -> bool: return self.dtstart < other.dtstart def __eq__(self, other: "CalEvent") -> bool: return self.dict() == other.dict() _validate_summary = validator( "summary", allow_reuse=True, )(_string_strip) _validate_description = validator( "description", allow_reuse=True, )(_string_strip) @classmethod def from_vevent(cls, event: VEvent) -> "CalEvent": data = {} for key in cls().dict().keys(): try: data[key] = event.contents[key][0].value except KeyError: pass return cls.parse_obj(data) @timed_alru_cache(maxsize=20) async def _get_calendar( calendar_name: str, ) -> Calendar: @run_in_executor def _inner() -> Calendar: return caldav_principal().calendar(calendar_name) return await _inner() @timed_alru_cache(maxsize=20) async def _get_calendar_events( calendar_name: str, ) -> list[CalEvent]: @run_in_executor def _inner() -> Iterator[VEvent]: _logger.info(f"updating {calendar_name!r} ...") calendar = caldav_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 + timedelta(days=365) try: search_result = calendar.date_search( start=dt_start, end=dt_end, expand=True, verify_expand=True, ) except ReportError: _logger.warn("CalDAV server does not support expanded search") search_result = calendar.date_search( start=dt_start, end=dt_end, expand=False, ) return ( vevent for event in search_result for vevent in event.vobject_instance.contents["vevent"] ) return sorted([ CalEvent.from_vevent(vevent) for vevent in await _inner() ]) @dataclass(frozen=True) class DavCalendar: calendar_name: str @property async def calendar(self) -> Calendar: return await _get_calendar( ttl_hash=get_ttl_hash(SETTINGS.cache_seconds), calendar_name=self.calendar_name, ) @property async def events(self) -> list[CalEvent]: return await _get_calendar_events( ttl_hash=get_ttl_hash(SETTINGS.cache_seconds), calendar_name=self.calendar_name, )