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, Event, Principal
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.
        """

        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()]

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

        search_result = calendar.search(
            start=dt_start,
            end=dt_end,
            expand=True,
            comp_class=Event,
            split_expanded=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)