Source code for galaxy.files.sources.webdav

from typing import (
    Annotated,
    Optional,
    Union,
)

from pydantic import (
    Field,
)

from galaxy.files.models import FilesSourceRuntimeContext
from galaxy.util.config_templates import TemplateExpansion
from ._fsspec import (
    CacheOptionsDictType,
    FsspecBaseFileSourceConfiguration,
    FsspecBaseFileSourceTemplateConfiguration,
    FsspecFilesSource,
)

try:
    from webdav4.fsspec import WebdavFileSystem
except ImportError:
    WebdavFileSystem = None


class WebDavFileSourceTemplateConfiguration(FsspecBaseFileSourceTemplateConfiguration):
    root: Optional[Union[str, TemplateExpansion]] = None
    base_url: Union[str, TemplateExpansion]
    login: Optional[Union[str, TemplateExpansion]] = None
    password: Optional[Union[str, TemplateExpansion]] = None


class WebDavFileSourceConfiguration(FsspecBaseFileSourceConfiguration):
    root: Optional[str] = None
    base_url: Annotated[
        str,
        Field(
            title="WebDAV base URL",
            description="The fully-qualified WebDAV endpoint URL used to access this file source.",
        ),
    ]
    login: Optional[str] = None
    password: Optional[str] = None



[docs]
class WebDavFilesSource(FsspecFilesSource[WebDavFileSourceTemplateConfiguration, WebDavFileSourceConfiguration]):
    plugin_type = "webdav"
    required_module = WebdavFileSystem
    required_package = "webdav4"

    template_config_class = WebDavFileSourceTemplateConfiguration
    resolved_config_class = WebDavFileSourceConfiguration

    @staticmethod
    def _webdav_endpoint(base_url: str, root: Optional[str]) -> str:
        # WebDAV "root" is the service endpoint path (for example Nextcloud's
        # /remote.php/dav/files/user), not a directory prefix inside the file source.
        base_url = base_url.strip().rstrip("/")
        if not base_url:
            raise ValueError("base_url is required for WebDAV file source")
        root = root.strip().strip("/") if root else ""
        if root:
            return f"{base_url}/{root}"
        return base_url

    def _open_fs(
        self,
        context: FilesSourceRuntimeContext[WebDavFileSourceConfiguration],
        cache_options: CacheOptionsDictType,
    ):
        if WebdavFileSystem is None:
            raise self.required_package_exception

        config = context.config
        auth = (config.login, config.password) if config.login or config.password else None
        return WebdavFileSystem(self._webdav_endpoint(config.base_url, config.root), auth=auth)

    def _to_filesystem_path(self, path: str, config: WebDavFileSourceConfiguration) -> str:
        if path in ("", "/"):
            return ""
        return path.lstrip("/")

    def _adapt_entry_path(self, filesystem_path: str, config: WebDavFileSourceConfiguration) -> str:
        if not filesystem_path or filesystem_path == "/":
            return "/"
        return filesystem_path if filesystem_path.startswith("/") else f"/{filesystem_path}"



__all__ = ("WebDavFilesSource",)