Source code for galaxy.files.sources.webdav

try:
    from webdavfs.webdavfs import WebDAVFS
except ImportError:
    WebDAVFS = None

import tempfile
from typing import (
    Annotated,
    Optional,
    Union,
)

from pydantic import (
    Field,
    field_validator,
)

from galaxy.files.models import (
    BaseFileSourceConfiguration,
    BaseFileSourceTemplateConfiguration,
    FilesSourceRuntimeContext,
)
from galaxy.util.config_templates import TemplateExpansion
from ._pyfilesystem2 import PyFilesystem2FilesSource


class WebDavFileSourceTemplateConfiguration(BaseFileSourceTemplateConfiguration):
    url: Union[str, TemplateExpansion, None] = None
    root: Optional[Union[str, TemplateExpansion]] = None
    login: Optional[Union[str, TemplateExpansion]] = None
    password: Optional[Union[str, TemplateExpansion]] = None
    temp_path: Optional[Union[str, TemplateExpansion]] = None
    use_temp_files: Union[bool, TemplateExpansion] = True


class WebDavFileSourceConfiguration(BaseFileSourceConfiguration):
    # Override url field to make it required for WebDAV - we keep a default but validate it's provided
    url: Annotated[
        str,
        Field(
            None,
            title="WebDAV URL",
            description="The URL of the WebDAV server. This is required for WebDAV file sources.",
        ),
    ] = None  # type: ignore[assignment]
    root: Optional[str] = None
    login: Optional[str] = None
    password: Optional[str] = None
    temp_path: Optional[str] = None
    use_temp_files: bool = True  # Default to True to avoid memory issues with large files.

    @field_validator("url")
    @classmethod
    def validate_url_required(cls, v):
        if v is None or v == "":
            raise ValueError("url is required for WebDAV file source")
        return v



[docs]
class WebDavFilesSource(PyFilesystem2FilesSource[WebDavFileSourceTemplateConfiguration, WebDavFileSourceConfiguration]):
    plugin_type = "webdav"
    required_module = WebDAVFS
    required_package = "fs.webdavfs"
    allow_key_error_on_empty_directories = True

    template_config_class = WebDavFileSourceTemplateConfiguration
    resolved_config_class = WebDavFileSourceConfiguration

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

        config = context.config
        file_sources_config = self._file_sources_config
        use_temp_files = config.use_temp_files
        if file_sources_config and file_sources_config.webdav_use_temp_files is not None:
            use_temp_files = file_sources_config.webdav_use_temp_files

        if use_temp_files:
            temp_path = config.temp_path
            if temp_path is None and file_sources_config and file_sources_config.tmp_dir:
                temp_path = file_sources_config.tmp_dir
            if temp_path is None:
                temp_path = tempfile.mkdtemp(prefix="webdav_")
            config.temp_path = temp_path
        config.use_temp_files = use_temp_files

        handle = WebDAVFS(
            url=config.url,
            root=config.root,
            login=config.login,
            password=config.password,
            temp_path=config.temp_path,
            use_temp_files=config.use_temp_files,
        )
        return handle



__all__ = ("WebDavFilesSource",)