We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Number of already present files should be reflected in push progress
Hey guys, IMHO the number of pushed files should also include the files already found in the remote cache.

For example, when I upload 2 new files out of my cache of 1000 files, the progress should say 1000/1000 at the end, instead of 2/1000.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dvc/cloud/base.py`
Content:
```
1 import os
2 import re
3 import tempfile
4 
5 from dvc.logger import Logger
6 from dvc.exceptions import DvcException
7 from dvc.config import Config, ConfigError
8 from dvc.cache import Cache
9 
10 
11 STATUS_UNKNOWN = 0
12 STATUS_OK = 1
13 STATUS_MODIFIED = 2
14 STATUS_NEW = 3
15 STATUS_DELETED = 4
16 
17 
18 STATUS_MAP = {
19     # (local_exists, remote_exists, cmp)
20     (True, True, True)  : STATUS_OK,
21     (True, True, False) : STATUS_MODIFIED,
22     (True, False, None) : STATUS_NEW,
23     (False, True, None) : STATUS_DELETED,
24 }
25 
26 
27 class DataCloudError(DvcException):
28     """ Data Cloud exception """
29     def __init__(self, msg):
30         super(DataCloudError, self).__init__('Data sync error: {}'.format(msg))
31 
32 
33 class CloudSettings(object):
34     def __init__(self, cache=None, global_storage_path=None, cloud_config=None):
35         self.cache = cache
36         self.cloud_config = cloud_config
37         self.global_storage_path = global_storage_path
38 
39 
40 class DataCloudBase(object):
41     """ Base class for DataCloud """
42 
43     REGEX = r''
44 
45     def __init__(self, cloud_settings):
46         self._cloud_settings = cloud_settings
47 
48     @property
49     def url(self):
50         config = self._cloud_settings.cloud_config
51         return config.get(Config.SECTION_REMOTE_URL, None)
52 
53     @classmethod
54     def match(cls, url):
55         return re.match(cls.REGEX, url)
56 
57     def group(self, group):
58         return self.match(self.url).group(group)
59 
60     @property
61     def path(self):
62         return self.group('path')
63 
64     @classmethod
65     def supported(cls, url):
66         return cls.match(url) != None
67 
68     # backward compatibility
69     @property
70     def storage_path(self):
71         """ get storage path
72 
73         Precedence: Storage, then cloud specific
74         """
75 
76         if self._cloud_settings.global_storage_path:
77             return self._cloud_settings.global_storage_path
78 
79         if not self.url:
80             path = self._cloud_settings.cloud_config.get(Config.SECTION_CORE_STORAGEPATH, None)
81             if path:
82                 Logger.warn('Using obsoleted config format. Consider updating.')
83         else:
84             path = self.path
85 
86         return path
87 
88     def _storage_path_parts(self):
89         """
90         Split storage path into parts. I.e. 'dvc-test/myrepo' -> ['dvc', 'myrepo']
91         """
92         return self.storage_path.strip('/').split('/', 1)
93 
94     @property
95     def storage_bucket(self):
96         """ Data -> StoragePath takes precedence; if doesn't exist, use cloud-specific """
97         return self._storage_path_parts()[0]
98 
99     @property
100     def storage_prefix(self):
101         """
102         Prefix within the bucket. I.e. 'myrepo' in 'dvc-test/myrepo'.
103         """
104         parts = self._storage_path_parts()
105         if len(parts) > 1:
106             return parts[1]
107         return ''
108 
109     def cache_file_key(self, fname):
110         """ Key of a file within the bucket """
111         relpath = os.path.relpath(fname, self._cloud_settings.cache.cache_dir)
112         relpath = relpath.replace('\\', '/')
113         return '{}/{}'.format(self.storage_prefix, relpath).strip('/')
114 
115     @staticmethod
116     def tmp_file(fname):
117         """ Temporary name for a partial download """
118         return fname + '.part'
119 
120     def sanity_check(self):
121         """
122         Cloud-specific method to check config for basic requirements.
123         """
124         pass
125 
126     def _push_key(self, key, path):
127         pass
128 
129     def collect(self, arg):
130         path, local = arg
131         ret = [path]
132 
133         if not Cache.is_dir_cache(path):
134             return ret
135 
136         if local:
137             if not os.path.isfile(path):
138                 return ret
139             dir_path = path
140         else:
141             key = self._get_key(path)
142             if not key:
143                 Logger.debug("File '{}' does not exist in the cloud".format(path))
144                 return ret
145             tmp = os.path.join(tempfile.mkdtemp(), os.path.basename(path))
146             self._pull_key(key, tmp, no_progress_bar=True)
147             dir_path = tmp
148 
149         for relpath, md5 in Cache.get_dir_cache(dir_path).items():
150             cache = self._cloud_settings.cache.get(md5)
151             ret.append(cache)
152 
153         return ret
154 
155     def _cmp_checksum(self, blob, fname):
156         md5 = self._cloud_settings.cache.path_to_md5(fname)
157         if self._cloud_settings.cache.state.changed(fname, md5=md5):
158             return False
159 
160         return True
161 
162     def push(self, path):
163         key = self._get_key(path)
164         if key:
165             Logger.debug("File '{}' already uploaded to the cloud. Validating checksum...".format(path))
166             if self._cmp_checksum(key, path):
167                 Logger.debug('File checksum matches. No uploading is needed.')
168                 return []
169             Logger.debug('Checksum mismatch. Reuploading is required.')
170 
171         key = self._new_key(path)
172         return self._push_key(key, path)
173 
174     def _makedirs(self, fname):
175         dname = os.path.dirname(fname)
176         try:
177             os.makedirs(dname)
178         except OSError as e:
179             if e.errno != os.errno.EEXIST:
180                 raise
181 
182     def _pull_key(self, key, path, no_progress_bar=False):
183         """ Cloud-specific method of pulling keys """
184         pass
185 
186     def _get_key(self, path):
187         """ Cloud-specific method of getting keys """
188         pass
189 
190     def pull(self, path):
191         """ Generic method for pulling data from the cloud """
192         key = self._get_key(path)
193         if not key:
194             Logger.error("File '{}' does not exist in the cloud".format(path))
195             return None
196 
197         return self._pull_key(key, path)
198 
199     def _status(self, key, path):
200         remote_exists = key != None
201         local_exists = os.path.exists(path)
202 
203         diff = None
204         if remote_exists and local_exists:
205             diff = self._cmp_checksum(key, path)
206 
207         return STATUS_MAP.get((local_exists, remote_exists, diff), STATUS_UNKNOWN)
208 
209     def status(self, path):
210         """
211         Generic method for checking data item status.
212         """
213         key = self._get_key(path)
214         if not key:
215             return STATUS_NEW
216 
217         return self._status(key, path)
218 
219     def connect(self):
220         pass
221 
222     def disconnect(self):
223         pass
224
```

Path: `dvc/cloud/data_cloud.py`
Content:
```
1 import re
2 from multiprocessing.pool import ThreadPool
3 
4 from dvc.logger import Logger
5 from dvc.exceptions import DvcException
6 from dvc.config import Config, ConfigError
7 from dvc.utils import map_progress
8 
9 from dvc.cloud.aws import DataCloudAWS
10 from dvc.cloud.gcp import DataCloudGCP
11 from dvc.cloud.ssh import DataCloudSSH
12 from dvc.cloud.local import DataCloudLOCAL
13 from dvc.cloud.base import DataCloudBase, CloudSettings
14 
15 
16 class DataCloud(object):
17     """ Generic class to do initial config parsing and redirect to proper DataCloud methods """
18 
19     CLOUD_MAP = {
20         'aws'   : DataCloudAWS,
21         'gcp'   : DataCloudGCP,
22         'ssh'   : DataCloudSSH,
23         'local' : DataCloudLOCAL,
24     }
25 
26     def __init__(self, cache=None, config=None):
27         self._cache = cache
28         self._config = config
29 
30         remote = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_REMOTE, '')
31         if remote == '':
32             if config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, None):
33                 # backward compatibility
34                 Logger.warn('Using obsoleted config format. Consider updating.')
35                 self._cloud = self.__init__compat()
36             else:
37                 self._cloud = None
38             return
39 
40         self._cloud = self._init_remote(remote)
41 
42     @staticmethod
43     def supported(url):
44         for cloud in DataCloud.CLOUD_MAP.values():
45             if cloud.supported(url):
46                 return cloud
47         return None
48 
49     def _init_remote(self, remote):
50         section = Config.SECTION_REMOTE_FMT.format(remote)
51         cloud_config = self._config.get(section, None)
52         if not cloud_config:
53             raise ConfigError("Can't find remote section '{}' in config".format(section))
54 
55         url = cloud_config[Config.SECTION_REMOTE_URL]
56         cloud_type = self.supported(url)
57         if not cloud_type:
58             raise ConfigError("Unsupported url '{}'".format(url))
59 
60         return self._init_cloud(cloud_config, cloud_type)
61 
62     def __init__compat(self):
63         cloud_name = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, '').strip().lower()
64         if cloud_name == '':
65             self._cloud = None
66             return
67 
68         cloud_type = self.CLOUD_MAP.get(cloud_name, None)
69         if not cloud_type:
70             raise ConfigError('Wrong cloud type %s specified' % cloud_name)
71 
72         cloud_config = self._config.get(cloud_name, None)
73         if not cloud_config:
74             raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_name)
75 
76         return self._init_cloud(cloud_config, cloud_type)
77 
78     def _init_cloud(self, cloud_config, cloud_type):
79         global_storage_path = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_STORAGEPATH, None)
80         if global_storage_path:
81             Logger.warn('Using obsoleted config format. Consider updating.')
82 
83         cloud_settings = CloudSettings(cache=self._cache,
84                                        global_storage_path=global_storage_path,
85                                        cloud_config=cloud_config)
86 
87         cloud = cloud_type(cloud_settings)
88         cloud.sanity_check()
89         return cloud
90 
91     def _collect(self, cloud, targets, jobs, local):
92         collected = set()
93         pool = ThreadPool(processes=jobs)
94         args = zip(targets, [local]*len(targets))
95         ret = pool.map(cloud.collect, args)
96 
97         for r in ret:
98             collected |= set(r)
99 
100         return collected
101 
102     def _map_targets(self, func, targets, jobs, collect_local=False, collect_cloud=False, remote=None):
103         """
104         Process targets as data items in parallel.
105         """
106 
107         if not remote:
108             cloud = self._cloud
109         else:
110             cloud = self._init_remote(remote)
111 
112         if not cloud:
113             return
114 
115         cloud.connect()
116 
117         collected = set()
118         if collect_local:
119             collected |= self._collect(cloud, targets, jobs, True)
120         if collect_cloud:
121             collected |= self._collect(cloud, targets, jobs, False)
122 
123         ret = map_progress(getattr(cloud, func), list(collected), jobs)
124 
125         cloud.disconnect()
126 
127         return ret
128 
129     def push(self, targets, jobs=1, remote=None):
130         """
131         Push data items in a cloud-agnostic way.
132         """
133         return self._map_targets('push', targets, jobs, collect_local=True, remote=remote)
134 
135     def pull(self, targets, jobs=1, remote=None):
136         """
137         Pull data items in a cloud-agnostic way.
138         """
139         return self._map_targets('pull', targets, jobs, collect_cloud=True, remote=remote)
140 
141     def status(self, targets, jobs=1, remote=None):
142         """
143         Check status of data items in a cloud-agnostic way.
144         """
145         return self._map_targets('status', targets, jobs, True, True, remote=remote)
146
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py
--- a/dvc/cloud/base.py
+++ b/dvc/cloud/base.py
@@ -221,3 +221,9 @@
 
     def disconnect(self):
         pass
+
+    def __enter__(self):
+        self.connect()
+
+    def __exit__(self, type, value, tb):
+        self.disconnect()
diff --git a/dvc/cloud/data_cloud.py b/dvc/cloud/data_cloud.py
--- a/dvc/cloud/data_cloud.py
+++ b/dvc/cloud/data_cloud.py
@@ -11,6 +11,7 @@
 from dvc.cloud.ssh import DataCloudSSH
 from dvc.cloud.local import DataCloudLOCAL
 from dvc.cloud.base import DataCloudBase, CloudSettings
+from dvc.cloud.base import STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED
 
 
 class DataCloud(object):
@@ -99,47 +100,54 @@
 
         return collected
 
-    def _map_targets(self, func, targets, jobs, collect_local=False, collect_cloud=False, remote=None):
-        """
-        Process targets as data items in parallel.
-        """
+    def _get_cloud(self, remote):
+        if remote:
+            return self._init_remote(remote)
 
-        if not remote:
-            cloud = self._cloud
-        else:
-            cloud = self._init_remote(remote)
+        return self._cloud
 
+    def _filter(self, func, status, targets, jobs, remote):
+        cloud = self._get_cloud(remote)
         if not cloud:
-            return
-
-        cloud.connect()
-
-        collected = set()
-        if collect_local:
-            collected |= self._collect(cloud, targets, jobs, True)
-        if collect_cloud:
-            collected |= self._collect(cloud, targets, jobs, False)
+            return []
 
-        ret = map_progress(getattr(cloud, func), list(collected), jobs)
+        with cloud:
+            filtered = []
+            for t, s in self._status(cloud, targets, jobs):
+                if s == STATUS_MODIFIED or s == status:
+                    filtered.append(t)
 
-        cloud.disconnect()
-
-        return ret
+            return map_progress(getattr(cloud, func), filtered, jobs)
 
     def push(self, targets, jobs=1, remote=None):
         """
         Push data items in a cloud-agnostic way.
         """
-        return self._map_targets('push', targets, jobs, collect_local=True, remote=remote)
+        return self._filter('push', STATUS_NEW, targets, jobs, remote)
 
     def pull(self, targets, jobs=1, remote=None):
         """
         Pull data items in a cloud-agnostic way.
         """
-        return self._map_targets('pull', targets, jobs, collect_cloud=True, remote=remote)
+        return self._filter('pull', STATUS_DELETED, targets, jobs, remote)
+
+    def _collect_targets(self, cloud, targets, jobs=1):
+        collected = set()
+        collected |= self._collect(cloud, targets, jobs, True)
+        collected |= self._collect(cloud, targets, jobs, False)
+        return list(collected)
+
+    def _status(self, cloud, targets, jobs=1):
+        collected = self._collect_targets(cloud, targets, jobs)
+        return map_progress(cloud.status, collected, jobs)
 
     def status(self, targets, jobs=1, remote=None):
         """
         Check status of data items in a cloud-agnostic way.
         """
-        return self._map_targets('status', targets, jobs, True, True, remote=remote)
+        cloud = self._get_cloud(remote)
+        if not cloud:
+            return []
+
+        with cloud:
+            return self._status(cloud, targets, jobs)

{"golden_diff": "diff --git a/dvc/cloud/base.py b/dvc/cloud/base.py\n--- a/dvc/cloud/base.py\n+++ b/dvc/cloud/base.py\n@@ -221,3 +221,9 @@\n \n     def disconnect(self):\n         pass\n+\n+    def __enter__(self):\n+        self.connect()\n+\n+    def __exit__(self, type, value, tb):\n+        self.disconnect()\ndiff --git a/dvc/cloud/data_cloud.py b/dvc/cloud/data_cloud.py\n--- a/dvc/cloud/data_cloud.py\n+++ b/dvc/cloud/data_cloud.py\n@@ -11,6 +11,7 @@\n from dvc.cloud.ssh import DataCloudSSH\n from dvc.cloud.local import DataCloudLOCAL\n from dvc.cloud.base import DataCloudBase, CloudSettings\n+from dvc.cloud.base import STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED\n \n \n class DataCloud(object):\n@@ -99,47 +100,54 @@\n \n         return collected\n \n-    def _map_targets(self, func, targets, jobs, collect_local=False, collect_cloud=False, remote=None):\n-        \"\"\"\n-        Process targets as data items in parallel.\n-        \"\"\"\n+    def _get_cloud(self, remote):\n+        if remote:\n+            return self._init_remote(remote)\n \n-        if not remote:\n-            cloud = self._cloud\n-        else:\n-            cloud = self._init_remote(remote)\n+        return self._cloud\n \n+    def _filter(self, func, status, targets, jobs, remote):\n+        cloud = self._get_cloud(remote)\n         if not cloud:\n-            return\n-\n-        cloud.connect()\n-\n-        collected = set()\n-        if collect_local:\n-            collected |= self._collect(cloud, targets, jobs, True)\n-        if collect_cloud:\n-            collected |= self._collect(cloud, targets, jobs, False)\n+            return []\n \n-        ret = map_progress(getattr(cloud, func), list(collected), jobs)\n+        with cloud:\n+            filtered = []\n+            for t, s in self._status(cloud, targets, jobs):\n+                if s == STATUS_MODIFIED or s == status:\n+                    filtered.append(t)\n \n-        cloud.disconnect()\n-\n-        return ret\n+            return map_progress(getattr(cloud, func), filtered, jobs)\n \n     def push(self, targets, jobs=1, remote=None):\n         \"\"\"\n         Push data items in a cloud-agnostic way.\n         \"\"\"\n-        return self._map_targets('push', targets, jobs, collect_local=True, remote=remote)\n+        return self._filter('push', STATUS_NEW, targets, jobs, remote)\n \n     def pull(self, targets, jobs=1, remote=None):\n         \"\"\"\n         Pull data items in a cloud-agnostic way.\n         \"\"\"\n-        return self._map_targets('pull', targets, jobs, collect_cloud=True, remote=remote)\n+        return self._filter('pull', STATUS_DELETED, targets, jobs, remote)\n+\n+    def _collect_targets(self, cloud, targets, jobs=1):\n+        collected = set()\n+        collected |= self._collect(cloud, targets, jobs, True)\n+        collected |= self._collect(cloud, targets, jobs, False)\n+        return list(collected)\n+\n+    def _status(self, cloud, targets, jobs=1):\n+        collected = self._collect_targets(cloud, targets, jobs)\n+        return map_progress(cloud.status, collected, jobs)\n \n     def status(self, targets, jobs=1, remote=None):\n         \"\"\"\n         Check status of data items in a cloud-agnostic way.\n         \"\"\"\n-        return self._map_targets('status', targets, jobs, True, True, remote=remote)\n+        cloud = self._get_cloud(remote)\n+        if not cloud:\n+            return []\n+\n+        with cloud:\n+            return self._status(cloud, targets, jobs)\n", "issue": "Number of already present files should be reflected in push progress\nHey guys, IMHO the number of pushed files should also include the files already found in the remote cache.\r\n\r\nFor example, when I upload 2 new files out of my cache of 1000 files, the progress should say 1000/1000 at the end, instead of 2/1000.\n", "before_files": [{"content": "import os\nimport re\nimport tempfile\n\nfrom dvc.logger import Logger\nfrom dvc.exceptions import DvcException\nfrom dvc.config import Config, ConfigError\nfrom dvc.cache import Cache\n\n\nSTATUS_UNKNOWN = 0\nSTATUS_OK = 1\nSTATUS_MODIFIED = 2\nSTATUS_NEW = 3\nSTATUS_DELETED = 4\n\n\nSTATUS_MAP = {\n    # (local_exists, remote_exists, cmp)\n    (True, True, True)  : STATUS_OK,\n    (True, True, False) : STATUS_MODIFIED,\n    (True, False, None) : STATUS_NEW,\n    (False, True, None) : STATUS_DELETED,\n}\n\n\nclass DataCloudError(DvcException):\n    \"\"\" Data Cloud exception \"\"\"\n    def __init__(self, msg):\n        super(DataCloudError, self).__init__('Data sync error: {}'.format(msg))\n\n\nclass CloudSettings(object):\n    def __init__(self, cache=None, global_storage_path=None, cloud_config=None):\n        self.cache = cache\n        self.cloud_config = cloud_config\n        self.global_storage_path = global_storage_path\n\n\nclass DataCloudBase(object):\n    \"\"\" Base class for DataCloud \"\"\"\n\n    REGEX = r''\n\n    def __init__(self, cloud_settings):\n        self._cloud_settings = cloud_settings\n\n    @property\n    def url(self):\n        config = self._cloud_settings.cloud_config\n        return config.get(Config.SECTION_REMOTE_URL, None)\n\n    @classmethod\n    def match(cls, url):\n        return re.match(cls.REGEX, url)\n\n    def group(self, group):\n        return self.match(self.url).group(group)\n\n    @property\n    def path(self):\n        return self.group('path')\n\n    @classmethod\n    def supported(cls, url):\n        return cls.match(url) != None\n\n    # backward compatibility\n    @property\n    def storage_path(self):\n        \"\"\" get storage path\n\n        Precedence: Storage, then cloud specific\n        \"\"\"\n\n        if self._cloud_settings.global_storage_path:\n            return self._cloud_settings.global_storage_path\n\n        if not self.url:\n            path = self._cloud_settings.cloud_config.get(Config.SECTION_CORE_STORAGEPATH, None)\n            if path:\n                Logger.warn('Using obsoleted config format. Consider updating.')\n        else:\n            path = self.path\n\n        return path\n\n    def _storage_path_parts(self):\n        \"\"\"\n        Split storage path into parts. I.e. 'dvc-test/myrepo' -> ['dvc', 'myrepo']\n        \"\"\"\n        return self.storage_path.strip('/').split('/', 1)\n\n    @property\n    def storage_bucket(self):\n        \"\"\" Data -> StoragePath takes precedence; if doesn't exist, use cloud-specific \"\"\"\n        return self._storage_path_parts()[0]\n\n    @property\n    def storage_prefix(self):\n        \"\"\"\n        Prefix within the bucket. I.e. 'myrepo' in 'dvc-test/myrepo'.\n        \"\"\"\n        parts = self._storage_path_parts()\n        if len(parts) > 1:\n            return parts[1]\n        return ''\n\n    def cache_file_key(self, fname):\n        \"\"\" Key of a file within the bucket \"\"\"\n        relpath = os.path.relpath(fname, self._cloud_settings.cache.cache_dir)\n        relpath = relpath.replace('\\\\', '/')\n        return '{}/{}'.format(self.storage_prefix, relpath).strip('/')\n\n    @staticmethod\n    def tmp_file(fname):\n        \"\"\" Temporary name for a partial download \"\"\"\n        return fname + '.part'\n\n    def sanity_check(self):\n        \"\"\"\n        Cloud-specific method to check config for basic requirements.\n        \"\"\"\n        pass\n\n    def _push_key(self, key, path):\n        pass\n\n    def collect(self, arg):\n        path, local = arg\n        ret = [path]\n\n        if not Cache.is_dir_cache(path):\n            return ret\n\n        if local:\n            if not os.path.isfile(path):\n                return ret\n            dir_path = path\n        else:\n            key = self._get_key(path)\n            if not key:\n                Logger.debug(\"File '{}' does not exist in the cloud\".format(path))\n                return ret\n            tmp = os.path.join(tempfile.mkdtemp(), os.path.basename(path))\n            self._pull_key(key, tmp, no_progress_bar=True)\n            dir_path = tmp\n\n        for relpath, md5 in Cache.get_dir_cache(dir_path).items():\n            cache = self._cloud_settings.cache.get(md5)\n            ret.append(cache)\n\n        return ret\n\n    def _cmp_checksum(self, blob, fname):\n        md5 = self._cloud_settings.cache.path_to_md5(fname)\n        if self._cloud_settings.cache.state.changed(fname, md5=md5):\n            return False\n\n        return True\n\n    def push(self, path):\n        key = self._get_key(path)\n        if key:\n            Logger.debug(\"File '{}' already uploaded to the cloud. Validating checksum...\".format(path))\n            if self._cmp_checksum(key, path):\n                Logger.debug('File checksum matches. No uploading is needed.')\n                return []\n            Logger.debug('Checksum mismatch. Reuploading is required.')\n\n        key = self._new_key(path)\n        return self._push_key(key, path)\n\n    def _makedirs(self, fname):\n        dname = os.path.dirname(fname)\n        try:\n            os.makedirs(dname)\n        except OSError as e:\n            if e.errno != os.errno.EEXIST:\n                raise\n\n    def _pull_key(self, key, path, no_progress_bar=False):\n        \"\"\" Cloud-specific method of pulling keys \"\"\"\n        pass\n\n    def _get_key(self, path):\n        \"\"\" Cloud-specific method of getting keys \"\"\"\n        pass\n\n    def pull(self, path):\n        \"\"\" Generic method for pulling data from the cloud \"\"\"\n        key = self._get_key(path)\n        if not key:\n            Logger.error(\"File '{}' does not exist in the cloud\".format(path))\n            return None\n\n        return self._pull_key(key, path)\n\n    def _status(self, key, path):\n        remote_exists = key != None\n        local_exists = os.path.exists(path)\n\n        diff = None\n        if remote_exists and local_exists:\n            diff = self._cmp_checksum(key, path)\n\n        return STATUS_MAP.get((local_exists, remote_exists, diff), STATUS_UNKNOWN)\n\n    def status(self, path):\n        \"\"\"\n        Generic method for checking data item status.\n        \"\"\"\n        key = self._get_key(path)\n        if not key:\n            return STATUS_NEW\n\n        return self._status(key, path)\n\n    def connect(self):\n        pass\n\n    def disconnect(self):\n        pass\n", "path": "dvc/cloud/base.py"}, {"content": "import re\nfrom multiprocessing.pool import ThreadPool\n\nfrom dvc.logger import Logger\nfrom dvc.exceptions import DvcException\nfrom dvc.config import Config, ConfigError\nfrom dvc.utils import map_progress\n\nfrom dvc.cloud.aws import DataCloudAWS\nfrom dvc.cloud.gcp import DataCloudGCP\nfrom dvc.cloud.ssh import DataCloudSSH\nfrom dvc.cloud.local import DataCloudLOCAL\nfrom dvc.cloud.base import DataCloudBase, CloudSettings\n\n\nclass DataCloud(object):\n    \"\"\" Generic class to do initial config parsing and redirect to proper DataCloud methods \"\"\"\n\n    CLOUD_MAP = {\n        'aws'   : DataCloudAWS,\n        'gcp'   : DataCloudGCP,\n        'ssh'   : DataCloudSSH,\n        'local' : DataCloudLOCAL,\n    }\n\n    def __init__(self, cache=None, config=None):\n        self._cache = cache\n        self._config = config\n\n        remote = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_REMOTE, '')\n        if remote == '':\n            if config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, None):\n                # backward compatibility\n                Logger.warn('Using obsoleted config format. Consider updating.')\n                self._cloud = self.__init__compat()\n            else:\n                self._cloud = None\n            return\n\n        self._cloud = self._init_remote(remote)\n\n    @staticmethod\n    def supported(url):\n        for cloud in DataCloud.CLOUD_MAP.values():\n            if cloud.supported(url):\n                return cloud\n        return None\n\n    def _init_remote(self, remote):\n        section = Config.SECTION_REMOTE_FMT.format(remote)\n        cloud_config = self._config.get(section, None)\n        if not cloud_config:\n            raise ConfigError(\"Can't find remote section '{}' in config\".format(section))\n\n        url = cloud_config[Config.SECTION_REMOTE_URL]\n        cloud_type = self.supported(url)\n        if not cloud_type:\n            raise ConfigError(\"Unsupported url '{}'\".format(url))\n\n        return self._init_cloud(cloud_config, cloud_type)\n\n    def __init__compat(self):\n        cloud_name = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, '').strip().lower()\n        if cloud_name == '':\n            self._cloud = None\n            return\n\n        cloud_type = self.CLOUD_MAP.get(cloud_name, None)\n        if not cloud_type:\n            raise ConfigError('Wrong cloud type %s specified' % cloud_name)\n\n        cloud_config = self._config.get(cloud_name, None)\n        if not cloud_config:\n            raise ConfigError('Can\\'t find cloud section \\'[%s]\\' in config' % cloud_name)\n\n        return self._init_cloud(cloud_config, cloud_type)\n\n    def _init_cloud(self, cloud_config, cloud_type):\n        global_storage_path = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_STORAGEPATH, None)\n        if global_storage_path:\n            Logger.warn('Using obsoleted config format. Consider updating.')\n\n        cloud_settings = CloudSettings(cache=self._cache,\n                                       global_storage_path=global_storage_path,\n                                       cloud_config=cloud_config)\n\n        cloud = cloud_type(cloud_settings)\n        cloud.sanity_check()\n        return cloud\n\n    def _collect(self, cloud, targets, jobs, local):\n        collected = set()\n        pool = ThreadPool(processes=jobs)\n        args = zip(targets, [local]*len(targets))\n        ret = pool.map(cloud.collect, args)\n\n        for r in ret:\n            collected |= set(r)\n\n        return collected\n\n    def _map_targets(self, func, targets, jobs, collect_local=False, collect_cloud=False, remote=None):\n        \"\"\"\n        Process targets as data items in parallel.\n        \"\"\"\n\n        if not remote:\n            cloud = self._cloud\n        else:\n            cloud = self._init_remote(remote)\n\n        if not cloud:\n            return\n\n        cloud.connect()\n\n        collected = set()\n        if collect_local:\n            collected |= self._collect(cloud, targets, jobs, True)\n        if collect_cloud:\n            collected |= self._collect(cloud, targets, jobs, False)\n\n        ret = map_progress(getattr(cloud, func), list(collected), jobs)\n\n        cloud.disconnect()\n\n        return ret\n\n    def push(self, targets, jobs=1, remote=None):\n        \"\"\"\n        Push data items in a cloud-agnostic way.\n        \"\"\"\n        return self._map_targets('push', targets, jobs, collect_local=True, remote=remote)\n\n    def pull(self, targets, jobs=1, remote=None):\n        \"\"\"\n        Pull data items in a cloud-agnostic way.\n        \"\"\"\n        return self._map_targets('pull', targets, jobs, collect_cloud=True, remote=remote)\n\n    def status(self, targets, jobs=1, remote=None):\n        \"\"\"\n        Check status of data items in a cloud-agnostic way.\n        \"\"\"\n        return self._map_targets('status', targets, jobs, True, True, remote=remote)\n", "path": "dvc/cloud/data_cloud.py"}], "after_files": [{"content": "import os\nimport re\nimport tempfile\n\nfrom dvc.logger import Logger\nfrom dvc.exceptions import DvcException\nfrom dvc.config import Config, ConfigError\nfrom dvc.cache import Cache\n\n\nSTATUS_UNKNOWN = 0\nSTATUS_OK = 1\nSTATUS_MODIFIED = 2\nSTATUS_NEW = 3\nSTATUS_DELETED = 4\n\n\nSTATUS_MAP = {\n    # (local_exists, remote_exists, cmp)\n    (True, True, True)  : STATUS_OK,\n    (True, True, False) : STATUS_MODIFIED,\n    (True, False, None) : STATUS_NEW,\n    (False, True, None) : STATUS_DELETED,\n}\n\n\nclass DataCloudError(DvcException):\n    \"\"\" Data Cloud exception \"\"\"\n    def __init__(self, msg):\n        super(DataCloudError, self).__init__('Data sync error: {}'.format(msg))\n\n\nclass CloudSettings(object):\n    def __init__(self, cache=None, global_storage_path=None, cloud_config=None):\n        self.cache = cache\n        self.cloud_config = cloud_config\n        self.global_storage_path = global_storage_path\n\n\nclass DataCloudBase(object):\n    \"\"\" Base class for DataCloud \"\"\"\n\n    REGEX = r''\n\n    def __init__(self, cloud_settings):\n        self._cloud_settings = cloud_settings\n\n    @property\n    def url(self):\n        config = self._cloud_settings.cloud_config\n        return config.get(Config.SECTION_REMOTE_URL, None)\n\n    @classmethod\n    def match(cls, url):\n        return re.match(cls.REGEX, url)\n\n    def group(self, group):\n        return self.match(self.url).group(group)\n\n    @property\n    def path(self):\n        return self.group('path')\n\n    @classmethod\n    def supported(cls, url):\n        return cls.match(url) != None\n\n    # backward compatibility\n    @property\n    def storage_path(self):\n        \"\"\" get storage path\n\n        Precedence: Storage, then cloud specific\n        \"\"\"\n\n        if self._cloud_settings.global_storage_path:\n            return self._cloud_settings.global_storage_path\n\n        if not self.url:\n            path = self._cloud_settings.cloud_config.get(Config.SECTION_CORE_STORAGEPATH, None)\n            if path:\n                Logger.warn('Using obsoleted config format. Consider updating.')\n        else:\n            path = self.path\n\n        return path\n\n    def _storage_path_parts(self):\n        \"\"\"\n        Split storage path into parts. I.e. 'dvc-test/myrepo' -> ['dvc', 'myrepo']\n        \"\"\"\n        return self.storage_path.strip('/').split('/', 1)\n\n    @property\n    def storage_bucket(self):\n        \"\"\" Data -> StoragePath takes precedence; if doesn't exist, use cloud-specific \"\"\"\n        return self._storage_path_parts()[0]\n\n    @property\n    def storage_prefix(self):\n        \"\"\"\n        Prefix within the bucket. I.e. 'myrepo' in 'dvc-test/myrepo'.\n        \"\"\"\n        parts = self._storage_path_parts()\n        if len(parts) > 1:\n            return parts[1]\n        return ''\n\n    def cache_file_key(self, fname):\n        \"\"\" Key of a file within the bucket \"\"\"\n        relpath = os.path.relpath(fname, self._cloud_settings.cache.cache_dir)\n        relpath = relpath.replace('\\\\', '/')\n        return '{}/{}'.format(self.storage_prefix, relpath).strip('/')\n\n    @staticmethod\n    def tmp_file(fname):\n        \"\"\" Temporary name for a partial download \"\"\"\n        return fname + '.part'\n\n    def sanity_check(self):\n        \"\"\"\n        Cloud-specific method to check config for basic requirements.\n        \"\"\"\n        pass\n\n    def _push_key(self, key, path):\n        pass\n\n    def collect(self, arg):\n        path, local = arg\n        ret = [path]\n\n        if not Cache.is_dir_cache(path):\n            return ret\n\n        if local:\n            if not os.path.isfile(path):\n                return ret\n            dir_path = path\n        else:\n            key = self._get_key(path)\n            if not key:\n                Logger.debug(\"File '{}' does not exist in the cloud\".format(path))\n                return ret\n            tmp = os.path.join(tempfile.mkdtemp(), os.path.basename(path))\n            self._pull_key(key, tmp, no_progress_bar=True)\n            dir_path = tmp\n\n        for relpath, md5 in Cache.get_dir_cache(dir_path).items():\n            cache = self._cloud_settings.cache.get(md5)\n            ret.append(cache)\n\n        return ret\n\n    def _cmp_checksum(self, blob, fname):\n        md5 = self._cloud_settings.cache.path_to_md5(fname)\n        if self._cloud_settings.cache.state.changed(fname, md5=md5):\n            return False\n\n        return True\n\n    def push(self, path):\n        key = self._get_key(path)\n        if key:\n            Logger.debug(\"File '{}' already uploaded to the cloud. Validating checksum...\".format(path))\n            if self._cmp_checksum(key, path):\n                Logger.debug('File checksum matches. No uploading is needed.')\n                return []\n            Logger.debug('Checksum mismatch. Reuploading is required.')\n\n        key = self._new_key(path)\n        return self._push_key(key, path)\n\n    def _makedirs(self, fname):\n        dname = os.path.dirname(fname)\n        try:\n            os.makedirs(dname)\n        except OSError as e:\n            if e.errno != os.errno.EEXIST:\n                raise\n\n    def _pull_key(self, key, path, no_progress_bar=False):\n        \"\"\" Cloud-specific method of pulling keys \"\"\"\n        pass\n\n    def _get_key(self, path):\n        \"\"\" Cloud-specific method of getting keys \"\"\"\n        pass\n\n    def pull(self, path):\n        \"\"\" Generic method for pulling data from the cloud \"\"\"\n        key = self._get_key(path)\n        if not key:\n            Logger.error(\"File '{}' does not exist in the cloud\".format(path))\n            return None\n\n        return self._pull_key(key, path)\n\n    def _status(self, key, path):\n        remote_exists = key != None\n        local_exists = os.path.exists(path)\n\n        diff = None\n        if remote_exists and local_exists:\n            diff = self._cmp_checksum(key, path)\n\n        return STATUS_MAP.get((local_exists, remote_exists, diff), STATUS_UNKNOWN)\n\n    def status(self, path):\n        \"\"\"\n        Generic method for checking data item status.\n        \"\"\"\n        key = self._get_key(path)\n        if not key:\n            return STATUS_NEW\n\n        return self._status(key, path)\n\n    def connect(self):\n        pass\n\n    def disconnect(self):\n        pass\n\n    def __enter__(self):\n        self.connect()\n\n    def __exit__(self, type, value, tb):\n        self.disconnect()\n", "path": "dvc/cloud/base.py"}, {"content": "import re\nfrom multiprocessing.pool import ThreadPool\n\nfrom dvc.logger import Logger\nfrom dvc.exceptions import DvcException\nfrom dvc.config import Config, ConfigError\nfrom dvc.utils import map_progress\n\nfrom dvc.cloud.aws import DataCloudAWS\nfrom dvc.cloud.gcp import DataCloudGCP\nfrom dvc.cloud.ssh import DataCloudSSH\nfrom dvc.cloud.local import DataCloudLOCAL\nfrom dvc.cloud.base import DataCloudBase, CloudSettings\nfrom dvc.cloud.base import STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED\n\n\nclass DataCloud(object):\n    \"\"\" Generic class to do initial config parsing and redirect to proper DataCloud methods \"\"\"\n\n    CLOUD_MAP = {\n        'aws'   : DataCloudAWS,\n        'gcp'   : DataCloudGCP,\n        'ssh'   : DataCloudSSH,\n        'local' : DataCloudLOCAL,\n    }\n\n    def __init__(self, cache=None, config=None):\n        self._cache = cache\n        self._config = config\n\n        remote = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_REMOTE, '')\n        if remote == '':\n            if config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, None):\n                # backward compatibility\n                Logger.warn('Using obsoleted config format. Consider updating.')\n                self._cloud = self.__init__compat()\n            else:\n                self._cloud = None\n            return\n\n        self._cloud = self._init_remote(remote)\n\n    @staticmethod\n    def supported(url):\n        for cloud in DataCloud.CLOUD_MAP.values():\n            if cloud.supported(url):\n                return cloud\n        return None\n\n    def _init_remote(self, remote):\n        section = Config.SECTION_REMOTE_FMT.format(remote)\n        cloud_config = self._config.get(section, None)\n        if not cloud_config:\n            raise ConfigError(\"Can't find remote section '{}' in config\".format(section))\n\n        url = cloud_config[Config.SECTION_REMOTE_URL]\n        cloud_type = self.supported(url)\n        if not cloud_type:\n            raise ConfigError(\"Unsupported url '{}'\".format(url))\n\n        return self._init_cloud(cloud_config, cloud_type)\n\n    def __init__compat(self):\n        cloud_name = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, '').strip().lower()\n        if cloud_name == '':\n            self._cloud = None\n            return\n\n        cloud_type = self.CLOUD_MAP.get(cloud_name, None)\n        if not cloud_type:\n            raise ConfigError('Wrong cloud type %s specified' % cloud_name)\n\n        cloud_config = self._config.get(cloud_name, None)\n        if not cloud_config:\n            raise ConfigError('Can\\'t find cloud section \\'[%s]\\' in config' % cloud_name)\n\n        return self._init_cloud(cloud_config, cloud_type)\n\n    def _init_cloud(self, cloud_config, cloud_type):\n        global_storage_path = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_STORAGEPATH, None)\n        if global_storage_path:\n            Logger.warn('Using obsoleted config format. Consider updating.')\n\n        cloud_settings = CloudSettings(cache=self._cache,\n                                       global_storage_path=global_storage_path,\n                                       cloud_config=cloud_config)\n\n        cloud = cloud_type(cloud_settings)\n        cloud.sanity_check()\n        return cloud\n\n    def _collect(self, cloud, targets, jobs, local):\n        collected = set()\n        pool = ThreadPool(processes=jobs)\n        args = zip(targets, [local]*len(targets))\n        ret = pool.map(cloud.collect, args)\n\n        for r in ret:\n            collected |= set(r)\n\n        return collected\n\n    def _get_cloud(self, remote):\n        if remote:\n            return self._init_remote(remote)\n\n        return self._cloud\n\n    def _filter(self, func, status, targets, jobs, remote):\n        cloud = self._get_cloud(remote)\n        if not cloud:\n            return []\n\n        with cloud:\n            filtered = []\n            for t, s in self._status(cloud, targets, jobs):\n                if s == STATUS_MODIFIED or s == status:\n                    filtered.append(t)\n\n            return map_progress(getattr(cloud, func), filtered, jobs)\n\n    def push(self, targets, jobs=1, remote=None):\n        \"\"\"\n        Push data items in a cloud-agnostic way.\n        \"\"\"\n        return self._filter('push', STATUS_NEW, targets, jobs, remote)\n\n    def pull(self, targets, jobs=1, remote=None):\n        \"\"\"\n        Pull data items in a cloud-agnostic way.\n        \"\"\"\n        return self._filter('pull', STATUS_DELETED, targets, jobs, remote)\n\n    def _collect_targets(self, cloud, targets, jobs=1):\n        collected = set()\n        collected |= self._collect(cloud, targets, jobs, True)\n        collected |= self._collect(cloud, targets, jobs, False)\n        return list(collected)\n\n    def _status(self, cloud, targets, jobs=1):\n        collected = self._collect_targets(cloud, targets, jobs)\n        return map_progress(cloud.status, collected, jobs)\n\n    def status(self, targets, jobs=1, remote=None):\n        \"\"\"\n        Check status of data items in a cloud-agnostic way.\n        \"\"\"\n        cloud = self._get_cloud(remote)\n        if not cloud:\n            return []\n\n        with cloud:\n            return self._status(cloud, targets, jobs)\n", "path": "dvc/cloud/data_cloud.py"}]}