repo: add API and CLI for reading artifacts

dvc/api/__init__.py

@@ -1,5 +1,6 @@
 from dvc.fs.dvc import _DVCFileSystem as DVCFileSystem
 
+from .artifacts import artifacts_show
 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_show",
     "exp_save",
     "exp_show",
     "get_url",

dvc/api/artifacts.py

@@ -0,0 +1,47 @@
+from typing import Any, Dict, Optional
+
+from dvc.repo import Repo
+
+
+def artifacts_show(
+    name: str,
+    version: Optional[str] = None,
+    stage: Optional[str] = None,
+    repo: Optional[str] = None,
+) -> Dict[str, str]:
+    """
+    Return path and Git revision for an artifact in a DVC project.
+
+    The resulting path and revision can be used in conjunction with other dvc.api
+    calls to open and read the artifact.
+
+    Args:
+        name (str): name of the artifact to open.
+        version (str, optional): version of the artifact to open. Defaults to
+            the latest version.
+        stage (str, optional): name of the model registry stage.
+        repo: (str, optional): path or URL for the DVC repo.
+
+    Returns:
+        Dictionary of the form:
+            {
+                "rev": ...,
+                "path": ...,
+            }
+
+    Raises:
+        dvc.exceptions.ArtifactNotFoundError: The specified artifact was not found in
+            the repo.
+    """
+    if version and stage:
+        raise ValueError("Artifact version and stage are mutually exclusive.")
+
+    repo_kwargs: Dict[str, Any] = {
+        "subrepos": True,
+        "uninitialized": True,
+    }
+    with Repo.open(repo, **repo_kwargs) as _repo:
+        rev = _repo.artifacts.get_rev(name, version=version, stage=stage)
+        with _repo.switch(rev):
+            path = _repo.artifacts.get_path(name)
+        return {"rev": rev, "path": path}

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,133 @@
+import argparse
+import logging
+
+from dvc.cli import completion
+from dvc.cli.command import CmdBaseNoRepo
+from dvc.cli.utils import DictAction, 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
+        from dvc.ui import ui
+
+        try:
+            count, out = 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,
+                remote=self.args.remote,
+                remote_config=self.args.remote_config,
+                out=self.args.out,
+            )
+            ui.write(f"Downloaded {count} file(s) to '{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.add_argument(
+        "--remote",
+        type=str,
+        help=(
+            "Remote name to set as a default in the target repository "
+            "(only applicable when downloading from DVC remote)."
+        ),
+    )
+    get_parser.add_argument(
+        "--remote-config",
+        type=str,
+        nargs="*",
+        action=DictAction,
+        help=(
+            "Remote config options to merge with a remote's config (default or one "
+            "specified by '--remote') in the target repository (only applicable "
+            "when downloading from DVC remote)."
+        ),
+    )
+    get_parser.set_defaults(func=CmdArtifactsGet)

dvc/exceptions.py

@@ -1,6 +1,6 @@
 """Exceptions raised by the dvc."""
 import errno
-from typing import TYPE_CHECKING, Dict, List, Set
+from typing import TYPE_CHECKING, Dict, List, Optional, Set
 
 from dvc.utils import format_link
 
@@ -336,3 +336,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}'")

dvc/fs/__init__.py

@@ -49,7 +49,9 @@
 # pylint: enable=unused-import
 
 
-def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None):
+def download(
+    fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
+) -> int:
     with Callback.as_tqdm_callback(
         desc=f"Downloading {fs.path.name(fs_path)}",
         unit="files",
@@ -63,7 +65,8 @@ def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
                 if not path.endswith(fs.path.flavour.sep)
             ]
             if not from_infos:
-                return localfs.makedirs(to, exist_ok=True)
+                localfs.makedirs(to, exist_ok=True)
+                return 0
             to_infos = [
                 localfs.path.join(to, *fs.path.relparts(info, fs_path))
                 for info in from_infos
@@ -82,6 +85,7 @@ def download(fs: "FileSystem", fs_path: str, to: str, jobs: Optional[int] = None
             callback=cb,
             batch_size=jobs,
         )
+        return len(to_infos)
 
 
 def parse_external_url(url, config=None):