ovdashboard/api/ovdashboard_api/dav_calendar.py

"""
Definition of an asyncio compatible CalDAV calendar.

Caches events using `timed_alru_cache`.
"""

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 .config import Config
from .dav_common import caldav_principal
from .settings import SETTINGS

_logger = logging.getLogger(__name__)


def _string_strip(in_str: str) -> str:
    """
    Wrapper for str.strip().

    Used to define `pydantic` validators.
    """
    return in_str.strip()


@total_ordering
class CalEvent(BaseModel):
    """
    A CalDAV calendar event.

    Properties are to be named as in the EVENT component of
    RFC5545 (iCalendar).

    https://icalendar.org/iCalendar-RFC-5545/3-6-1-event-component.html
    """

    summary: str = ""
    description: str = ""
    dtstart: datetime = datetime.utcnow()
    dtend: datetime = datetime.utcnow()

    class Config:
        frozen = True

    def __lt__(self, other: "CalEvent") -> bool:
        """
        Order Events by start time.
        """

        return self.dtstart < other.dtstart

    def __eq__(self, other: "CalEvent") -> bool:
        """
        Compare all properties.
        """

        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":
        """
        Create a CalEvent instance from a `VObject.VEvent` object.
        """

        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=SETTINGS.cache_size)
async def _get_calendar(
    calendar_name: str,
) -> Calendar:
    """
    Get a calendar by name using the CalDAV principal object.
    """

    @run_in_executor
    def _inner() -> Calendar:
        return caldav_principal().calendar(calendar_name)

    return await _inner()


@timed_alru_cache(maxsize=SETTINGS.cache_size)
async def _get_calendar_events(
    calendar_name: str,
) -> list[CalEvent]:
    """
    Get a sorted list of events by CalDAV calendar name.

    Do not return an iterator here - this result is cached and
    an iterator would get consumed.
    """

    cfg = await Config.get()
    search_span = timedelta(days=cfg.calendar.future_days)

    @run_in_executor
    def _inner() -> Iterator[VEvent]:
        """
        Get events by CalDAV calendar name.

        This can return an iterator - only the outer function is
        cached.
        """
        _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 + search_span

        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:
    """
    Object representation of a CalDAV calendar.
    """

    calendar_name: str

    @property
    async def calendar(self) -> Calendar:
        """
        Calendar as `caldav` library representation.
        """

        return await _get_calendar(
            ttl_hash=get_ttl_hash(SETTINGS.cache_seconds),
            calendar_name=self.calendar_name,
        )

    @property
    async def events(self) -> list[CalEvent]:
        """
        Calendar events in object representation.
        """

        return await _get_calendar_events(
            ttl_hash=get_ttl_hash(SETTINGS.cache_seconds),
            calendar_name=self.calendar_name,
        )