From f23d31af644ab4ad4492b9cfa1000d58420c238d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Saugat=20Pachhai=20=28=E0=A4=B8=E0=A5=8C=E0=A4=97=E0=A4=BE?= =?UTF-8?q?=E0=A4=A4=29?= Date: Wed, 18 May 2022 21:43:52 +0545 Subject: [PATCH] fs/data: move repo related utils to dvc This moves get_fs_config/get_fs_cloud and ODBManager to DVC as they require DVC related utilities, and we don't need them strictly in dvc-objects/dvc-data. This also adds support to ReferenceHashFile to load from any filesystem, using it's module name and filename. Now we can get rid of DVCFileSystem dependency in ReferenceHashFile. This is a temporary solution, we really need a registry. (I am not quite sure if this works in pyinstaller mode, but I think it should, it's how fsspec does it if I understand correctly.) --- dvc/data/db/__init__.py | 65 +---------------- dvc/data/reference.py | 14 ++-- dvc/fs/__init__.py | 93 +++++++++++++++++++++++- dvc/objects/fs/__init__.py | 105 ++++----------------------- dvc/odbmgr.py | 64 ++++++++++++++++ dvc/repo/__init__.py | 2 +- dvc/repo/experiments/executor/ssh.py | 2 +- dvc/testing/fixtures.py | 2 +- dvc/testing/test_workspace.py | 2 +- tests/func/test_add.py | 2 +- tests/func/test_get.py | 2 +- tests/func/test_import.py | 2 +- tests/func/test_odb.py | 2 +- tests/func/test_repo.py | 2 +- tests/unit/stage/test_cache.py | 2 +- 15 files changed, 190 insertions(+), 171 deletions(-) create mode 100644 dvc/odbmgr.py diff --git a/dvc/data/db/__init__.py b/dvc/data/db/__init__.py index 049037e85..eb5909333 100644 --- a/dvc/data/db/__init__.py +++ b/dvc/data/db/__init__.py @@ -1,13 +1,12 @@ from typing import TYPE_CHECKING -from dvc.objects.fs import Schemes - if TYPE_CHECKING: from .index import ObjectDBIndexBase def get_odb(fs, fs_path, **config): from dvc.objects.db import ObjectDB + from dvc.objects.fs import Schemes from .local import LocalObjectDB @@ -17,17 +16,6 @@ def get_odb(fs, fs_path, **config): return ObjectDB(fs, fs_path, **config) -def _get_odb(repo, settings): - from dvc.objects.fs import get_cloud_fs - - if not settings: - return None - - cls, config, fs_path = get_cloud_fs(repo, **settings) - config["tmp_dir"] = repo.tmp_dir - return get_odb(cls(**config), fs_path, state=repo.state, **config) - - def get_index(odb) -> "ObjectDBIndexBase": import hashlib @@ -40,54 +28,3 @@ def get_index(odb) -> "ObjectDBIndexBase": odb.fs.unstrip_protocol(odb.fs_path).encode("utf-8") ).hexdigest(), ) - - -class ODBManager: - CACHE_DIR = "cache" - CLOUD_SCHEMES = [ - Schemes.S3, - Schemes.GS, - Schemes.SSH, - Schemes.HDFS, - Schemes.WEBHDFS, - ] - - def __init__(self, repo): - self.repo = repo - self.config = config = repo.config["cache"] - self._odb = {} - - local = config.get("local") - - if local: - settings = {"name": local} - elif "dir" not in config: - settings = None - else: - from dvc.config_schema import LOCAL_COMMON - - settings = {"url": config["dir"]} - for opt in LOCAL_COMMON.keys(): - if opt in config: - settings[str(opt)] = config.get(opt) - - self._odb[Schemes.LOCAL] = _get_odb(repo, settings) - - def _init_odb(self, schemes): - for scheme in schemes: - remote = self.config.get(scheme) - settings = {"name": remote} if remote else None - self._odb[scheme] = _get_odb(self.repo, settings) - - def __getattr__(self, name): - if name not in self._odb and name in self.CLOUD_SCHEMES: - self._init_odb([name]) - - try: - return self._odb[name] - except KeyError as exc: - raise AttributeError from exc - - def by_scheme(self): - self._init_odb(self.CLOUD_SCHEMES) - yield from self._odb.items() diff --git a/dvc/data/reference.py b/dvc/data/reference.py index 16449fa50..2b9fd18d8 100644 --- a/dvc/data/reference.py +++ b/dvc/data/reference.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, Optional from dvc.objects.errors import ObjectFormatError from dvc.objects.file import HashFile +from dvc.objects.fs import FS_MAP, LocalFileSystem if TYPE_CHECKING: from dvc.objects.db import ObjectDB @@ -20,6 +21,7 @@ class ReferenceHashFile(HashFile): PARAM_HASH = "hash" PARAM_CHECKSUM = "checksum" PARAM_FS_CONFIG = "fs_config" + PARAM_FS_CLS = "fs_name" def __init__( self, @@ -57,11 +59,16 @@ class ReferenceHashFile(HashFile): # ReferenceHashFiles should currently only be serialized in # memory and not to disk fs_path = self.fs_path + fs_cls = type(self.fs) + mod = None + if fs_cls not in FS_MAP.values() and fs_cls != LocalFileSystem: + mod = ".".join((fs_cls.__module__, fs_cls.__name__)) dict_ = { self.PARAM_PATH: fs_path, self.PARAM_HASH: self.hash_info, self.PARAM_CHECKSUM: self.checksum, self.PARAM_FS_CONFIG: self.config_tuple(self.fs), + self.PARAM_FS_CLS: mod, } try: return pickle.dumps(dict_) @@ -70,7 +77,6 @@ class ReferenceHashFile(HashFile): @classmethod def from_bytes(cls, data: bytes, fs_cache: Optional[dict] = None): - from dvc.fs.dvc import DvcFileSystem, _DvcFileSystem from dvc.objects.fs import get_fs_cls try: @@ -88,10 +94,8 @@ class ReferenceHashFile(HashFile): fs = fs_cache.get((protocol, config_pairs)) if fs_cache else None if not fs: config = dict(config_pairs) - if _DvcFileSystem.PARAM_REPO_URL in config: - fs_cls = DvcFileSystem - else: - fs_cls = get_fs_cls(config, scheme=protocol) + mod = dict_.get(cls.PARAM_FS_CLS, None) + fs_cls = get_fs_cls(config, cls=mod, scheme=protocol) fs = fs_cls(**config) return ReferenceHashFile( fs_path, diff --git a/dvc/fs/__init__.py b/dvc/fs/__init__.py index 312b254b7..cbae931e9 100644 --- a/dvc/fs/__init__.py +++ b/dvc/fs/__init__.py @@ -1,3 +1,5 @@ +from urllib.parse import urlparse + # pylint: disable=unused-import from dvc.objects.fs import utils # noqa: F401 from dvc.objects.fs import ( # noqa: F401 @@ -17,11 +19,8 @@ from dvc.objects.fs import ( # noqa: F401 WebDAVFileSystem, WebDAVSFileSystem, WebHDFSFileSystem, - _resolve_remote_refs, generic, - get_cloud_fs, get_fs_cls, - get_fs_config, system, ) from dvc.objects.fs.base import AnyFSPath, FileSystem # noqa: F401 @@ -43,3 +42,91 @@ from .dvc import DvcFileSystem # noqa: F401 from .git import GitFileSystem # noqa: F401 # pylint: enable=unused-import + + +def get_fs_config(repo, config, **kwargs): + name = kwargs.get("name") + if name: + try: + remote_conf = config["remote"][name.lower()] + except KeyError: + from dvc.config import RemoteNotFoundError + + raise RemoteNotFoundError(f"remote '{name}' doesn't exist") + else: + remote_conf = kwargs + return _resolve_remote_refs(repo, config, remote_conf) + + +def _resolve_remote_refs(repo, config, remote_conf): + # Support for cross referenced remotes. + # This will merge the settings, shadowing base ref with remote_conf. + # For example, having: + # + # dvc remote add server ssh://localhost + # dvc remote modify server user root + # dvc remote modify server ask_password true + # + # dvc remote add images remote://server/tmp/pictures + # dvc remote modify images user alice + # dvc remote modify images ask_password false + # dvc remote modify images password asdf1234 + # + # Results on a config dictionary like: + # + # { + # "url": "ssh://localhost/tmp/pictures", + # "user": "alice", + # "password": "asdf1234", + # "ask_password": False, + # } + parsed = urlparse(remote_conf["url"]) + if parsed.scheme != "remote": + return remote_conf + + base = get_fs_config(repo, config, name=parsed.netloc) + cls, _, _ = get_cloud_fs(repo, **base) + relpath = parsed.path.lstrip("/").replace("/", cls.sep) + url = cls.sep.join((base["url"], relpath)) + return {**base, **remote_conf, "url": url} + + +def get_cloud_fs(repo, **kwargs): + from dvc.config import ConfigError as RepoConfigError + from dvc.config_schema import SCHEMA, Invalid + + repo_config = repo.config if repo else {} + core_config = repo_config.get("core", {}) + + remote_conf = get_fs_config(repo, repo_config, **kwargs) + try: + remote_conf = SCHEMA["remote"][str](remote_conf) + except Invalid as exc: + raise RepoConfigError(str(exc)) from None + + if "jobs" not in remote_conf: + jobs = core_config.get("jobs") + if jobs: + remote_conf["jobs"] = jobs + + if "checksum_jobs" not in remote_conf: + checksum_jobs = core_config.get("checksum_jobs") + if checksum_jobs: + remote_conf["checksum_jobs"] = checksum_jobs + + cls = get_fs_cls(remote_conf) + + if cls == GDriveFileSystem and repo: + remote_conf["gdrive_credentials_tmp_dir"] = repo.tmp_dir + + url = remote_conf.pop("url") + if issubclass(cls, WebDAVFileSystem): + # For WebDAVFileSystem, provided url is the base path itself, so it + # should be treated as being a root path. + fs_path = cls.root_marker + else: + fs_path = cls._strip_protocol(url) # pylint:disable=protected-access + + extras = cls._get_kwargs_from_urls(url) # pylint:disable=protected-access + conf = {**extras, **remote_conf} # remote config takes priority + return cls, conf, fs_path diff --git a/dvc/objects/fs/__init__.py b/dvc/objects/fs/__init__.py index 0a6e9e9eb..1087c8941 100644 --- a/dvc/objects/fs/__init__.py +++ b/dvc/objects/fs/__init__.py @@ -33,95 +33,22 @@ FS_MAP = { } -def get_fs_cls(remote_conf, scheme=None): +def _import_class(cls: str): + """Take a string FQP and return the imported class or identifier + + clas is of the form "package.module.klass". + """ + import importlib + + mod, name = cls.rsplit(".", maxsplit=1) + module = importlib.import_module(mod) + return getattr(module, name) + + +def get_fs_cls(remote_conf, cls=None, scheme=None): + if cls: + return _import_class(cls) + if not scheme: scheme = urlparse(remote_conf["url"]).scheme return FS_MAP.get(scheme, LocalFileSystem) - - -def get_fs_config(repo, config, **kwargs): - name = kwargs.get("name") - if name: - try: - remote_conf = config["remote"][name.lower()] - except KeyError: - from dvc.config import RemoteNotFoundError - - raise RemoteNotFoundError(f"remote '{name}' doesn't exist") - else: - remote_conf = kwargs - return _resolve_remote_refs(repo, config, remote_conf) - - -def _resolve_remote_refs(repo, config, remote_conf): - # Support for cross referenced remotes. - # This will merge the settings, shadowing base ref with remote_conf. - # For example, having: - # - # dvc remote add server ssh://localhost - # dvc remote modify server user root - # dvc remote modify server ask_password true - # - # dvc remote add images remote://server/tmp/pictures - # dvc remote modify images user alice - # dvc remote modify images ask_password false - # dvc remote modify images password asdf1234 - # - # Results on a config dictionary like: - # - # { - # "url": "ssh://localhost/tmp/pictures", - # "user": "alice", - # "password": "asdf1234", - # "ask_password": False, - # } - parsed = urlparse(remote_conf["url"]) - if parsed.scheme != "remote": - return remote_conf - - base = get_fs_config(repo, config, name=parsed.netloc) - cls, _, _ = get_cloud_fs(repo, **base) - relpath = parsed.path.lstrip("/").replace("/", cls.sep) - url = cls.sep.join((base["url"], relpath)) - return {**base, **remote_conf, "url": url} - - -def get_cloud_fs(repo, **kwargs): - from dvc.config import ConfigError - from dvc.config_schema import SCHEMA, Invalid - - repo_config = repo.config if repo else {} - core_config = repo_config.get("core", {}) - - remote_conf = get_fs_config(repo, repo_config, **kwargs) - try: - remote_conf = SCHEMA["remote"][str](remote_conf) - except Invalid as exc: - raise ConfigError(str(exc)) from None - - if "jobs" not in remote_conf: - jobs = core_config.get("jobs") - if jobs: - remote_conf["jobs"] = jobs - - if "checksum_jobs" not in remote_conf: - checksum_jobs = core_config.get("checksum_jobs") - if checksum_jobs: - remote_conf["checksum_jobs"] = checksum_jobs - - cls = get_fs_cls(remote_conf) - - if cls == GDriveFileSystem and repo: - remote_conf["gdrive_credentials_tmp_dir"] = repo.tmp_dir - - url = remote_conf.pop("url") - if issubclass(cls, WebDAVFileSystem): - # For WebDAVFileSystem, provided url is the base path itself, so it - # should be treated as being a root path. - fs_path = cls.root_marker - else: - fs_path = cls._strip_protocol(url) # pylint:disable=protected-access - - extras = cls._get_kwargs_from_urls(url) # pylint:disable=protected-access - conf = {**extras, **remote_conf} # remote config takes priority - return cls, conf, fs_path diff --git a/dvc/odbmgr.py b/dvc/odbmgr.py new file mode 100644 index 000000000..9cc82cc83 --- /dev/null +++ b/dvc/odbmgr.py @@ -0,0 +1,64 @@ +from dvc.data.db import get_odb +from dvc.fs import Schemes + + +def _get_odb(repo, settings): + from dvc.fs import get_cloud_fs + + if not settings: + return None + + cls, config, fs_path = get_cloud_fs(repo, **settings) + config["tmp_dir"] = repo.tmp_dir + return get_odb(cls(**config), fs_path, state=repo.state, **config) + + +class ODBManager: + CACHE_DIR = "cache" + CLOUD_SCHEMES = [ + Schemes.S3, + Schemes.GS, + Schemes.SSH, + Schemes.HDFS, + Schemes.WEBHDFS, + ] + + def __init__(self, repo): + self.repo = repo + self.config = config = repo.config["cache"] + self._odb = {} + + local = config.get("local") + + if local: + settings = {"name": local} + elif "dir" not in config: + settings = None + else: + from dvc.config_schema import LOCAL_COMMON + + settings = {"url": config["dir"]} + for opt in LOCAL_COMMON.keys(): + if opt in config: + settings[str(opt)] = config.get(opt) + + self._odb[Schemes.LOCAL] = _get_odb(repo, settings) + + def _init_odb(self, schemes): + for scheme in schemes: + remote = self.config.get(scheme) + settings = {"name": remote} if remote else None + self._odb[scheme] = _get_odb(self.repo, settings) + + def __getattr__(self, name): + if name not in self._odb and name in self.CLOUD_SCHEMES: + self._init_odb([name]) + + try: + return self._odb[name] + except KeyError as exc: + raise AttributeError from exc + + def by_scheme(self): + self._init_odb(self.CLOUD_SCHEMES) + yield from self._odb.items() diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 2bdcdbbff..410c4a76f 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -155,11 +155,11 @@ class Repo: scm=None, ): from dvc.config import Config - from dvc.data.db import ODBManager from dvc.data_cloud import DataCloud from dvc.fs import GitFileSystem, localfs from dvc.lock import LockNoop, make_lock from dvc.objects.state import State, StateNoop + from dvc.odbmgr import ODBManager from dvc.repo.metrics import Metrics from dvc.repo.params import Params from dvc.repo.plots import Plots diff --git a/dvc/repo/experiments/executor/ssh.py b/dvc/repo/experiments/executor/ssh.py index 080e9d8e3..0d03a6f62 100644 --- a/dvc/repo/experiments/executor/ssh.py +++ b/dvc/repo/experiments/executor/ssh.py @@ -204,7 +204,7 @@ class SSHExecutor(BaseExecutor): @contextmanager def get_odb(self): - from dvc.data.db import ODBManager, get_odb + from dvc.odbmgr import ODBManager, get_odb cache_path = posixpath.join( self._repo_abspath, diff --git a/dvc/testing/fixtures.py b/dvc/testing/fixtures.py index 3c9c541f0..0e79a18f6 100644 --- a/dvc/testing/fixtures.py +++ b/dvc/testing/fixtures.py @@ -149,7 +149,7 @@ def local_remote(make_remote): @pytest.fixture def make_workspace(tmp_dir, dvc, make_cloud): def _make_workspace(name, typ="local"): - from dvc.data.db import ODBManager + from dvc.odbmgr import ODBManager cloud = make_cloud(typ) # pylint: disable=W0621 diff --git a/dvc/testing/test_workspace.py b/dvc/testing/test_workspace.py index f6225a2fb..0b883a7ba 100644 --- a/dvc/testing/test_workspace.py +++ b/dvc/testing/test_workspace.py @@ -20,7 +20,7 @@ class TestImport: pytest.skip() def test_import_dir(self, tmp_dir, dvc, workspace, stage_md5, dir_md5): - from dvc.data.db import ODBManager + from dvc.odbmgr import ODBManager workspace.gen( {"dir": {"file": "file", "subdir": {"subfile": "subfile"}}} diff --git a/tests/func/test_add.py b/tests/func/test_add.py index de4cd4a80..7e0dcf5b8 100644 --- a/tests/func/test_add.py +++ b/tests/func/test_add.py @@ -12,7 +12,6 @@ import pytest import dvc as dvc_module from dvc.cli import main -from dvc.data.db import ODBManager from dvc.dvcfile import DVC_FILE_SUFFIX from dvc.exceptions import ( DvcException, @@ -24,6 +23,7 @@ from dvc.exceptions import ( from dvc.fs import LocalFileSystem, system from dvc.objects.hash import file_md5 from dvc.objects.hash_info import HashInfo +from dvc.odbmgr import ODBManager from dvc.output import ( OutputAlreadyTrackedError, OutputDoesNotExistError, diff --git a/tests/func/test_get.py b/tests/func/test_get.py index 6171bdb19..2f9a0a81c 100644 --- a/tests/func/test_get.py +++ b/tests/func/test_get.py @@ -4,8 +4,8 @@ import os import pytest from dvc.cli import main -from dvc.data.db import ODBManager from dvc.fs import system +from dvc.odbmgr import ODBManager from dvc.repo import Repo from dvc.repo.get import GetDVCFileError from dvc.utils.fs import makedirs diff --git a/tests/func/test_import.py b/tests/func/test_import.py index 279dd4e89..14a935afd 100644 --- a/tests/func/test_import.py +++ b/tests/func/test_import.py @@ -7,10 +7,10 @@ from funcy import first from scmrepo.git import Git from dvc.config import NoRemoteError -from dvc.data.db import ODBManager from dvc.dvcfile import Dvcfile from dvc.exceptions import DownloadError, PathMissingError from dvc.fs import system +from dvc.odbmgr import ODBManager from dvc.stage.exceptions import StagePathNotFoundError from dvc.utils.fs import makedirs, remove from tests.unit.fs.test_dvc import make_subrepo diff --git a/tests/func/test_odb.py b/tests/func/test_odb.py index 4a9b0deb5..f89ef63f2 100644 --- a/tests/func/test_odb.py +++ b/tests/func/test_odb.py @@ -5,9 +5,9 @@ import configobj import pytest from dvc.cli import main -from dvc.data.db import ODBManager from dvc.objects.errors import ObjectFormatError from dvc.objects.hash_info import HashInfo +from dvc.odbmgr import ODBManager from dvc.utils import relpath diff --git a/tests/func/test_repo.py b/tests/func/test_repo.py index 947f7f7c8..d149bbedc 100644 --- a/tests/func/test_repo.py +++ b/tests/func/test_repo.py @@ -1,6 +1,6 @@ -from dvc.data.db import ODBManager from dvc.dvcfile import PIPELINE_FILE, PIPELINE_LOCK from dvc.fs import system +from dvc.odbmgr import ODBManager def test_destroy(tmp_dir, dvc, run_copy): diff --git a/tests/unit/stage/test_cache.py b/tests/unit/stage/test_cache.py index c1dbf3947..9c0bc41eb 100644 --- a/tests/unit/stage/test_cache.py +++ b/tests/unit/stage/test_cache.py @@ -167,7 +167,7 @@ def test_stage_cache_wdir(tmp_dir, dvc, mocker): def test_shared_stage_cache(tmp_dir, dvc, run_copy): import stat - from dvc.data.db import ODBManager + from dvc.odbmgr import ODBManager tmp_dir.gen("foo", "foo")