ovdashboard/api/ovdashboard_api/core/dav/caldav.py

import functools
import logging
from datetime import datetime, timedelta
from typing import cast

from asyncify import asyncify
from cachetools import TTLCache, cachedmethod
from caldav import Calendar, DAVClient, Event, Principal
from vobject.base import Component, toVName

from ..calevent import CalEvent
from ..config import Config
from ..settings import SETTINGS
from .webdav import davkey

_logger = logging.getLogger(__name__)


class CalDAV:
    _caldav_client = DAVClient(
        url=SETTINGS.caldav.url,
        username=SETTINGS.caldav.username,
        password=SETTINGS.caldav.password,
    )

    _cache = TTLCache(
        ttl=SETTINGS.caldav.cache_ttl,
        maxsize=SETTINGS.caldav.cache_size,
    )

    @classmethod
    @property
    def principal(cls) -> Principal:
        """
        Gets the `Principal` object of the main CalDAV client.
        """

        return cls._caldav_client.principal()

    @classmethod
    @property
    @asyncify
    @cachedmethod(
        cache=lambda cls: cls._cache,
        key=functools.partial(davkey, "calendars"),
    )
    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
    @asyncify
    @cachedmethod(
        cache=lambda cls: cls._cache,
        key=functools.partial(davkey, "get_calendar"),
    )
    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
    @asyncify
    @cachedmethod(
        cache=lambda cls: cls._cache,
        key=functools.partial(davkey, "get_events"),
    )
    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} ...")

        dt_start = datetime.combine(
            datetime.now().date(),
            datetime.min.time(),
        )
        dt_end = dt_start + timedelta(days=cfg.calendar.future_days)

        search_result = cls.principal.calendar(calendar_name).search(
            start=dt_start,
            end=dt_end,
            expand=True,
            comp_class=Event,
            split_expanded=False,
        )

        vevents = []
        for event in search_result:
            vobject = cast(Component, event.vobject_instance)
            vevents.extend(vobject.contents[toVName("vevent")])

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