Subcommand help

dvc/cli.py

@@ -81,18 +81,33 @@
 class DvcParser(argparse.ArgumentParser):
     """Custom parser class for dvc CLI."""
 
-    def error(self, message):
+    def error(self, message, command=None):
         """Custom error method.
         Args:
             message (str): error message.
-
+            command (str): subcommand name for help message
         Raises:
             dvc.exceptions.DvcParser: dvc parser exception.
+
         """
         logger.error(message)
+        if command is not None:
+            for action in self._actions:
+                if action.dest == "cmd" and command in action.choices:
+                    subparser = action.choices[command]
+                    subparser.print_help()
+                    raise DvcParserError()
         self.print_help()
         raise DvcParserError()
 
+    # override this to send subcommand name to error method
+    def parse_args(self, args=None, namespace=None):
+        args, argv = self.parse_known_args(args, namespace)
+        if argv:
+            msg = "unrecognized arguments: %s"
+            self.error(msg % " ".join(argv), args.cmd)
+        return args
+
 
 class VersionAction(argparse.Action):  # pragma: no cover
     # pylint: disable=too-few-public-methods

dvc/command/run.py

@@ -133,16 +133,15 @@ def add_parser(subparsers, parent_parser):
         "--metrics",
         action="append",
         default=[],
-        help="Declare output metric file or directory.",
+        help="Declare output metric file.",
         metavar="<path>",
     )
     run_parser.add_argument(
         "-M",
         "--metrics-no-cache",
         action="append",
         default=[],
-        help="Declare output metric file or directory "
-        "(do not put into DVC cache).",
+        help="Declare output metric file (do not put into DVC cache).",
         metavar="<path>",
     )
     run_parser.add_argument(

dvc/dependency/__init__.py

@@ -87,26 +87,29 @@ def loads_from(stage, s_list, erepo=None):
     return ret
 
 
-def _parse_params(path_params):
-    path, _, params_str = path_params.rpartition(":")
-    params = params_str.split(",")
-    return path, params
+def _merge_params(s_list):
+    d = defaultdict(list)
+    default_file = ParamsDependency.DEFAULT_PARAMS_FILE
+    for key in s_list:
+        if isinstance(key, str):
+            d[default_file].append(key)
+            continue
+        if not isinstance(key, dict):
+            msg = "Only list of str/dict is supported. Got: "
+            msg += f"'{type(key).__name__}'."
+            raise ValueError(msg)
+
+        for k, params in key.items():
+            if not isinstance(params, list):
+                msg = "Expected list of params for custom params file "
+                msg += f"'{k}', got '{type(params).__name__}'."
+                raise ValueError(msg)
+            d[k].extend(params)
+    return d
 
 
 def loads_params(stage, s_list):
-    # Creates an object for each unique file that is referenced in the list
-    params_by_path = defaultdict(list)
-    for s in s_list:
-        path, params = _parse_params(s)
-        params_by_path[path].extend(params)
-
-    d_list = []
-    for path, params in params_by_path.items():
-        d_list.append(
-            {
-                BaseOutput.PARAM_PATH: path,
-                ParamsDependency.PARAM_PARAMS: params,
-            }
-        )
-
-    return loadd_from(stage, d_list)
+    d = _merge_params(s_list)
+    return [
+        ParamsDependency(stage, path, params) for path, params in d.items()
+    ]

dvc/dependency/param.py

@@ -40,7 +40,7 @@ def __init__(self, stage, path, params):
             info=info,
         )
 
-    def _dyn_load(self, values=None):
+    def fill_values(self, values=None):
         """Load params values dynamically."""
         if not values:
             return

dvc/output/__init__.py

@@ -1,5 +1,7 @@
+from collections import defaultdict
 from urllib.parse import urlparse
 
+from funcy import collecting, project
 from voluptuous import And, Any, Coerce, Length, Lower, Required, SetTo
 
 from dvc.output.base import BaseOutput
@@ -58,7 +60,9 @@
 SCHEMA[BaseOutput.PARAM_PERSIST] = bool
 
 
-def _get(stage, p, info, cache, metric, plot=False, persist=False):
+def _get(
+    stage, p, info=None, cache=True, metric=False, plot=False, persist=False
+):
     parsed = urlparse(p)
 
     if parsed.scheme == "remote":
@@ -135,3 +139,45 @@ def loads_from(
         )
         for s in s_list
     ]
+
+
+def _split_dict(d, keys):
+    return project(d, keys), project(d, d.keys() - keys)
+
+
+def _merge_data(s_list):
+    d = defaultdict(dict)
+    for key in s_list:
+        if isinstance(key, str):
+            d[key].update({})
+            continue
+        if not isinstance(key, dict):
+            raise ValueError(f"'{type(key).__name__}' not supported.")
+
+        for k, flags in key.items():
+            if not isinstance(flags, dict):
+                raise ValueError(
+                    f"Expected dict for '{k}', got: '{type(flags).__name__}'"
+                )
+            d[k].update(flags)
+    return d
+
+
+@collecting
+def load_from_pipeline(stage, s_list, typ="outs"):
+    if typ not in (stage.PARAM_OUTS, stage.PARAM_METRICS, stage.PARAM_PLOTS):
+        raise ValueError(f"'{typ}' key is not allowed for pipeline files.")
+
+    metric = typ == stage.PARAM_METRICS
+    plot = typ == stage.PARAM_PLOTS
+
+    d = _merge_data(s_list)
+
+    for path, flags in d.items():
+        plt_d = {}
+        if plot:
+            from dvc.schema import PLOT_PROPS
+
+            plt_d, flags = _split_dict(flags, keys=PLOT_PROPS.keys())
+        extra = project(flags, ["cache", "persist"])
+        yield _get(stage, path, {}, plot=plt_d or plot, metric=metric, **extra)

dvc/output/base.py

@@ -119,7 +119,7 @@ def _parse_path(self, remote, path):
         if remote:
             parsed = urlparse(path)
             return remote.path_info / parsed.path.lstrip("/")
-        return self.REMOTE.path_cls(path)
+        return self.REMOTE.TREE_CLS.PATH_CLS(path)
 
     def __repr__(self):
         return "{class_name}: '{def_path}'".format(
@@ -246,10 +246,11 @@ def save(self):
 
         self.ignore()
 
+        if self.metric or self.plot:
+            self.verify_metric()
+
         if not self.use_cache:
             self.info = self.save_info()
-            if self.metric or self.plot:
-                self.verify_metric()
             if not self.IS_DEPENDENCY:
                 logger.debug(
                     "Output '%s' doesn't use cache. Skipping saving.", self
@@ -300,7 +301,7 @@ def verify_metric(self):
         raise DvcException(f"verify metric is not supported for {self.scheme}")
 
     def download(self, to):
-        self.remote.download(self.path_info, to.path_info)
+        self.remote.tree.download(self.path_info, to.path_info)
 
     def checkout(
         self,

dvc/output/local.py

@@ -33,12 +33,12 @@ def _parse_path(self, remote, path):
             #
             # FIXME: if we have Windows path containing / or posix one with \
             # then we have #2059 bug and can't really handle that.
-            p = self.REMOTE.path_cls(path)
+            p = self.REMOTE.TREE_CLS.PATH_CLS(path)
             if not p.is_absolute():
                 p = self.stage.wdir / p
 
         abs_p = os.path.abspath(os.path.normpath(p))
-        return self.REMOTE.path_cls(abs_p)
+        return self.REMOTE.TREE_CLS.PATH_CLS(abs_p)
 
     def __str__(self):
         if not self.is_in_repo:
@@ -81,7 +81,7 @@ def verify_metric(self):
         name = "metrics" if self.metric else "plot"
         if os.path.isdir(path):
             msg = "directory '{}' cannot be used as {}."
-            raise DvcException(msg.format(self.path_info, name))
+            raise IsADirectoryError(msg.format(self.path_info, name))
 
         if not istextfile(path):
             msg = "binary file '{}' cannot be used as {}."

dvc/remote/azure.py

@@ -1,6 +1,5 @@
 import logging
 import os
-import posixpath
 import threading
 from datetime import datetime, timedelta
 
@@ -15,56 +14,17 @@
 
 
 class AzureRemoteTree(BaseRemoteTree):
-    @property
-    def blob_service(self):
-        return self.remote.blob_service
-
-    def _generate_download_url(self, path_info, expires=3600):
-        from azure.storage.blob import BlobPermissions
-
-        expires_at = datetime.utcnow() + timedelta(seconds=expires)
-
-        sas_token = self.blob_service.generate_blob_shared_access_signature(
-            path_info.bucket,
-            path_info.path,
-            permission=BlobPermissions.READ,
-            expiry=expires_at,
-        )
-        download_url = self.blob_service.make_blob_url(
-            path_info.bucket, path_info.path, sas_token=sas_token
-        )
-        return download_url
-
-    def exists(self, path_info):
-        paths = self.remote.list_paths(path_info.bucket, path_info.path)
-        return any(path_info.path == path for path in paths)
-
-    def remove(self, path_info):
-        if path_info.scheme != self.scheme:
-            raise NotImplementedError
-
-        logger.debug(f"Removing {path_info}")
-        self.blob_service.delete_blob(path_info.bucket, path_info.path)
+    PATH_CLS = CloudURLInfo
 
-
-class AzureRemote(BaseRemote):
-    scheme = Schemes.AZURE
-    path_cls = CloudURLInfo
-    REQUIRES = {"azure-storage-blob": "azure.storage.blob"}
-    PARAM_CHECKSUM = "etag"
-    COPY_POLL_SECONDS = 5
-    LIST_OBJECT_PAGE_SIZE = 5000
-    TREE_CLS = AzureRemoteTree
-
-    def __init__(self, repo, config):
-        super().__init__(repo, config)
+    def __init__(self, remote, config):
+        super().__init__(remote, config)
 
         url = config.get("url", "azure://")
-        self.path_info = self.path_cls(url)
+        self.path_info = self.PATH_CLS(url)
 
         if not self.path_info.bucket:
             container = os.getenv("AZURE_STORAGE_CONTAINER_NAME")
-            self.path_info = self.path_cls(f"azure://{container}")
+            self.path_info = self.PATH_CLS(f"azure://{container}")
 
         self.connection_string = config.get("connection_string") or os.getenv(
             "AZURE_STORAGE_CONNECTION_STRING"
@@ -96,10 +56,27 @@ def get_etag(self, path_info):
         ).properties.etag
         return etag.strip('"')
 
-    def get_file_checksum(self, path_info):
-        return self.get_etag(path_info)
+    def _generate_download_url(self, path_info, expires=3600):
+        from azure.storage.blob import BlobPermissions
+
+        expires_at = datetime.utcnow() + timedelta(seconds=expires)
 
-    def list_paths(self, bucket, prefix, progress_callback=None):
+        sas_token = self.blob_service.generate_blob_shared_access_signature(
+            path_info.bucket,
+            path_info.path,
+            permission=BlobPermissions.READ,
+            expiry=expires_at,
+        )
+        download_url = self.blob_service.make_blob_url(
+            path_info.bucket, path_info.path, sas_token=sas_token
+        )
+        return download_url
+
+    def exists(self, path_info):
+        paths = self._list_paths(path_info.bucket, path_info.path)
+        return any(path_info.path == path for path in paths)
+
+    def _list_paths(self, bucket, prefix):
         blob_service = self.blob_service
         next_marker = None
         while True:
@@ -108,25 +85,28 @@ def list_paths(self, bucket, prefix, progress_callback=None):
             )
 
             for blob in blobs:
-                if progress_callback:
-                    progress_callback()
                 yield blob.name
 
             if not blobs.next_marker:
                 break
 
             next_marker = blobs.next_marker
 
-    def list_cache_paths(self, prefix=None, progress_callback=None):
-        if prefix:
-            prefix = posixpath.join(
-                self.path_info.path, prefix[:2], prefix[2:]
-            )
-        else:
-            prefix = self.path_info.path
-        return self.list_paths(
-            self.path_info.bucket, prefix, progress_callback
-        )
+    def walk_files(self, path_info, **kwargs):
+        for fname in self._list_paths(
+            path_info.bucket, path_info.path, **kwargs
+        ):
+            if fname.endswith("/"):
+                continue
+
+            yield path_info.replace(path=fname)
+
+    def remove(self, path_info):
+        if path_info.scheme != self.scheme:
+            raise NotImplementedError
+
+        logger.debug(f"Removing {path_info}")
+        self.blob_service.delete_blob(path_info.bucket, path_info.path)
 
     def _upload(
         self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
@@ -149,3 +129,15 @@ def _download(
                 to_file,
                 progress_callback=pbar.update_to,
             )
+
+
+class AzureRemote(BaseRemote):
+    scheme = Schemes.AZURE
+    REQUIRES = {"azure-storage-blob": "azure.storage.blob"}
+    PARAM_CHECKSUM = "etag"
+    COPY_POLL_SECONDS = 5
+    LIST_OBJECT_PAGE_SIZE = 5000
+    TREE_CLS = AzureRemoteTree
+
+    def get_file_checksum(self, path_info):
+        return self.tree.get_etag(path_info)