Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-10 07:18:27 +01:00
Code Issues Packages Projects Releases Wiki Activity

fs/data: move repo related utils to dvc

Browse Source 
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.)
This commit is contained in:
Saugat Pachhai (सौगात)
2022-05-18 19:39:09 +03:00
committed by Ruslan Kuprieiev
parent cf5d3830c5
commit f23d31af64
15 changed files with 190 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/data/db/__init__.py
+1 -64
View File
@@ -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()
dvc/data/reference.py
+9 -5
View File
@@ -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,
dvc/fs/__init__.py
+90 -3
View File
@@ -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
dvc/objects/fs/__init__.py
+16 -89
View File
@@ -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
dvc/odbmgr.py
+64
View File
@@ -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()
dvc/repo/__init__.py
+1 -1
View File
@@ -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
dvc/repo/experiments/executor/ssh.py
+1 -1
View File
@@ -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,
dvc/testing/fixtures.py
+1 -1
View File
@@ -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
dvc/testing/test_workspace.py
+1 -1
View File
@@ -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"}}}
tests/func/test_add.py
+1 -1
View File
@@ -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,
tests/func/test_get.py
+1 -1
View File
@@ -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
tests/func/test_import.py
+1 -1
View File
@@ -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
tests/func/test_odb.py
+1 -1
View File
@@ -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
tests/func/test_repo.py
+1 -1
View File
@@ -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):
tests/unit/stage/test_cache.py
+1 -1
View File
@@ -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")