172 lines
4.2 KiB
Python
172 lines
4.2 KiB
Python
"""
|
|
Definition of WebDAV and CalDAV clients.
|
|
"""
|
|
|
|
from functools import lru_cache
|
|
from logging import getLogger
|
|
from os import path
|
|
from pathlib import Path
|
|
from time import sleep
|
|
from typing import Any, Iterator
|
|
|
|
from asyncify import asyncify
|
|
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 . import __file__ as OVD_INIT
|
|
from .core.settings import SETTINGS
|
|
|
|
_WEBDAV_CLIENT = WebDAVclient(
|
|
{
|
|
"webdav_hostname": SETTINGS.webdav.url,
|
|
"webdav_login": SETTINGS.webdav.username,
|
|
"webdav_password": SETTINGS.webdav.password,
|
|
"disable_check": SETTINGS.webdav_disable_check,
|
|
}
|
|
)
|
|
|
|
_logger = getLogger(__name__)
|
|
|
|
|
|
def webdav_check() -> None:
|
|
"""
|
|
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),
|
|
)
|
|
raise ConnectionError(SETTINGS.webdav.url)
|
|
|
|
_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),
|
|
)
|
|
raise FileNotFoundError(SETTINGS.webdav_prefix)
|
|
|
|
_logger.debug("WebDAV prefix directory found.")
|
|
|
|
|
|
def webdav_ensure_path(remote_path: str) -> bool:
|
|
remote_path = f"{SETTINGS.webdav_prefix}/{remote_path}"
|
|
|
|
if _WEBDAV_CLIENT.check(remote_path):
|
|
_logger.debug(
|
|
"WebDAV path %s found.",
|
|
repr(remote_path),
|
|
)
|
|
return True
|
|
|
|
_logger.info(
|
|
"WebDAV path %s not found, creating ...",
|
|
repr(remote_path),
|
|
)
|
|
_WEBDAV_CLIENT.mkdir(remote_path)
|
|
|
|
return False
|
|
|
|
|
|
def get_skel_path(skel_file: str) -> Path:
|
|
skel_path = path.dirname(Path(OVD_INIT).absolute())
|
|
return Path(skel_path).joinpath("skel", skel_file)
|
|
|
|
|
|
def webdav_upload_skel(remote_path: str, *skel_files: str) -> None:
|
|
remote_path = f"{SETTINGS.webdav_prefix}/{remote_path}"
|
|
|
|
for skel_file in skel_files:
|
|
_logger.debug(
|
|
"Creating WebDAV file %s ...",
|
|
repr(skel_file),
|
|
)
|
|
|
|
_WEBDAV_CLIENT.upload_file(
|
|
f"{remote_path}/{skel_file}",
|
|
get_skel_path(skel_file),
|
|
)
|
|
|
|
|
|
def webdav_ensure_files(remote_path: str, *file_names: str) -> None:
|
|
missing_files = (
|
|
file_name
|
|
for file_name in file_names
|
|
if not _WEBDAV_CLIENT.check(
|
|
path.join(
|
|
SETTINGS.webdav_prefix,
|
|
remote_path,
|
|
file_name,
|
|
)
|
|
)
|
|
)
|
|
|
|
webdav_upload_skel(
|
|
remote_path,
|
|
*missing_files,
|
|
)
|
|
|
|
|
|
@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}")
|
|
|
|
|
|
@asyncify
|
|
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()
|
|
|
|
|
|
@asyncify
|
|
def caldav_list() -> Iterator[str]:
|
|
"""
|
|
Asynchronously lists all calendars using the main WebDAV client.
|
|
"""
|
|
|
|
return (str(cal.name) for cal in caldav_principal().calendars())
|