artifacts: add dvc artifacts get

dvc/api/__init__.py

@@ -1,5 +1,6 @@
 from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
+from . import artifacts
 from .data import open  # pylint: disable=redefined-builtin
 from .data import get_url, read
 from .experiments import exp_save, exp_show
@@ -10,6 +11,7 @@
     "all_branches",
     "all_commits",
     "all_tags",
+    "artifacts",
     "exp_save",
     "exp_show",
     "get_url",

dvc/cli/parser.py

@@ -7,6 +7,7 @@
 from dvc import __version__
 from dvc.commands import (
     add,
+    artifacts,
     cache,
     check_ignore,
     checkout,
@@ -89,6 +90,7 @@
     experiments,
     check_ignore,
     data,
+    artifacts,
 ]
 
 

dvc/commands/artifacts.py

@@ -0,0 +1,110 @@
+import argparse
+import logging
+
+from dvc.cli import completion
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import append_doc_link, fix_subparsers
+from dvc.exceptions import DvcException
+
+logger = logging.getLogger(__name__)
+
+
+class CmdArtifactsGet(CmdBaseNoRepo):
+    def run(self):
+        from dvc.repo.artifacts import Artifacts
+        from dvc.scm import CloneError
+
+        try:
+            Artifacts.get(
+                self.args.url,
+                name=self.args.name,
+                version=self.args.rev,
+                stage=self.args.stage,
+                force=self.args.force,
+                config=self.args.config,
+                out=self.args.out,
+            )
+            return 0
+        except CloneError:
+            logger.exception("failed to get '%s'", self.args.name)
+            return 1
+        except DvcException:
+            logger.exception(
+                "failed to get '%s' from '%s'", self.args.name, self.args.url
+            )
+            return 1
+
+
+def add_parser(subparsers, parent_parser):
+    ARTIFACTS_HELP = "DVC model registry artifact commands."
+
+    artifacts_parser = subparsers.add_parser(
+        "artifacts",
+        parents=[parent_parser],
+        description=append_doc_link(ARTIFACTS_HELP, "artifacts"),
+        help=ARTIFACTS_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    artifacts_subparsers = artifacts_parser.add_subparsers(
+        dest="cmd",
+        help="Use `dvc artifacts CMD --help` to display command-specific help.",
+    )
+    fix_subparsers(artifacts_subparsers)
+
+    ARTIFACTS_GET_HELP = "Download an artifact from a DVC project."
+    get_parser = artifacts_subparsers.add_parser(
+        "get",
+        parents=[parent_parser],
+        description=append_doc_link(ARTIFACTS_GET_HELP, "artifacts/get"),
+        help=ARTIFACTS_HELP,
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    get_parser.add_argument("url", help="Location of DVC repository to download from")
+    get_parser.add_argument(
+        "name", help="Name of artifact in the repository"
+    ).complete = completion.FILE
+    get_parser.add_argument(
+        "--rev",
+        nargs="?",
+        help="Artifact version",
+        metavar="<version>",
+    )
+    get_parser.add_argument(
+        "--stage",
+        nargs="?",
+        help="Artifact stage",
+        metavar="<stage>",
+    )
+    get_parser.add_argument(
+        "-o",
+        "--out",
+        nargs="?",
+        help="Destination path to download artifact to",
+        metavar="<path>",
+    ).complete = completion.DIR
+    get_parser.add_argument(
+        "-j",
+        "--jobs",
+        type=int,
+        help=(
+            "Number of jobs to run simultaneously. "
+            "The default value is 4 * cpu_count(). "
+        ),
+        metavar="<number>",
+    )
+    get_parser.add_argument(
+        "-f",
+        "--force",
+        action="store_true",
+        default=False,
+        help="Override local file or folder if exists.",
+    )
+    get_parser.add_argument(
+        "--config",
+        type=str,
+        help=(
+            "Path to a config file that will be merged with the config "
+            "in the target repository."
+        ),
+    )
+    get_parser.set_defaults(func=CmdArtifactsGet)

dvc/exceptions.py

@@ -1,6 +1,6 @@
 """Exceptions raised by the dvc."""
 import errno
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 from dvc.utils import format_link
 
@@ -332,3 +332,24 @@ def __init__(self, fs_paths):
 class PrettyDvcException(DvcException):
     def __pretty_exc__(self, **kwargs):
         """Print prettier exception message."""
+
+
+class ArtifactNotFoundError(DvcException):
+    """Thrown if an artifact is not found in the DVC repo.
+
+    Args:
+        name (str): artifact name.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        version: Optional[str] = None,
+        stage: Optional[str] = None,
+    ):
+        self.name = name
+        self.version = version
+        self.stage = stage
+
+        desc = f" @ {stage or version}" if (stage or version) else ""
+        super().__init__(f"Unable to find artifact '{name}{desc}'")