show versions registered in specified git ref

gto/api.py

@@ -4,13 +4,15 @@
 from funcy import distinct
 from git import Repo
 
+from gto.base import Version, filter_versions
 from gto.constants import (
     ARTIFACT,
     ASSIGNMENTS_PER_VERSION,
     COMMIT,
     STAGE,
     VERSION,
     VERSIONS_PER_STAGE,
+    Shortcut,
     VersionSort,
     is_hexsha,
     mark_artifact_unregistered,
@@ -21,6 +23,7 @@
 from gto.index import Artifact, FileIndexManager, RepoIndexManager
 from gto.registry import GitRegistry
 from gto.tag import parse_name as parse_tag_name
+from gto.utils import resolve_ref
 
 
 def _is_gto_repo(repo: Union[str, Repo]):
@@ -101,7 +104,7 @@ def describe(
 ) -> Optional[Artifact]:
     """Find enrichments for the artifact"""
     shortcut = parse_shortcut(name)
-    if shortcut.shortcut:
+    if shortcut:
         if rev:
             raise WrongArgs("Either specify revision or use naming shortcut.")
         # clones a remote repo second time, can be optimized
@@ -113,17 +116,16 @@ def describe(
                 "Ambiguous naming shortcut: multiple variants found."
             )
         rev = versions[0]["commit_hexsha"]
+        name = shortcut.name
 
     repo_path = repo.working_dir if isinstance(repo, Repo) else repo
     if not is_url_of_remote_repo(repo_path) and rev is None:
         # read artifacts.yaml without using Git
-        artifact = (
-            FileIndexManager.from_path(repo_path).get_index().state.get(shortcut.name)
-        )
+        artifact = FileIndexManager.from_path(repo_path).get_index().state.get(name)
     else:
         # read Git repo
         with RepoIndexManager.from_repo(repo) as index:
-            artifact = index.get_commit_index(rev).state.get(shortcut.name)
+            artifact = index.get_commit_index(rev).state.get(name)
     return artifact
 
 
@@ -458,46 +460,55 @@ def format_hexsha(hexsha):
         return hexsha[:7] if truncate_hexsha and is_hexsha(hexsha) else hexsha
 
     shortcut = parse_shortcut(name)
+    name = shortcut.name if shortcut else name
 
+    def get_versions(artifact):
+        versions = []
+        stages = artifact.get_vstages(
+            registered_only=registered_only,
+            assignments_per_version=assignments_per_version,
+            versions_per_stage=versions_per_stage,
+            sort=sort,
+        )
+        for v in artifact.get_versions(
+            active_only=not deprecated,
+            include_non_explicit=not registered_only,
+            include_discovered=False,
+        ):
+            v = v.dict_state()
+            v["stages"] = [
+                vstage.dict_state()
+                for vstages in stages.values()
+                for vstage in vstages
+                if vstage.version == v["version"]
+            ]
+            if artifact.is_active or deprecated:
+                versions.append(v)
+        return versions
+
+    versions = []
     with GitRegistry.from_repo(repo=repo) as reg:
         if raw:
-            return reg.find_artifact(shortcut.name).versions
+            return reg.find_artifact(name).versions
+
+        if name:
+            artifacts = [
+                reg.find_artifact(
+                    name,
+                    all_branches=all_branches,
+                    all_commits=all_commits,
+                )
+            ]
+        else:
+            artifacts = reg.get_artifacts(
+                all_branches=all_branches,
+                all_commits=all_commits,
+            ).values()
+        for artifact in artifacts:
+            versions.extend(get_versions(artifact))
 
-        artifact = reg.find_artifact(
-            shortcut.name,
-            all_branches=all_branches,
-            all_commits=all_commits,
-        )
-    stages = artifact.get_vstages(
-        registered_only=registered_only,
-        assignments_per_version=assignments_per_version,
-        versions_per_stage=versions_per_stage,
-        sort=sort,
-    )
-    versions = []
-    for v in artifact.get_versions(
-        active_only=not deprecated,
-        include_non_explicit=not registered_only,
-        include_discovered=True,
-    ):
-        v = v.dict_state()
-        v["stages"] = [
-            vstage.dict_state()
-            for vstages in stages.values()
-            for vstage in vstages
-            if vstage.version == v["version"]
-        ]
-        if artifact.is_active or deprecated:
-            versions.append(v)
-
-    if shortcut.latest:
-        versions = versions[:1]
-    elif shortcut.version:
-        versions = [v for v in versions if shortcut.version == v["version"]]
-    elif shortcut.stage:
-        versions = [
-            v for v in versions for a in v["stages"] if shortcut.stage == a["stage"]
-        ]
+    if shortcut:
+        versions = filter_versions(repo=repo, versions=versions, shortcut=shortcut)
 
     if not table:
         return versions
@@ -511,6 +522,8 @@ def format_hexsha(hexsha):
             else mark_artifact_unregistered(v["artifact"])
         )
         v["version"] = format_hexsha(v["version"])
+        if v["discovered"]:
+            v["version"] = mark_artifact_unregistered(v["version"])
         v["stage"] = ", ".join(
             distinct(  # TODO: remove? no longer necessary
                 s["stage"] for s in v["stages"]

gto/base.py

@@ -9,8 +9,10 @@
     ASSIGNMENTS_PER_VERSION,
     VERSIONS_PER_STAGE,
     Action,
+    Shortcut,
     VersionSort,
 )
+from gto.utils import resolve_ref
 from gto.versions import SemVer
 
 from .exceptions import (
@@ -376,6 +378,21 @@ def get(obj, key):
     return sorted_versions
 
 
+def filter_versions(repo: git.Repo, versions: List[Version], shortcut: Shortcut):
+    if shortcut.latest:
+        versions = versions[:1]
+    elif shortcut.version:
+        versions = [v for v in versions if shortcut.version == v["version"]]
+    elif shortcut.stage:
+        versions = [
+            v for v in versions for a in v["stages"] if shortcut.stage == a["stage"]
+        ]
+    elif shortcut.ref:
+        commit_hexsha = resolve_ref(repo, shortcut.ref).hexsha
+        versions = [v for v in versions if commit_hexsha == v["commit_hexsha"]]
+    return versions
+
+
 class Artifact(BaseObject):
     versions: List[Version]
     creations: List[Creation] = []

gto/cli.py

@@ -845,11 +845,14 @@ def show(  # pylint: disable=too-many-locals
         arg = "stage" if show_stage else arg
         arg = "ref" if show_ref else arg
         if arg:
-            if arg not in output[0][0]:
-                raise WrongArgs(f"Cannot apply --{arg}")
-            format_echo(
-                [v[arg] if isinstance(v, dict) else v for v in output[0]], "lines"
-            )
+            if output[0]:
+                if arg not in output[0][0]:
+                    raise WrongArgs(f"Cannot apply --{arg}")
+                format_echo(
+                    [v[arg] if isinstance(v, dict) else v for v in output[0]], "lines"
+                )
+            else:
+                format_echo([], "lines")
         else:
             format_echo(
                 output,

gto/constants.py

@@ -36,8 +36,9 @@ class Action(Enum):
 tag_re = re.compile(
     f"^(?P<artifact>{name})(((#(?P<stage>{name})|@(?P<version>v{semver}))(?P<cancel>!?))|@((?P<deprecated>deprecated)|(?P<created>created)))(#({counter}))?$"
 )
+git_ref = "[^\s\r\n]+"  # this can be further restricted to allowed chars only
 shortcut_re = re.compile(
-    f"^(?P<artifact>{name})(#(?P<stage>{name})|@(?P<version>latest|greatest|v{semver}))$"
+    f"^(?P<artifact>{name})?(#(?P<stage>{name})|@(?P<version>latest|greatest|v{semver})|:(?P<ref>{git_ref}))$"
 )
 git_hexsha_re = re.compile(r"^[0-9a-fA-F]{40}$")
 
@@ -59,26 +60,27 @@ def assert_name_is_valid(value):
 
 
 class Shortcut(BaseModel):
-    name: str
+    name: Optional[str] = None
     stage: Optional[str] = None
     version: Optional[str] = None
+    ref: Optional[str] = None
     latest: bool = False
-    shortcut: bool = False
 
 
-def parse_shortcut(value):
+def parse_shortcut(value) -> Optional[Shortcut]:
     match = re.search(shortcut_re, value)
-    if match:
-        value = match["artifact"]
-        if match["stage"]:
-            assert_name_is_valid(match["stage"])
-    latest = bool(match and (match["version"] in ("latest", "greatest")))
+    if not match:
+        return None
+    value = match["artifact"]
+    if match["stage"]:
+        assert_name_is_valid(match["stage"])
+    latest = match["version"] in ("latest", "greatest")
     return Shortcut(
         name=value,
-        stage=match["stage"] if match and match["stage"] else None,
-        version=match["version"] if match and match["version"] and not latest else None,
+        stage=match["stage"] if match["stage"] else None,
+        version=match["version"] if match["version"] and not latest else None,
+        ref=match["ref"] if match["ref"] else None,
         latest=latest,
-        shortcut=bool(match),
     )
 
 

tests/test_cli.py

@@ -146,6 +146,46 @@ def test_commands(showcase):
         ["-r", path, "rf#production", "--ref"],
         "[email protected]\n",
     )
+    _check_successful_cmd(
+        "show",
+        ["-r", path, "nn", "--ro", "--ref"],
+        "[email protected]\n",
+    )
+    _check_successful_cmd(
+        "show",
+        ["-r", path, "#production", "--ref"],
+        "[email protected]\n",
+    )
+    _check_successful_cmd(
+        "show",
+        ["-r", path, "#staging", "--ref"],
+        "[email protected]\n" "[email protected]\n",
+    )
+    _check_successful_cmd(
+        "show",
+        ["-r", path, ":HEAD", "--ref", "--ro"],
+        "[email protected]\n",
+    )
+    _check_successful_cmd(
+        "show",
+        ["-r", path, "nn:HEAD", "--ref", "--ro"],
+        "",
+    )
+    _check_successful_cmd(
+        "show",
+        ["-r", path, "rf:HEAD", "--ref", "--ro"],
+        "[email protected]\n",
+    )
+    _check_successful_cmd(
+        "show",
+        ["-r", path, "@v1.2.4", "--ref"],
+        "[email protected]\n",
+    )
+    _check_successful_cmd(
+        "show",
+        ["-r", path, "--long"],
+        None,
+    )
     _check_successful_cmd("describe", ["-r", path, "artifactnotexist"], "")
     _check_successful_cmd("describe", ["-r", path, "rf#stagenotexist"], "")
     _check_successful_cmd("describe", ["-r", path, "rf"], EXPECTED_DESCRIBE_OUTPUT)

tests/test_constants.py

@@ -1,6 +1,6 @@
 import pytest
 
-from gto.constants import check_name_is_valid
+from gto.constants import Shortcut, check_name_is_valid, parse_shortcut
 
 
 @pytest.mark.parametrize(
@@ -39,3 +39,24 @@ def test_check_name_is_valid(name):
 )
 def test_check_name_is_invalid(name):
     assert not check_name_is_valid(name)
+
+
+@pytest.mark.parametrize(
+    "name,shortcut",
+    [
+        ("", None),
+        ("m", None),
+        ("m/", None),
+        ("nn", None),
+        ("[email protected]", Shortcut(name="model", version="v1.2.3")),
+        ("model#prod", Shortcut(name="model", stage="prod")),
+        ("model:HEAD", Shortcut(name="model", ref="HEAD")),
+        (":HEAD", Shortcut(name=None, ref="HEAD")),
+        ("@v1.2.3", Shortcut(name=None, version="v1.2.3")),
+        ("#prod", Shortcut(name=None, stage="prod")),
+        # ("model:HEAD#prod", Shortcut(name="model", ref="HEAD", stage="prod")),
+        # ("model#prod:HEAD", Shortcut(name="model", ref="HEAD", stage="prod")),
+    ],
+)
+def test_parse_shortcut(name, shortcut):
+    assert parse_shortcut(name) == shortcut