mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-10 07:18:27 +01:00
webdav4: use webdav4 (#6099)
This commit is contained in:
Saugat Pachhai
+48
-224
@@ -1,264 +1,88 @@
|
||||
import io
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from collections import deque
|
||||
from functools import lru_cache
|
||||
|
||||
from funcy import cached_property, nullcontext, wrap_prop
|
||||
from funcy import cached_property, wrap_prop
|
||||
|
||||
from dvc.config import ConfigError
|
||||
from dvc.exceptions import DvcException
|
||||
from dvc.path_info import HTTPURLInfo, WebDAVURLInfo
|
||||
from dvc.progress import Tqdm
|
||||
from dvc.path_info import WebDAVURLInfo
|
||||
from dvc.scheme import Schemes
|
||||
|
||||
from .base import BaseFileSystem
|
||||
from .fsspec_wrapper import FSSpecWrapper
|
||||
from .http import ask_password
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class WebDAVConnectionError(DvcException):
|
||||
def __init__(self, host):
|
||||
super().__init__(f"Unable to connect to WebDAV {host}.")
|
||||
|
||||
|
||||
class WebDAVFileSystem(BaseFileSystem): # pylint:disable=abstract-method
|
||||
# Use webdav scheme
|
||||
class WebDAVFileSystem(FSSpecWrapper): # pylint:disable=abstract-method
|
||||
scheme = Schemes.WEBDAV
|
||||
|
||||
# URLInfo for Webdav ~ replaces webdav -> http
|
||||
PATH_CLS = WebDAVURLInfo
|
||||
|
||||
# Traversable as walk_files is implemented
|
||||
CAN_TRAVERSE = True
|
||||
|
||||
# Length of walk_files prefix
|
||||
TRAVERSE_PREFIX_LEN = 2
|
||||
|
||||
# Implementation based on webdav3.client
|
||||
REQUIRES = {"webdavclient3": "webdav3.client"}
|
||||
|
||||
# Chunk size for buffered upload/download with progress bar
|
||||
REQUIRES = {"webdav4": "webdav4"}
|
||||
CHUNK_SIZE = 2 ** 16
|
||||
|
||||
PARAM_CHECKSUM = "etag"
|
||||
DETAIL_FIELDS = frozenset(("etag", "size"))
|
||||
|
||||
# Constructor
|
||||
def __init__(self, **config):
|
||||
# Call BaseFileSystem constructor
|
||||
super().__init__(**config)
|
||||
|
||||
# Get username from configuration
|
||||
self.user = config.get("user", None)
|
||||
cert_path = config.get("cert_path", None)
|
||||
key_path = config.get("key_path", None)
|
||||
cert = cert_path if not key_path else (cert_path, key_path)
|
||||
|
||||
# Get password from configuration (might be None ~ not set)
|
||||
self.password = config.get("password", None)
|
||||
|
||||
# Whether to ask for password if it is not set
|
||||
self.ask_password = config.get("ask_password", False)
|
||||
|
||||
# Use token for webdav auth
|
||||
self.token = config.get("token", None)
|
||||
|
||||
# Path to certificate
|
||||
self.cert_path = config.get("cert_path", None)
|
||||
|
||||
# Path to private key
|
||||
self.key_path = config.get("key_path", None)
|
||||
|
||||
# Connection timeout
|
||||
self.timeout = config.get("timeout", 30)
|
||||
|
||||
self.hostname = config["host"]
|
||||
self.prefix = config.get("prefix")
|
||||
self.user = config.get("user")
|
||||
self.fs_args.update({"base_url": config["url"], "cert": cert})
|
||||
self.prefix = config.get("prefix", "")
|
||||
|
||||
@staticmethod
|
||||
def _get_kwargs_from_urls(urlpath):
|
||||
path_info = WebDAVURLInfo(urlpath)
|
||||
http_info = HTTPURLInfo(path_info.url)
|
||||
return {
|
||||
"prefix": path_info.path,
|
||||
"host": http_info.replace(path="").url,
|
||||
"prefix": path_info.path.rstrip("/"),
|
||||
"host": path_info.replace(path="").url,
|
||||
"url": path_info.url.rstrip("/"),
|
||||
}
|
||||
|
||||
# Webdav client
|
||||
def _prepare_credentials(self, **config):
|
||||
self.user = user = config.get("user", None)
|
||||
self.password = password = config.get("password", None)
|
||||
|
||||
headers = {}
|
||||
token = config.get("token")
|
||||
if token:
|
||||
headers.update({"Authorization": f"Bearer {token}"})
|
||||
elif user and not password and config.get("ask_password"):
|
||||
self.password = password = ask_password(config["host"], self.user)
|
||||
|
||||
auth = (user, password) if user and password else None
|
||||
return {"headers": headers, "auth": auth}
|
||||
|
||||
@wrap_prop(threading.Lock())
|
||||
@cached_property
|
||||
def _client(self):
|
||||
from webdav3.client import Client
|
||||
def fs(self):
|
||||
from webdav4.fsspec import WebdavFileSystem
|
||||
|
||||
# Set password or ask for it
|
||||
if self.ask_password and self.password is None and self.token is None:
|
||||
self.password = ask_password(self.hostname, self.user)
|
||||
|
||||
# Setup webdav client options dictionary
|
||||
options = {
|
||||
"webdav_hostname": self.hostname,
|
||||
"webdav_login": self.user,
|
||||
"webdav_password": self.password,
|
||||
"webdav_token": self.token,
|
||||
"webdav_cert_path": self.cert_path,
|
||||
"webdav_key_path": self.key_path,
|
||||
"webdav_timeout": self.timeout,
|
||||
"webdav_chunk_size": self.CHUNK_SIZE,
|
||||
}
|
||||
|
||||
client = Client(options)
|
||||
|
||||
# Check whether client options are valid
|
||||
if not client.valid():
|
||||
raise ConfigError(
|
||||
f"Configuration for WebDAV {self.hostname} is invalid."
|
||||
)
|
||||
|
||||
return client
|
||||
|
||||
def open(self, path_info, mode="r", encoding=None, **kwargs):
|
||||
from webdav3.exceptions import RemoteResourceNotFound
|
||||
|
||||
assert mode in {"r", "rt", "rb"}
|
||||
|
||||
fobj = io.BytesIO()
|
||||
|
||||
try:
|
||||
self._client.download_from(buff=fobj, remote_path=path_info.path)
|
||||
except RemoteResourceNotFound as exc:
|
||||
raise FileNotFoundError from exc
|
||||
|
||||
fobj.seek(0)
|
||||
|
||||
if "mode" == "rb":
|
||||
return fobj
|
||||
|
||||
return io.TextIOWrapper(fobj, encoding=encoding)
|
||||
|
||||
# Checks whether file/directory exists at remote
|
||||
def exists(self, path_info) -> bool:
|
||||
# Use webdav check to test for file existence
|
||||
return self._client.check(path_info.path)
|
||||
|
||||
# Checks whether path points to directory
|
||||
def isdir(self, path_info):
|
||||
# Use webdav is_dir to test whether path points to a directory
|
||||
return self._client.is_dir(path_info.path)
|
||||
|
||||
def walk_files(self, path_info, **kwargs):
|
||||
if not self.exists(path_info):
|
||||
return
|
||||
|
||||
for file in self.find(path_info):
|
||||
yield path_info.replace(path=file)
|
||||
|
||||
def ls(self, path_info, detail=False):
|
||||
for entry in self._client.list(path_info.path):
|
||||
path = entry["path"]
|
||||
if detail:
|
||||
if entry["isdir"]:
|
||||
yield {"type": "directory", "name": path}
|
||||
else:
|
||||
yield {
|
||||
"type": "file",
|
||||
"name": path,
|
||||
"size": entry["size"],
|
||||
"etag": entry["etag"],
|
||||
}
|
||||
else:
|
||||
yield path
|
||||
|
||||
def find(self, path_info, detail=False):
|
||||
dirs = deque([path_info.path])
|
||||
|
||||
while dirs:
|
||||
for entry in self._client.list(dirs.pop(), get_info=True):
|
||||
path = entry["path"]
|
||||
if entry["isdir"]:
|
||||
dirs.append(path)
|
||||
continue
|
||||
|
||||
if detail:
|
||||
yield {
|
||||
"type": "file",
|
||||
"name": path,
|
||||
"size": entry["size"],
|
||||
"etag": entry["etag"],
|
||||
}
|
||||
else:
|
||||
yield path
|
||||
|
||||
# Removes file/directory
|
||||
def remove(self, path_info):
|
||||
# Use webdav client clean (DELETE) method to remove file/directory
|
||||
self._client.clean(path_info.path)
|
||||
|
||||
# Creates directories
|
||||
def makedirs(self, path_info):
|
||||
# Terminate recursion
|
||||
if path_info.path == self.prefix or self.exists(path_info):
|
||||
return
|
||||
|
||||
# Recursively descent to root
|
||||
self.makedirs(path_info.parent)
|
||||
|
||||
# Construct directory at current recursion depth
|
||||
self._client.mkdir(path_info.path)
|
||||
|
||||
# Moves file/directory at remote
|
||||
def move(self, from_info, to_info):
|
||||
# Webdav client move
|
||||
self._client.move(from_info.path, to_info.path)
|
||||
return WebdavFileSystem(**self.fs_args)
|
||||
|
||||
def _upload_fobj(self, fobj, to_info):
|
||||
# In contrast to other upload_fobj implementations, this one does not
|
||||
# exactly do a chunked-upload but rather put everything in one request.
|
||||
self.makedirs(to_info.parent)
|
||||
self._client.upload_to(buff=fobj, remote_path=to_info.path)
|
||||
rpath = self.translate_path_info(to_info)
|
||||
self.makedirs(os.path.dirname(rpath))
|
||||
# using upload_fileobj to directly upload fileobj
|
||||
# rather than buffering it.
|
||||
# TODO: retry upload on failure
|
||||
return self.fs.client.upload_fileobj(fobj, rpath)
|
||||
|
||||
# Downloads file from remote to file
|
||||
def _download(self, from_info, to_file, name=None, no_progress_bar=False):
|
||||
# Progress from HTTPFileSystem
|
||||
with open(to_file, "wb") as fd:
|
||||
with Tqdm.wrapattr(
|
||||
fd,
|
||||
"write",
|
||||
total=None if no_progress_bar else self.getsize(from_info),
|
||||
leave=False,
|
||||
desc=from_info.url if name is None else name,
|
||||
disable=no_progress_bar,
|
||||
) as fd_wrapped:
|
||||
# Download from WebDAV via buffer
|
||||
self._client.download_from(
|
||||
buff=fd_wrapped, remote_path=from_info.path
|
||||
)
|
||||
def makedirs(self, path_info):
|
||||
path = self.translate_path_info(path_info)
|
||||
return self.fs.makedirs(path, exist_ok=True)
|
||||
|
||||
# Uploads file to remote
|
||||
def _upload(
|
||||
self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
|
||||
):
|
||||
# First try to create parent directories
|
||||
self.makedirs(to_info.parent)
|
||||
@lru_cache(512)
|
||||
def translate_path_info(self, path):
|
||||
if isinstance(path, self.PATH_CLS):
|
||||
return path.path[len(self.prefix) :].lstrip("/")
|
||||
return path
|
||||
|
||||
file_size = os.path.getsize(from_file)
|
||||
with open(from_file, "rb") as fd:
|
||||
progress_context = (
|
||||
nullcontext(fd)
|
||||
if file_size == 0
|
||||
else Tqdm.wrapattr(
|
||||
fd,
|
||||
"read",
|
||||
total=None if no_progress_bar else file_size,
|
||||
leave=False,
|
||||
desc=to_info.url if name is None else name,
|
||||
disable=no_progress_bar,
|
||||
)
|
||||
)
|
||||
with progress_context as fd_wrapped:
|
||||
self._client.upload_to(
|
||||
buff=fd_wrapped, remote_path=to_info.path
|
||||
)
|
||||
_with_bucket = translate_path_info
|
||||
|
||||
def info(self, path_info):
|
||||
info = self._client.info(path_info.path)
|
||||
return {"size": int(info["size"]), "etag": info["etag"]}
|
||||
def _strip_bucket(self, entry):
|
||||
return entry
|
||||
|
||||
@@ -106,7 +106,7 @@ ssh = ["paramiko[invoke]>=2.7.0"]
|
||||
# Remove the env marker if/when pyarrow is available for Python3.9
|
||||
hdfs = ["pyarrow>=2.0.0"]
|
||||
webhdfs = ["hdfs==2.5.8"]
|
||||
webdav = ["webdavclient3>=3.14.5"]
|
||||
webdav = ["webdav4>=0.6.1"]
|
||||
# gssapi should not be included in all_remotes, because it doesn't have wheels
|
||||
# for linux and mac, so it will fail to compile if user doesn't have all the
|
||||
# requirements, including kerberos itself. Once all the wheels are available,
|
||||
|
||||
@@ -21,25 +21,25 @@ class Webdav(Base, WebDAVURLInfo):
|
||||
|
||||
@cached_property
|
||||
def client(self):
|
||||
from webdav3.client import Client
|
||||
from webdav4.client import Client
|
||||
|
||||
user, secrets = first(AUTH.items())
|
||||
return Client(
|
||||
{
|
||||
"webdav_hostname": self.replace(path="").url,
|
||||
"webdav_login": user,
|
||||
"webdav_password": secrets["password"],
|
||||
}
|
||||
self.replace(path="").url, auth=(user, secrets["password"])
|
||||
)
|
||||
|
||||
def mkdir(self, mode=0o777, parents=False, exist_ok=False):
|
||||
assert mode == 0o777
|
||||
parent_dirs = list(reversed(self.parents))[1:] if parents else []
|
||||
for d in parent_dirs + [self]:
|
||||
self.client.mkdir(d.path) # pylint: disable=no-member
|
||||
path = d.path # pylint: disable=no-member
|
||||
if not self.client.exists(path):
|
||||
self.client.mkdir(path)
|
||||
|
||||
def write_bytes(self, contents):
|
||||
self.client.upload_to(contents, self.path)
|
||||
from io import BytesIO
|
||||
|
||||
self.client.upload_fileobj(BytesIO(contents), self.path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -8,13 +8,11 @@ password = "password"
|
||||
|
||||
|
||||
def test_user(dvc):
|
||||
fs = WebDAVFileSystem(host=url, user=user)
|
||||
|
||||
fs = WebDAVFileSystem(url=url, user=user)
|
||||
assert fs.user == user
|
||||
|
||||
|
||||
def test_password(dvc):
|
||||
config = {"host": url, "user": user, "password": password}
|
||||
config = {"url": url, "user": user, "password": password}
|
||||
fs = WebDAVFileSystem(**config)
|
||||
|
||||
assert fs.password == password
|
||||
|
||||
Reference in New Issue
Block a user