From 60a983b63ff9e22965094481e2ee376eb9e5e6da Mon Sep 17 00:00:00 2001 From: Alexander Schepanovski Date: Tue, 13 Aug 2019 18:19:52 +0700 Subject: [PATCH 1/7] api: stream api.open() for local, SSH and HDFS --- dvc/remote/hdfs.py | 14 ++++++++++++++ dvc/remote/local/__init__.py | 6 +++++- dvc/remote/ssh/__init__.py | 14 ++++++++++++++ tests/func/test_api.py | 8 +++++++- 4 files changed, 40 insertions(+), 2 deletions(-) diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index ea13a176a..2506c0724 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -2,10 +2,12 @@ from __future__ import unicode_literals import os import re +import io import getpass import posixpath import logging from subprocess import Popen, PIPE +from contextlib import contextmanager, closing try: import pyarrow @@ -139,6 +141,18 @@ class RemoteHDFS(RemoteBASE): with open(to_file, "wb+") as fobj: hdfs.download(from_info.path, fobj) + @contextmanager + def open(self, path_info, mode="r", encoding=None): + assert mode in {"r", "rt", "rb"} + + with self.hdfs(path_info) as hdfs, closing( + hdfs.open(path_info.path, mode="rb") + ) as fd: + if mode == "rb": + yield fd + else: + yield io.TextIOWrapper(fd, encoding=encoding) + def list_cache_paths(self): if not self.exists(self.path_info): return [] diff --git a/dvc/remote/local/__init__.py b/dvc/remote/local/__init__.py index 816cdaa2e..a201ba5bc 100644 --- a/dvc/remote/local/__init__.py +++ b/dvc/remote/local/__init__.py @@ -4,7 +4,7 @@ from copy import copy from dvc.scheme import Schemes from dvc.remote.local.slow_link_detection import slow_link_guard -from dvc.utils.compat import str, fspath_py35 +from dvc.utils.compat import str, fspath_py35, open import os import stat @@ -280,6 +280,10 @@ class RemoteLOCAL(RemoteBASE): name=name, ) + def open(self, path_info, mode="r", encoding=None): + assert mode in {"r", "rt", "rb"} + return open(fspath_py35(path_info), mode=mode, encoding=encoding) + def _group(self, checksum_infos, show_checksums=False): by_md5 = {} diff --git a/dvc/remote/ssh/__init__.py b/dvc/remote/ssh/__init__.py index 4202ba61a..fee5fd092 100644 --- a/dvc/remote/ssh/__init__.py +++ b/dvc/remote/ssh/__init__.py @@ -2,11 +2,13 @@ from __future__ import unicode_literals import errno import itertools +import io import os import getpass import logging import threading from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager, closing from dvc.progress import ProgressCallback from dvc.utils import to_chunks @@ -202,6 +204,18 @@ class RemoteSSH(RemoteBASE): no_progress_bar=no_progress_bar, ) + @contextmanager + def open(self, path_info, mode="r", encoding=None): + assert mode in {"r", "rt", "rb"} + + with self.ssh(path_info) as ssh, closing( + ssh.sftp.file(path_info.path, mode="r") + ) as fd: + if mode == "rb": + yield fd + else: + yield io.TextIOWrapper(fd, encoding=encoding) + def list_cache_paths(self): with self.ssh(self.path_info) as ssh: return list(ssh.walk_files(self.path_info.path)) diff --git a/tests/func/test_api.py b/tests/func/test_api.py index afc71d9ec..adbeebb81 100644 --- a/tests/func/test_api.py +++ b/tests/func/test_api.py @@ -7,6 +7,7 @@ from dvc.main import main from dvc.path_info import URLInfo from dvc.remote.config import RemoteConfig from .test_data_cloud import ( + get_local_url, _should_test_aws, get_aws_url, _should_test_gcp, @@ -23,6 +24,11 @@ from .test_data_cloud import ( # NOTE: staticmethod is only needed in Python 2 +class Local: + should_test = staticmethod(lambda: True) + get_url = staticmethod(get_local_url) + + class S3: should_test = staticmethod(_should_test_aws) get_url = staticmethod(get_aws_url) @@ -53,7 +59,7 @@ class HDFS: get_url = staticmethod(get_hdfs_url) -remote_params = [S3, GCP, Azure, OSS, SSH, HDFS] +remote_params = [Local, S3, GCP, Azure, OSS, SSH, HDFS] @pytest.fixture From e1f676a1915e5810bf2af3c654eaa31e245fce0f Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Thu, 15 Aug 2019 14:06:58 +0300 Subject: [PATCH 2/7] dvc: bump to 0.55.0 --- dvc/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dvc/version.py b/dvc/version.py index eab40ad98..6415e3f76 100644 --- a/dvc/version.py +++ b/dvc/version.py @@ -7,7 +7,7 @@ import os import subprocess -_BASE_VERSION = "0.54.1" +_BASE_VERSION = "0.55.0" def _generate_version(base_version): From 85289ad00bde5ee32fbe7732f2974295568b7166 Mon Sep 17 00:00:00 2001 From: Alexander Schepanovski Date: Thu, 15 Aug 2019 19:04:03 +0700 Subject: [PATCH 3/7] api: wrap not found errors api.open()/read() will now raise OutputFileMissingError and Remote*.open() will raise FileNotFoundError if target file is absent. --- dvc/remote/gs.py | 4 +++- dvc/remote/hdfs.py | 23 +++++++++++++++-------- dvc/repo/__init__.py | 16 ++++++++++------ dvc/utils/compat.py | 2 ++ dvc/utils/http.py | 4 ++++ tests/func/test_api.py | 23 +++++++++++++---------- 6 files changed, 47 insertions(+), 25 deletions(-) diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index e64a41e42..5ef730dd0 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -3,6 +3,7 @@ from __future__ import unicode_literals import logging from datetime import timedelta from funcy import cached_property +from dvc.utils.compat import FileNotFoundError try: from google.cloud import storage @@ -15,7 +16,6 @@ from dvc.exceptions import DvcException from dvc.path_info import CloudURLInfo from dvc.scheme import Schemes - logger = logging.getLogger(__name__) @@ -104,4 +104,6 @@ class RemoteGS(RemoteBASE): bucket = self.gs.bucket(path_info.bucket) blob = bucket.get_blob(path_info.path) + if blob is None: + raise FileNotFoundError return blob.generate_signed_url(expiration=expiration) diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 2506c0724..efdbfbba4 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -17,7 +17,7 @@ except ImportError: from dvc.config import Config from dvc.scheme import Schemes -from dvc.utils.compat import urlparse +from dvc.utils.compat import urlparse, FileNotFoundError from dvc.utils import fix_env, tmp_fname from .pool import get_connection @@ -145,13 +145,20 @@ class RemoteHDFS(RemoteBASE): def open(self, path_info, mode="r", encoding=None): assert mode in {"r", "rt", "rb"} - with self.hdfs(path_info) as hdfs, closing( - hdfs.open(path_info.path, mode="rb") - ) as fd: - if mode == "rb": - yield fd - else: - yield io.TextIOWrapper(fd, encoding=encoding) + try: + with self.hdfs(path_info) as hdfs, closing( + hdfs.open(path_info.path, mode="rb") + ) as fd: + if mode == "rb": + yield fd + else: + yield io.TextIOWrapper(fd, encoding=encoding) + except IOError as e: + # Empty .errno and not specific enough error class in pyarrow, + # see https://issues.apache.org/jira/browse/ARROW-6248 + if "file does not exist" in str(e): + raise FileNotFoundError(*e.args) + raise def list_cache_paths(self): if not self.exists(self.path_info): diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index 9b842b5f5..e1139d90a 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -2,7 +2,7 @@ from __future__ import unicode_literals import os import logging - +from contextlib import contextmanager from itertools import chain from funcy import cached_property @@ -17,7 +17,7 @@ from dvc.exceptions import ( from dvc.ignore import DvcIgnoreFilter from dvc.path_info import PathInfo from dvc.remote.base import RemoteActionNotImplemented -from dvc.utils.compat import open as _open, fspath_py35 +from dvc.utils.compat import open as _open, fspath_py35, FileNotFoundError from dvc.utils import relpath logger = logging.getLogger(__name__) @@ -456,8 +456,16 @@ class Repo(object): path_parts = os.path.normpath(path).split(os.path.sep) return self.DVC_DIR in path_parts + @contextmanager def open(self, path, remote=None, mode="r", encoding=None): """Opens a specified resource as a file descriptor""" + try: + with self._open(path, remote, mode, encoding) as fd: + yield fd + except FileNotFoundError: + raise OutputFileMissingError(relpath(path, self.root_dir)) + + def _open(self, path, remote=None, mode="r", encoding=None): out, = self.find_outs_by_path(path) if out.isdir(): raise ValueError("Can't open a dir") @@ -477,10 +485,6 @@ class Repo(object): cache_info = out.get_used_cache(remote=remote) self.cloud.pull(cache_info, remote=remote) - # Since pull may just skip with a warning, we need to check it here - if not os.path.exists(cache_file): - raise OutputFileMissingError(relpath(path, self.root_dir)) - return _open(cache_file, mode=mode, encoding=encoding) @cached_property diff --git a/dvc/utils/compat.py b/dvc/utils/compat.py index 4c4c95cf7..9dad37d3f 100644 --- a/dvc/utils/compat.py +++ b/dvc/utils/compat.py @@ -119,6 +119,7 @@ if is_py2: cast_bytes_py2 = cast_bytes makedirs = _makedirs range = xrange # noqa: F821 + FileNotFoundError = IOError import StringIO import io @@ -165,6 +166,7 @@ elif is_py3: open = open # noqa: F821 cast_bytes_py2 = no_code range = range # noqa: F821 + FileNotFoundError = FileNotFoundError # Backport os.fspath() from Python 3.6 diff --git a/dvc/utils/http.py b/dvc/utils/http.py index b6c0af596..b2ecf4bb3 100644 --- a/dvc/utils/http.py +++ b/dvc/utils/http.py @@ -3,6 +3,8 @@ import io import requests +from dvc.utils.compat import FileNotFoundError + @contextmanager def open_url(url, mode="r", encoding=None): @@ -30,6 +32,8 @@ def iter_url(url, chunk_size=io.DEFAULT_BUFFER_SIZE): def request(headers=None): the_url = url() if callable(url) else url response = requests.get(the_url, stream=True, headers=headers) + if response.status_code == 404: + raise FileNotFoundError("Can't open {}".format(the_url)) response.raise_for_status() return response diff --git a/tests/func/test_api.py b/tests/func/test_api.py index adbeebb81..2bba711b1 100644 --- a/tests/func/test_api.py +++ b/tests/func/test_api.py @@ -59,7 +59,8 @@ class HDFS: get_url = staticmethod(get_hdfs_url) -remote_params = [Local, S3, GCP, Azure, OSS, SSH, HDFS] +remote_params = [S3, GCP, Azure, OSS, SSH, HDFS] +all_remote_params = [Local] + remote_params @pytest.fixture @@ -69,15 +70,11 @@ def remote_url(request): return request.param.get_url() -def pytest_generate_tests(metafunc): - if "remote_url" in metafunc.fixturenames: - metafunc.parametrize("remote_url", remote_params, indirect=True) - - def run_dvc(*argv): assert main(argv) == 0 +@pytest.mark.parametrize("remote_url", remote_params, indirect=True) def test_get_url(repo_dir, dvc_repo, remote_url): run_dvc("remote", "add", "-d", "upstream", remote_url) dvc_repo.add(repo_dir.FOO) @@ -86,6 +83,7 @@ def test_get_url(repo_dir, dvc_repo, remote_url): assert api.get_url(repo_dir.FOO) == expected_url +@pytest.mark.parametrize("remote_url", remote_params, indirect=True) def test_get_url_external(repo_dir, dvc_repo, erepo, remote_url): _set_remote_url_and_commit(erepo.dvc, remote_url) @@ -95,6 +93,7 @@ def test_get_url_external(repo_dir, dvc_repo, erepo, remote_url): assert api.get_url(repo_dir.FOO, repo=repo_url) == expected_url +@pytest.mark.parametrize("remote_url", all_remote_params, indirect=True) def test_open(repo_dir, dvc_repo, remote_url): run_dvc("remote", "add", "-d", "upstream", remote_url) dvc_repo.add(repo_dir.FOO) @@ -107,6 +106,7 @@ def test_open(repo_dir, dvc_repo, remote_url): assert fd.read() == repo_dir.FOO_CONTENTS +@pytest.mark.parametrize("remote_url", all_remote_params, indirect=True) def test_open_external(repo_dir, dvc_repo, erepo, remote_url): erepo.dvc.scm.checkout("branch") _set_remote_url_and_commit(erepo.dvc, remote_url) @@ -126,13 +126,16 @@ def test_open_external(repo_dir, dvc_repo, erepo, remote_url): assert api.read("version", repo=repo_url, rev="branch") == "branch" -def test_open_missing(erepo): +@pytest.mark.parametrize("remote_url", all_remote_params, indirect=True) +def test_missing(repo_dir, dvc_repo, remote_url): + run_dvc("add", repo_dir.FOO) + run_dvc("remote", "add", "-d", "upstream", remote_url) + # Remove cache to make foo missing - shutil.rmtree(erepo.dvc.cache.local.cache_dir) + shutil.rmtree(dvc_repo.cache.local.cache_dir) - repo_url = "file://" + erepo.dvc.root_dir with pytest.raises(OutputFileMissingError): - api.read(erepo.FOO, repo=repo_url) + api.read(repo_dir.FOO) def _set_remote_url_and_commit(repo, remote_url): From 875186d5a60e97c4f45deb2c0696283a0e0c5a6a Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 18 Aug 2019 05:09:30 +0300 Subject: [PATCH 4/7] dvc: remove unused code A bunch of leftovers from a very old pkg implementation. --- dvc/output/base.py | 5 ----- dvc/output/local.py | 14 -------------- dvc/scm/git/__init__.py | 8 ++------ 3 files changed, 2 insertions(+), 25 deletions(-) diff --git a/dvc/output/base.py b/dvc/output/base.py index c887518be..f91a939e3 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -138,11 +138,6 @@ class OutputBase(object): def dir_cache(self): return self.cache.get_dir_cache(self.checksum) - def assign_to_stage_file(self, target_repo): - raise DvcException( - "change repo is not supported for {}".format(self.scheme) - ) - @classmethod def supported(cls, url): return cls.REMOTE.supported(url) diff --git a/dvc/output/local.py b/dvc/output/local.py index 13ae650f2..777ac3756 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -47,20 +47,6 @@ class OutputLOCAL(OutputBase): def_scheme = urlparse(self.def_path).scheme return def_scheme != "remote" and not os.path.isabs(self.def_path) - def assign_to_stage_file(self, stage): - from dvc.repo import Repo - - # Make path info the same but relative to passed stage - rel_path = self.path_info.relpath(self.stage.wdir) - self.path_info = ( - self.REMOTE.path_cls(os.path.abspath(stage.wdir)) / rel_path - ) - - self.repo = Repo(self.fspath) - - self.stage = stage - return self - def dumpd(self): ret = super(OutputLOCAL, self).dumpd() if self.is_in_repo: diff --git a/dvc/scm/git/__init__.py b/dvc/scm/git/__init__.py index 290ca5802..9c18b216e 100644 --- a/dvc/scm/git/__init__.py +++ b/dvc/scm/git/__init__.py @@ -114,12 +114,8 @@ class Git(Base): ) return False - def ignore(self, path, in_curr_dir=False): - base_dir = ( - os.path.realpath(os.curdir) - if in_curr_dir - else os.path.dirname(path) - ) + def ignore(self, path): + base_dir = os.path.dirname(path) entry, gitignore = self._get_gitignore(path, base_dir) if self._ignored(entry, gitignore): From 5915f17fcf61b4b58d6cbb98f35494bcea6f82d6 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Sun, 18 Aug 2019 05:15:08 +0300 Subject: [PATCH 5/7] output: base: remove unnecessary 'else' Reported by DeepSource. --- dvc/output/base.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/dvc/output/base.py b/dvc/output/base.py index f91a939e3..8b55bd71f 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -104,8 +104,7 @@ class OutputBase(object): if remote: parsed = urlparse(path) return remote.path_info / parsed.path.lstrip("/") - else: - return self.REMOTE.path_cls(path) + return self.REMOTE.path_cls(path) def __repr__(self): return "{class_name}: '{def_path}'".format( From 5a9dfbb6a4c6c41768abe53f0e6538bf32b4efa2 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 19 Aug 2019 09:43:24 +0300 Subject: [PATCH 6/7] config: use new remote url --- .dvc/config | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.dvc/config b/.dvc/config index 5950d7ed1..16f5743a4 100644 --- a/.dvc/config +++ b/.dvc/config @@ -1,4 +1,4 @@ ['remote "dvc"'] -url = https://dvc.org/s3/dvc +url = https://remote.dvc.org/dvc [core] remote = dvc From 30a7b77f0edf549273c9c837d0c2db8e549cb4fd Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Mon, 19 Aug 2019 10:55:22 +0300 Subject: [PATCH 7/7] stage: run: don't use unassigned old_handler Fix #2406 --- dvc/stage.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/dvc/stage.py b/dvc/stage.py index e93d5ab92..373de499e 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -775,6 +775,7 @@ class Stage(object): main_thread = isinstance( threading.current_thread(), threading._MainThread ) + old_handler = None p = None try: @@ -790,7 +791,7 @@ class Stage(object): old_handler = signal.signal(signal.SIGINT, signal.SIG_IGN) p.communicate() finally: - if main_thread: + if old_handler: signal.signal(signal.SIGINT, old_handler) if (p is None) or (p.returncode != 0):