ovdashboard/api/ovdashboard_api/core/caldav.py

import logging
from datetime import datetime, timedelta

from asyncify import asyncify
from cache import AsyncTTL
from caldav import Calendar, DAVClient, Principal
from caldav.lib.error import ReportError
from vobject.base import Component

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.
        """

        _logger.debug("principal")
        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()]

    @AsyncTTL(
        time_to_live=SETTINGS.caldav.cache_ttl,
        maxsize=SETTINGS.caldav.cache_size,
        skip_args=1,
    )
    @asyncify
    @classmethod
    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
    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} ...")

        search_span = timedelta(days=cfg.calendar.future_days)
        calendar = cls.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.warning("CalDAV server does not support expanded search")

            search_result = calendar.date_search(
                start=dt_start,
                end=dt_end,
                expand=False,
            )

        vevents = []
        for event in search_result:
            vobject: Component = event.vobject_instance  # type: ignore
            vevents.extend(vobject.vevent_list)

        return sorted(CalEvent.from_vevent(vevent) for vevent in vevents)