96 lines
2.5 KiB
Python
96 lines
2.5 KiB
Python
import logging
|
|
from datetime import datetime, timedelta
|
|
from typing import cast
|
|
|
|
from asyncify import asyncify
|
|
from cache import AsyncTTL
|
|
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
|
|
|
|
_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} ...")
|
|
|
|
dt_start = datetime.combine(
|
|
datetime.utcnow().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)
|