ovdashboard/api/ovdashboard_api/dav_calendar.py

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

_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(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,
        )