mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
migrate webdav to fsspec
This commit is contained in:
@@ -258,12 +258,14 @@ class ConditionalDependencies:
|
||||
def check_fs_dropboxfs(self):
|
||||
return "dropbox" in self.file_sources
|
||||
|
||||
def check_fs_webdavfs(self):
|
||||
def check_webdav4(self):
|
||||
return "webdav" in self.file_sources
|
||||
|
||||
def check_fs_webdavfs(self):
|
||||
return self.check_webdav4()
|
||||
|
||||
def check_webdavclient3(self):
|
||||
# fs.webdavfs dependency for which we need an unreleased version
|
||||
return self.check_fs_webdavfs()
|
||||
return self.check_webdav4()
|
||||
|
||||
def check_fs_anvilfs(self):
|
||||
# pyfilesystem plugin access to terra on anvil
|
||||
|
||||
@@ -21,9 +21,7 @@ fastmcp>=2.13.0
|
||||
redis>=5.3.0,<6
|
||||
|
||||
# For file sources plugins
|
||||
fs.webdavfs>=0.4.2 # type: webdav
|
||||
# webdavclient3 on the develop branch contains an important fix for username-based authentification containing the '@' symbol
|
||||
webdavclient3 @ git+https://github.com/ezhov-evgeny/webdav-client-python-3@98c23d1abd15efc3db9cfc756429f00041578bc2
|
||||
webdav4[fsspec] # type: webdav
|
||||
fs.dropboxfs>=1.0.3 # type: dropbox
|
||||
fs.sshfs # type: ssh
|
||||
fs.anvilfs # type: anvil
|
||||
|
||||
@@ -1,11 +1,6 @@
|
||||
try:
|
||||
from webdavfs.webdavfs import WebDAVFS
|
||||
except ImportError:
|
||||
WebDAVFS = None
|
||||
|
||||
import tempfile
|
||||
from typing import (
|
||||
Annotated,
|
||||
Any,
|
||||
Optional,
|
||||
Union,
|
||||
)
|
||||
@@ -13,86 +8,153 @@ from typing import (
|
||||
from pydantic import (
|
||||
Field,
|
||||
field_validator,
|
||||
model_validator,
|
||||
)
|
||||
|
||||
from galaxy.files.models import (
|
||||
BaseFileSourceConfiguration,
|
||||
BaseFileSourceTemplateConfiguration,
|
||||
FilesSourceRuntimeContext,
|
||||
)
|
||||
from galaxy.files.models import FilesSourceRuntimeContext
|
||||
from galaxy.util.config_templates import TemplateExpansion
|
||||
from ._pyfilesystem2 import PyFilesystem2FilesSource
|
||||
from ._fsspec import (
|
||||
CacheOptionsDictType,
|
||||
FsspecBaseFileSourceConfiguration,
|
||||
FsspecBaseFileSourceTemplateConfiguration,
|
||||
FsspecFilesSource,
|
||||
)
|
||||
|
||||
try:
|
||||
from webdav4.fsspec import WebdavFileSystem
|
||||
except ImportError:
|
||||
WebdavFileSystem = None
|
||||
|
||||
|
||||
class WebDavFileSourceTemplateConfiguration(BaseFileSourceTemplateConfiguration):
|
||||
def _normalize_root(root: Optional[str]) -> Optional[str]:
|
||||
if root is None:
|
||||
return None
|
||||
root = root.strip()
|
||||
if not root or root == "/":
|
||||
return None
|
||||
return f"/{root.strip('/')}"
|
||||
|
||||
|
||||
def _normalize_base_url(base_url: Optional[str]) -> Optional[str]:
|
||||
if base_url is None:
|
||||
return None
|
||||
base_url = base_url.strip()
|
||||
if not base_url:
|
||||
return None
|
||||
return base_url.rstrip("/")
|
||||
|
||||
|
||||
def _compose_base_url(url: Optional[str], root: Optional[str]) -> Optional[str]:
|
||||
if not url:
|
||||
return None
|
||||
url = url.rstrip("/")
|
||||
normalized_root = _normalize_root(root)
|
||||
if normalized_root:
|
||||
return f"{url}{normalized_root}"
|
||||
return url
|
||||
|
||||
|
||||
class WebDavFileSourceTemplateConfiguration(FsspecBaseFileSourceTemplateConfiguration):
|
||||
url: Union[str, TemplateExpansion, None] = None
|
||||
root: Optional[Union[str, TemplateExpansion]] = None
|
||||
base_url: Union[str, TemplateExpansion, None] = 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
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def normalize_endpoint(cls, data: Any) -> Any:
|
||||
if not isinstance(data, dict):
|
||||
return data
|
||||
normalized = dict(data)
|
||||
normalized["root"] = _normalize_root(normalized.get("root"))
|
||||
normalized["base_url"] = _normalize_base_url(
|
||||
normalized.get("base_url") or _compose_base_url(normalized.get("url"), normalized.get("root"))
|
||||
)
|
||||
return normalized
|
||||
|
||||
class WebDavFileSourceConfiguration(BaseFileSourceConfiguration):
|
||||
# Override url field to make it required for WebDAV - we keep a default but validate it's provided
|
||||
url: Annotated[
|
||||
|
||||
class WebDavFileSourceConfiguration(FsspecBaseFileSourceConfiguration):
|
||||
url: Optional[str] = None
|
||||
root: Optional[str] = None
|
||||
base_url: Annotated[
|
||||
str,
|
||||
Field(
|
||||
title="WebDAV URL",
|
||||
description="The URL of the WebDAV server. This is required for WebDAV file sources.",
|
||||
title="WebDAV base URL",
|
||||
description="The fully-qualified WebDAV endpoint URL used to access this file source.",
|
||||
),
|
||||
] = 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.
|
||||
use_temp_files: bool = True
|
||||
|
||||
@field_validator("url")
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def validate_url_required(cls, v):
|
||||
if v is None or v == "":
|
||||
raise ValueError("url is required for WebDAV file source")
|
||||
def normalize_endpoint(cls, data: Any) -> Any:
|
||||
if not isinstance(data, dict):
|
||||
return data
|
||||
normalized = dict(data)
|
||||
normalized["root"] = _normalize_root(normalized.get("root"))
|
||||
normalized["base_url"] = _normalize_base_url(
|
||||
normalized.get("base_url") or _compose_base_url(normalized.get("url"), normalized.get("root"))
|
||||
)
|
||||
return normalized
|
||||
|
||||
@field_validator("base_url")
|
||||
@classmethod
|
||||
def validate_base_url_required(cls, v: str) -> str:
|
||||
if not v:
|
||||
raise ValueError("base_url is required for WebDAV file source")
|
||||
return v
|
||||
|
||||
|
||||
class WebDavFilesSource(PyFilesystem2FilesSource[WebDavFileSourceTemplateConfiguration, WebDavFileSourceConfiguration]):
|
||||
class WebDavFilesSource(FsspecFilesSource[WebDavFileSourceTemplateConfiguration, WebDavFileSourceConfiguration]):
|
||||
plugin_type = "webdav"
|
||||
required_module = WebDAVFS
|
||||
required_package = "fs.webdavfs"
|
||||
allow_key_error_on_empty_directories = True
|
||||
required_module = WebdavFileSystem
|
||||
required_package = "webdav4"
|
||||
|
||||
template_config_class = WebDavFileSourceTemplateConfiguration
|
||||
resolved_config_class = WebDavFileSourceConfiguration
|
||||
|
||||
def _open_fs(self, context: FilesSourceRuntimeContext[WebDavFileSourceConfiguration]):
|
||||
if WebDAVFS is None:
|
||||
def __init__(self, template_config: WebDavFileSourceTemplateConfiguration):
|
||||
defaults: dict[str, Any] = {}
|
||||
if (
|
||||
"use_temp_files" not in template_config.model_fields_set
|
||||
and template_config.file_sources_config.webdav_use_temp_files is not None
|
||||
):
|
||||
defaults["use_temp_files"] = template_config.file_sources_config.webdav_use_temp_files
|
||||
if defaults:
|
||||
template_config = self._apply_defaults_to_template(defaults, template_config)
|
||||
super().__init__(template_config)
|
||||
|
||||
def _open_fs(
|
||||
self,
|
||||
context: FilesSourceRuntimeContext[WebDavFileSourceConfiguration],
|
||||
cache_options: CacheOptionsDictType,
|
||||
):
|
||||
if WebdavFileSystem 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
|
||||
auth = (config.login, config.password) if config.login or config.password else None
|
||||
return WebdavFileSystem(config.base_url, auth=auth, **cache_options)
|
||||
|
||||
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
|
||||
def _to_filesystem_path(self, path: str, config: WebDavFileSourceConfiguration) -> str:
|
||||
if path in ("", "/"):
|
||||
return ""
|
||||
return path.lstrip("/")
|
||||
|
||||
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
|
||||
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}"
|
||||
|
||||
def _get_cache_options(self, config: WebDavFileSourceConfiguration) -> dict[str, Any]:
|
||||
# webdav4 does not accept fsspec listing-cache constructor kwargs and forwards unexpected values to its client implementation.
|
||||
return {}
|
||||
|
||||
|
||||
__all__ = ("WebDavFilesSource",)
|
||||
|
||||
@@ -11,6 +11,8 @@ from galaxy_test.base import api_asserts
|
||||
from galaxy_test.base.populators import DatasetPopulator
|
||||
from galaxy_test.driver import integration_util
|
||||
|
||||
pytest.importorskip("webdav4.fsspec")
|
||||
|
||||
SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
|
||||
FILE_SOURCES_JOB_CONF = os.path.join(SCRIPT_DIRECTORY, "file_sources_conf.yml")
|
||||
|
||||
|
||||
@@ -18,6 +18,8 @@ from ._util import (
|
||||
)
|
||||
from .test_posix import _download_and_check_file
|
||||
|
||||
pytest.importorskip("webdav4.fsspec")
|
||||
|
||||
SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__))
|
||||
FILE_SOURCES_CONF = os.path.join(SCRIPT_DIRECTORY, "webdav_file_sources_conf.yml")
|
||||
FILE_SOURCES_CONF_NO_USE_TEMP_FILES = os.path.join(SCRIPT_DIRECTORY, "webdav_file_sources_without_use_temp_conf.yml")
|
||||
@@ -69,9 +71,15 @@ def test_sniff_to_tmp():
|
||||
def test_serialization():
|
||||
configs = [FILE_SOURCES_CONF_NO_USE_TEMP_FILES, FILE_SOURCES_CONF]
|
||||
for config in configs:
|
||||
file_sources_o = configured_file_sources(config)
|
||||
original = file_source_as_webdav(file_sources_o._file_sources[0])
|
||||
assert original._get_runtime_context().config.base_url == "http://127.0.0.1:7083"
|
||||
|
||||
# serialize the configured file sources and rematerialize them,
|
||||
# ensure they still function. This is needed for uploading files.
|
||||
file_sources = serialize_and_recover(configured_file_sources(config))
|
||||
file_sources = serialize_and_recover(file_sources_o)
|
||||
recovered = file_source_as_webdav(file_sources._file_sources[0])
|
||||
assert recovered._get_runtime_context().config.base_url == "http://127.0.0.1:7083"
|
||||
|
||||
res = list_root(file_sources, "gxfiles://test1", recursive=True)
|
||||
assert find_file_a(res)
|
||||
@@ -114,9 +122,15 @@ def test_serialization_user():
|
||||
file_sources_o = configured_file_sources(USER_FILE_SOURCES_CONF)
|
||||
user_context = user_context_fixture()
|
||||
|
||||
original = file_source_as_webdav(file_sources_o._file_sources[0])
|
||||
assert original._get_runtime_context(user_context=user_context).config.base_url == "http://127.0.0.1:7083"
|
||||
|
||||
res = list_root(file_sources_o, "gxfiles://test1", recursive=True, user_context=user_context)
|
||||
assert find_file_a(res)
|
||||
|
||||
file_sources = serialize_and_recover(file_sources_o, user_context=user_context)
|
||||
recovered = file_source_as_webdav(file_sources._file_sources[0])
|
||||
assert recovered._get_runtime_context().config.base_url == "http://127.0.0.1:7083"
|
||||
|
||||
res = list_root(file_sources, "gxfiles://test1", recursive=True, user_context=None)
|
||||
assert find_file_a(res)
|
||||
|
||||
Reference in New Issue
Block a user