ovdashboard/api/ovdashboard_api/__init__.py

import asyncio
import functools
import time
from typing import Any

import caldav
from async_lru import alru_cache
from webdav3 import client as WebDAVclient

from .config import SETTINGS

_WEBDAV_CLIENT = WebDAVclient.Client({
    "webdav_hostname": SETTINGS.webdav_url,
    "webdav_login":    SETTINGS.dav_username,
    "webdav_password": SETTINGS.dav_password,
})


def run_in_executor(f):
    """
    Decorator to make blocking function call asyncio compatible
    https://stackoverflow.com/questions/41063331/how-to-use-asyncio-with-existing-blocking-library/
    """

    @functools.wraps(f)
    def inner(*args, **kwargs):
        loop = asyncio.get_running_loop()
        return loop.run_in_executor(
            None,
            functools.partial(f, *args, **kwargs),
        )

    return inner


def get_ttl_hash(seconds: int = 20) -> int:
    """
    Return the same value within `seconds` time period
    https://stackoverflow.com/a/55900800
    """
    return round(time.time() / seconds)


def timed_alru_cache(**decorator_kwargs):
    def decorate(f):
        @alru_cache(**decorator_kwargs)
        @functools.wraps(f)
        async def wrapper(ttl_hash: int, *args, **kwargs):
            del ttl_hash

            return await f(*args, **kwargs)

        return wrapper

    return decorate


@functools.lru_cache
def webdav_resource(remote_path: Any) -> WebDAVclient.Resource:
    return _WEBDAV_CLIENT.resource(remote_path)


@run_in_executor
def webdav_list(remote_path: str) -> list:
    return _WEBDAV_CLIENT.list(remote_path)


_CALDAV_CLIENT = caldav.DAVClient(
    url=SETTINGS.caldav_url,
    username=SETTINGS.dav_username,
    password=SETTINGS.dav_password,
)


def caldav_principal() -> caldav.Principal:
    return _CALDAV_CLIENT.principal()