ovdashboard/api/ovdashboard_api/dav_common.py

"""
Definition of WebDAV and CalDAV clients.
"""

from functools import lru_cache
from logging import getLogger
from time import sleep
from typing import Any, Iterator

from caldav import DAVClient as CalDAVclient
from caldav import Principal as CalDAVPrincipal
from webdav3.client import Client as WebDAVclient
from webdav3.client import Resource as WebDAVResource

from .async_helpers import run_in_executor
from .settings import SETTINGS

_WEBDAV_CLIENT = WebDAVclient({
    "webdav_hostname": SETTINGS.webdav.url,
    "webdav_login":    SETTINGS.webdav.username,
    "webdav_password": SETTINGS.webdav.password,
})

_logger = getLogger(__name__)


def webdav_check() -> bool:
    """
    Checks if base resources are available.
    """

    _logger.info(
        "Production mode is %s.",
        "enabled" if SETTINGS.production_mode else "disabled",
    )

    if SETTINGS.production_mode:
        for _ in range(SETTINGS.webdav_retries):
            if _WEBDAV_CLIENT.check(""):
                break

            _logger.warning(
                "Waiting for WebDAV connection to %s ...",
                repr(SETTINGS.webdav.url),
            )
            sleep(30)

        _logger.debug("WebDAV connection ok.")

    elif not _WEBDAV_CLIENT.check(""):
        _logger.error(
            "WebDAV connection to %s FAILED!",
            repr(SETTINGS.webdav.url),
        )
        return False

    _logger.debug("WebDAV connection ok.")

    if not _WEBDAV_CLIENT.check(SETTINGS.webdav_prefix):
        _logger.error(
            "WebDAV prefix directory %s NOT FOUND, please create it!",
            repr(SETTINGS.webdav_prefix),
        )
        return False

    _logger.debug("WebDAV prefix directory found.")

    return True


@lru_cache(maxsize=SETTINGS.cache_size)
def webdav_resource(remote_path: Any) -> WebDAVResource:
    """
    Gets a resource using the main WebDAV client.
    """

    return _WEBDAV_CLIENT.resource(
        f"{SETTINGS.webdav_prefix}/{remote_path}"
    )


@run_in_executor
def webdav_list(remote_path: str) -> list[str]:
    """
    Asynchronously lists a WebDAV path using the main WebDAV client.
    """

    return _WEBDAV_CLIENT.list(
        f"{SETTINGS.webdav_prefix}/{remote_path}"
    )


_CALDAV_CLIENT = CalDAVclient(
    url=SETTINGS.caldav.url,
    username=SETTINGS.caldav.username,
    password=SETTINGS.caldav.password,
)


def caldav_principal() -> CalDAVPrincipal:
    """
    Gets the `Principal` object of the main CalDAV client.
    """

    return _CALDAV_CLIENT.principal()


@run_in_executor
def caldav_list() -> Iterator[str]:
    """
    Asynchronously lists all calendars using the main WebDAV client.
    """

    return (
        str(cal.name)
        for cal in caldav_principal().calendars()
    )