diff --git a/dvc/cli.py b/dvc/cli.py index 28ec63836..24edf0d34 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -167,6 +167,7 @@ def parse_args(argv=None): help="URL. Supported urls: " "'/path/to/file', " "'C:\\\\path\\to\\file', " + "'https://example.com/path/to/file', " "'s3://bucket/path/to/file', " "'gs://bucket/path/to/file', " "'hdfs://example.com/path/to/file', " diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index bb12acdff..f6b2b20df 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -7,6 +7,7 @@ from dvc.remote.azure import RemoteAzure from dvc.remote.ssh import RemoteSSH from dvc.remote.hdfs import RemoteHDFS from dvc.remote.local import RemoteLOCAL +from dvc.remote.http import RemoteHTTP class DataCloud(object): @@ -17,6 +18,8 @@ class DataCloud(object): 'ssh': RemoteSSH, 'hdfs': RemoteHDFS, 'local': RemoteLOCAL, + 'http': RemoteHTTP, + 'https': RemoteHTTP, } def __init__(self, project, config=None): diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 4c0ae0a53..590787726 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -14,24 +14,33 @@ from dvc.dependency.gs import DependencyGS from dvc.dependency.local import DependencyLOCAL from dvc.dependency.hdfs import DependencyHDFS from dvc.dependency.ssh import DependencySSH +from dvc.dependency.http import DependencyHTTP from dvc.remote import Remote from dvc.remote.local import RemoteLOCAL from dvc.remote.s3 import RemoteS3 from dvc.remote.hdfs import RemoteHDFS -DEPS = [DependencyHDFS, - DependencyS3, - DependencyGS, - DependencySSH, - DependencyLOCAL] +DEPS = [ + DependencyGS, + DependencyHDFS, + DependencyHTTP, + DependencyLOCAL, + DependencyS3, + DependencySSH, +] -DEP_MAP = {'': DependencyLOCAL, - 'ssh': DependencySSH, - 's3': DependencyS3, - 'gs': DependencyGS, - 'hdfs': DependencyHDFS} +DEP_MAP = { + '': DependencyLOCAL, + 'ssh': DependencySSH, + 's3': DependencyS3, + 'gs': DependencyGS, + 'hdfs': DependencyHDFS, + 'http': DependencyHTTP, + 'https': DependencyHTTP, +} +# We are skipping RemoteHTTP.PARAM_ETAG because is the same as RemoteS3 SCHEMA = { DependencyBase.PARAM_PATH: str, schema.Optional(RemoteLOCAL.PARAM_MD5): schema.Or(str, None), diff --git a/dvc/dependency/http.py b/dvc/dependency/http.py new file mode 100644 index 000000000..23f52b316 --- /dev/null +++ b/dvc/dependency/http.py @@ -0,0 +1,36 @@ +try: + from urlparse import urlparse, urljoin +except ImportError: + from urllib.parse import urlparse, urljoin + +from dvc.dependency.base import DependencyBase +from dvc.remote.http import RemoteHTTP + + +class DependencyHTTP(DependencyBase): + REGEX = RemoteHTTP.REGEX + + def __init__(self, stage, path, info=None, remote=None): + super(DependencyHTTP, self).__init__(stage, path) + + self.info = info + self.remote = remote or RemoteHTTP(stage.project, {}) + + if path.startswith('remote'): + path = urljoin(self.remote.cache_dir, urlparse(path).path) + + self.path_info = { + 'scheme': urlparse(path).scheme, + 'url': path, + } + + def save(self): + self.info = self.remote.save_info(self.path_info) + + def dumpd(self): + ret = self.info + ret[self.PARAM_PATH] = self.path + return ret + + def changed(self): + return self.info != self.remote.save_info(self.path_info) diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index f8dd499f1..4ca6cd14c 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -15,7 +15,7 @@ class DependencyS3(DependencyBase): def __init__(self, stage, path, info=None, remote=None): super(DependencyS3, self).__init__(stage, path) self.info = info - self.remote = remote if remote else RemoteS3(stage.project, {}) + self.remote = remote or RemoteS3(stage.project, {}) bucket = remote.bucket if remote else urlparse(path).netloc key = urlparse(path).path.lstrip('/') diff --git a/dvc/remote/__init__.py b/dvc/remote/__init__.py index 6b71f1d8a..f385108e9 100644 --- a/dvc/remote/__init__.py +++ b/dvc/remote/__init__.py @@ -1,15 +1,24 @@ -from dvc.remote.local import RemoteLOCAL -from dvc.remote.s3 import RemoteS3 +from dvc.remote.azure import RemoteAzure from dvc.remote.gs import RemoteGS from dvc.remote.hdfs import RemoteHDFS +from dvc.remote.local import RemoteLOCAL +from dvc.remote.s3 import RemoteS3 from dvc.remote.ssh import RemoteSSH -from dvc.remote.azure import RemoteAzure +from dvc.remote.http import RemoteHTTP from dvc.config import Config from dvc.exceptions import UnsupportedRemoteError -REMOTES = [RemoteHDFS, RemoteSSH, RemoteS3, RemoteGS, RemoteAzure, RemoteLOCAL] +REMOTES = [ + RemoteAzure, + RemoteGS, + RemoteHDFS, + RemoteHTTP, + RemoteLOCAL, + RemoteS3, + RemoteSSH, +] def supported_url(url): diff --git a/dvc/remote/http.py b/dvc/remote/http.py new file mode 100644 index 000000000..10390715b --- /dev/null +++ b/dvc/remote/http.py @@ -0,0 +1,142 @@ +import os +import threading +import requests +import posixpath + +from dvc.logger import Logger +from dvc.progress import progress +from dvc.exceptions import DvcException +from dvc.config import Config +from dvc.remote.base import RemoteBase + + +class ProgressBarCallback(object): + def __init__(self, name, total): + self.name = name + self.total = total + self.current = 0 + self.lock = threading.Lock() + + def __call__(self, byts): + with self.lock: + self.current += byts + progress.update_target(self.name, self.current, self.total) + + +class RemoteHTTP(RemoteBase): + scheme = 'http' + REGEX = r'^https?://.*$' + REQUEST_TIMEOUT = 10 + CHUNK_SIZE = 1000000 # Megabyte + PARAM_ETAG = 'etag' + + def __init__(self, project, config): + self.project = project + self.cache_dir = config.get(Config.SECTION_REMOTE_URL) + + @property + def prefix(self): + return self.cache_dir + + def download(self, + from_infos, + to_infos, + no_progress_bar=False, + names=None): + names = self._verify_path_args(to_infos, from_infos, names) + + for to_info, from_info, name in zip(to_infos, from_infos, names): + if from_info['scheme'] not in ['http', 'https']: + raise NotImplementedError + + if to_info['scheme'] != 'local': + raise NotImplementedError + + msg = "Downloading '{}' to '{}'".format(from_info['url'], + to_info['path']) + Logger.debug(msg) + + tmp_file = self.tmp_file(to_info['path']) + if not name: + name = os.path.basename(to_info['path']) + + self._makedirs(to_info['path']) + + total = self._content_length(from_info['url']) + + if no_progress_bar or not total: + cb = None + else: + cb = ProgressBarCallback(name, total) + + try: + self._download_to(from_info['url'], tmp_file, callback=cb) + except Exception as exc: + msg = "Failed to download '{}'".format(from_info['url']) + Logger.warn(msg, exc) + continue + + os.rename(tmp_file, to_info['path']) + + if not no_progress_bar: + progress.finish_target(name) + + def exists(self, path_infos): + return [ + bool(self._request('HEAD', path_info.get('url'))) + for path_info in path_infos + ] + + def save_info(self, path_info): + if path_info['scheme'] not in ['http', 'https']: + raise NotImplementedError + + return {self.PARAM_ETAG: self._etag(path_info['url'])} + + def md5s_to_path_infos(self, md5s): + return [ + { + 'scheme': 'http', + 'url': posixpath.join(self.prefix, md5[0:2], md5[2:]), + } + for md5 in md5s + ] + + def _content_length(self, url): + return self._request('HEAD', url).headers.get('Content-Length') + + def _etag(self, url): + etag = self._request('HEAD', url).headers.get('ETag') + + if not etag: + raise DvcException("Could not find an ETag for '{}'".format(url)) + + if etag.startswith('W/'): + raise DvcException( + "Weak ETags are not supported." + " (Etag: '{etag}', URL: '{url}')".format(etag=etag, url=url) + ) + + return etag + + def _download_to(self, url, file, callback=None): + r = self._request('GET', url, stream=True) + + with open(file, 'wb') as fd: + bytes_transfered = 0 + + for chunk in r.iter_content(chunk_size=self.CHUNK_SIZE): + fd.write(chunk) + + if callback: + bytes_transfered += len(chunk) + callback(bytes_transfered) + + def _request(self, method, url, **kwargs): + kwargs.setdefault('allow_redirects', True) + kwargs.setdefault('timeout', self.REQUEST_TIMEOUT) + + try: + return requests.request(method, url, **kwargs) + except requests.exceptions.RequestException: + raise DvcException('Could not perform a {} request'.format(method)) diff --git a/dvc/remote/local.py b/dvc/remote/local.py index b969ee992..40e54372d 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -451,6 +451,13 @@ class RemoteLOCAL(RemoteBase): return {self.PARAM_MD5: self.state.update(path_info['path'])} def changed(self, path_info, checksum_info): + """ + A file is considered changed if: + - It doesn't exist on the working directory (was unlinked) + - Checksum is not computed (saving a new file) + - The checkusm stored in the State is different from the given one + - There's no file in the cache + """ if not self.exists([path_info])[0]: return True diff --git a/dvc/state.py b/dvc/state.py index 0e81a87ff..ce1dd4a90 100644 --- a/dvc/state.py +++ b/dvc/state.py @@ -267,6 +267,10 @@ class State(object): return System.inode(path) def _do_update(self, path): + """ + Search for the current inode on the table or update it with + gathered information. + """ if not os.path.exists(path): return (None, None) diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index 023c34d9e..ac8dcff44 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -11,7 +11,7 @@ import platform from dvc.main import main from dvc.config import Config, ConfigError from dvc.data_cloud import (DataCloud, RemoteS3, RemoteGS, RemoteAzure, - RemoteLOCAL, RemoteSSH, RemoteHDFS) + RemoteLOCAL, RemoteSSH, RemoteHDFS, RemoteHTTP) from dvc.remote.base import STATUS_OK, STATUS_NEW, STATUS_DELETED from tests.basic_env import TestDvc @@ -163,6 +163,7 @@ class TestDataCloud(TestDvc): clist = [('s3://mybucket/', RemoteS3), ('gs://mybucket/', RemoteGS), ('ssh://user@localhost:/', RemoteSSH), + ('http://localhost:8000/', RemoteHTTP), (TestDvc.mkdtemp(), RemoteLOCAL)] if _should_test_azure(): diff --git a/tests/test_remote.py b/tests/test_remote.py index e91d90e0c..9f1f575bc 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -28,18 +28,18 @@ class TestRemote(TestDvc): self.assertEqual(main(['remote', 'list']), 0) - def test_failed_write(self): - class A(object): - local = False - name = 'myremote' - url = 's3://remote' - unset = False - - args = A() - cmd = CmdRemoteAdd(args) - - cmd.configobj.write = None - ret = cmd.run() + def test_failed_write(self): + class A(object): + local = False + name = 'myremote' + url = 's3://remote' + unset = False + + args = A() + cmd = CmdRemoteAdd(args) + + cmd.configobj.write = None + ret = cmd.run() self.assertNotEqual(ret, 0) def test_unsupported(self): diff --git a/tests/test_repro.py b/tests/test_repro.py index a1976c08d..b26cbf51f 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -7,6 +7,11 @@ import getpass import posixpath from subprocess import Popen, PIPE +try: + from urlparse import urljoin +except ImportError: + from urllib.parse import urljoin + import boto3 import uuid from google.cloud import storage as gc @@ -24,6 +29,7 @@ from tests.test_data_cloud import _should_test_aws, TEST_AWS_REPO_BUCKET from tests.test_data_cloud import _should_test_gcp, TEST_GCP_REPO_BUCKET from tests.test_data_cloud import _should_test_ssh, _should_test_hdfs from tests.test_data_cloud import sleep +from tests.utils.httpd import StaticFileServer class TestRepro(TestDvc): @@ -59,7 +65,7 @@ class TestReproDepUnderDir(TestDvc): self.assertTrue(self.dir_stage is not None) sleep() - + self.file1 = 'file1' self.file1_stage = self.file1 + '.dvc' self.dvc.run(fname=self.file1_stage, @@ -404,7 +410,7 @@ class TestReproChangedDirData(TestDvc): fd.write("import os; import sys; import shutil; shutil.copytree(sys.argv[1], sys.argv[2])") sleep() - + stage = self.dvc.run(outs=[dir_name], deps=[self.DATA_DIR, dir_code], cmd="python {} {} {}".format(dir_code, @@ -421,12 +427,12 @@ class TestReproChangedDirData(TestDvc): fd.write('add') sleep() - + stages = self.dvc.reproduce(stage.path) self.assertEqual(len(stages), 1) self.assertTrue(stages[0] is not None) sleep() - + # Check that dvc indeed registers changed output dir shutil.move(self.BAR, dir_name) sleep() @@ -463,7 +469,7 @@ class TestCmdRepro(TestReproChangedData): ret = main(['status']) self.assertEqual(ret, 0) - + ret = main(['repro', self.file1_stage]) self.assertEqual(ret, 0) @@ -738,7 +744,7 @@ class TestReproExternalSSH(TestReproExternalBase): print(err) self.assertEqual(p.returncode, 0) - + class TestReproExternalLOCAL(TestReproExternalBase): def setUp(self): super(TestReproExternalLOCAL, self).setUp() @@ -783,6 +789,73 @@ class TestReproExternalLOCAL(TestReproExternalBase): fd.write(body) +class TestReproExternalHTTP(TestReproExternalBase): + _external_cache_id = None + + @property + def remote(self): + return 'http://localhost:8000/' + + @property + def local_cache(self): + return os.path.join(self.dvc.dvc_dir, 'cache') + + @property + def external_cache_id(self): + if not self._external_cache_id: + self._external_cache_id = str(uuid.uuid4()) + + return self._external_cache_id + + @property + def external_cache(self): + return urljoin(self.remote, self.external_cache_id) + + def test(self): + ret1 = main(['remote', 'add', 'mycache', self.external_cache]) + ret2 = main(['remote', 'add', 'myremote', self.remote]) + self.assertEqual(ret1, 0) + self.assertEqual(ret2, 0) + + self.dvc = Project('.') + + # Import + with StaticFileServer() as server: + import_url = urljoin(self.remote, self.FOO) + import_output = 'imported_file' + import_stage = self.dvc.imp(import_url, import_output) + + self.assertTrue(os.path.exists(import_output)) + self.assertTrue(filecmp.cmp(import_output, self.FOO, shallow=False)) + + # Run --deps + with StaticFileServer() as server: + run_dependency = urljoin(self.remote, self.BAR) + run_output = 'remote_file' + cmd = 'open("{}", "w+")'.format(run_output) + + with open('create-output.py', 'w') as fd: + fd.write(cmd) + + run_stage = self.dvc.run(deps=[run_dependency], + outs=[run_output], + cmd='python create-output.py') + + self.assertTrue(os.path.exists(run_output)) + + # Pull + self.dvc.remove(import_stage.path, outs_only=True) + self.assertFalse(os.path.exists(import_output)) + + shutil.move(self.local_cache, self.external_cache_id) + self.assertFalse(os.path.exists(self.local_cache)) + + with StaticFileServer() as server: + self.dvc.pull(import_stage.path, remote='mycache') + + self.assertTrue(os.path.exists(import_output)) + + class TestReproShell(TestDvc): def test(self): if os.name == 'nt': diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/utils/httpd.py b/tests/utils/httpd.py new file mode 100644 index 000000000..3ea180446 --- /dev/null +++ b/tests/utils/httpd.py @@ -0,0 +1,35 @@ +import hashlib +import os +import threading + +try: + from http.server import HTTPServer, SimpleHTTPRequestHandler +except ImportError: + from BaseHTTPServer import HTTPServer + from SimpleHTTPServer import SimpleHTTPRequestHandler + + +class ETagHandler(SimpleHTTPRequestHandler): + def end_headers(self): + file = self.translate_path(self.path) + + if not os.path.isdir(file) and os.path.exists(file): + with open(file, 'r') as fd: + etag = hashlib.md5(fd.read().encode('utf8')).hexdigest() + self.send_header('ETag', etag) + + SimpleHTTPRequestHandler.end_headers(self) + + +class StaticFileServer(): + def __init__(self): + self.httpd = HTTPServer(('localhost', 8000), ETagHandler) + + def __enter__(self): + self.server_thread = threading.Thread(target=self.httpd.serve_forever) + self.server_thread.daemon = True + self.server_thread.start() + + def __exit__(self, *args): + self.httpd.shutdown() + self.httpd.server_close()