import logging from dataclasses import dataclass from datetime import datetime, timedelta from typing import Iterator from caldav import Calendar from pydantic import BaseModel from vobject.icalendar import VEvent from . import caldav_principal, get_ttl_hash, run_in_executor, timed_alru_cache _logger = logging.getLogger(__name__) class CalEvent(BaseModel): summary: str description: str dtstart: datetime dtend: datetime @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}") print(f"updating {calendar_name}") calendar = caldav_principal().calendar(calendar_name) return ( vevent for event in calendar.date_search( start=datetime.now(), end=datetime.now() + timedelta(days=365), expand=True, ) for vevent in event.vobject_instance.contents["vevent"] ) return [ CalEvent( summary=vevent.summary.value, description=vevent.description.value, dtstart=vevent.dtstart.value, dtend=vevent.dtend.value, ) 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(20), calendar_name=self.calendar_name, ) @property async def events(self) -> list[CalEvent]: return await _get_calendar_events( ttl_hash=get_ttl_hash(20), calendar_name=self.calendar_name, )