Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-08 15:28:26 +01:00
Code Issues Packages Projects Releases Wiki Activity

remote: HTTP implemented (support only for downloading) (#1350)

Browse Source 
* remote: HTTP implemented (support only for downloading)

* Remove duplicated content in the schema

* Chomp trailing spaces in test_remote

* remote http: Implement pull/fetch

* add some tests

* Update dvc/remote/http.py

Co-Authored-By: mroutis <[email protected]>

* Update dvc/remote/http.py

Co-Authored-By: mroutis <[email protected]>

* correct test

* rehprase error message

* add __init__ for utils

* try to fix windows test
This commit is contained in:
Mr. Outis
2018-11-24 21:22:05 +02:00
committed by Ruslan Kuprieiev
parent f3f5837e20
commit 1e3d752917
14 changed files with 354 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cli.py
+1
View File
@@ -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', "
dvc/data_cloud.py
+3
View File
@@ -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):
dvc/dependency/__init__.py
+19 -10
View File
@@ -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),
dvc/dependency/http.py
+36
View File
@@ -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)
dvc/dependency/s3.py
+1 -1
View File
@@ -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('/')
dvc/remote/__init__.py
+13 -4
View File
@@ -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):
dvc/remote/http.py
+142
View File
@@ -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))
dvc/remote/local.py
+7
View File
@@ -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
dvc/state.py
+4
View File
@@ -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)
tests/test_data_cloud.py
+2 -1
View File
@@ -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():
tests/test_remote.py
+12 -12
View File
@@ -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):
tests/test_repro.py
+79 -6
View File
@@ -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':
tests/utils/__init__.py
View File
tests/utils/httpd.py
+35
View File
@@ -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()