base_commands: improve download_fileset._find_matching_artifact for git artifacts

Author: Thomas Desveaux

nimp/artifacts.py

@@ -24,7 +24,6 @@
 
 from __future__ import annotations
 
-import copy
 import datetime
 import hashlib
 import json
@@ -66,15 +65,18 @@
 
 class Artifact(TypedDict):
     revision: str
-    sortable_revision: str
     uri: str
 
 
 def _is_http_url(string: str) -> bool:
     return re.match(r'^http[s]?:\/\/.*$', string) is not None
 
 
-def list_artifacts(artifact_pattern: str, format_arguments: Mapping[str, Any], api_context) -> list[Artifact]:
+def list_artifacts(
+    artifact_pattern: str,
+    format_arguments: Mapping[str, Any],
+    api_context_: nimp.utils.git.GitApiContext | None,
+) -> list[Artifact]:
     '''List all artifacts and their revision using the provided pattern after formatting'''
 
     artifact_pattern = artifact_pattern.format_map(
@@ -103,17 +105,12 @@ def list_artifacts(artifact_pattern: str, format_arguments: Mapping[str, Any], a
             continue
 
         group_revision = artifact_match.group('revision')
-        sortable_revision = copy.deepcopy(group_revision)
-        if api_context:
-            sortable_revision = nimp.utils.git.get_gitea_commit_timestamp(api_context, group_revision)
-        if sortable_revision is not None:
-            all_artifacts.append(
-                {
-                    'revision': group_revision,
-                    'sortable_revision': sortable_revision,
-                    'uri': file_uri,
-                }
-            )
+        all_artifacts.append(
+            {
+                'revision': group_revision,
+                'uri': file_uri,
+            }
+        )
     return all_artifacts
 
 

nimp/base_commands/download_fileset.py

@@ -22,15 +22,24 @@
 
 '''Downloads a previously uploaded fileset to the local workspace'''
 
+from __future__ import annotations
+
 import copy
+import io
 import logging
 import os
 import shutil
+import subprocess
+import tempfile
+from pathlib import Path
 from pathlib import PurePosixPath
+from typing import Iterator
 
 import nimp.artifacts
 import nimp.command
 import nimp.system
+from nimp.environment import Environment
+from nimp.utils import git
 
 
 class DownloadFileset(nimp.command.Command):
@@ -59,31 +68,35 @@ def configure_arguments(self, env, parser):
     def is_available(self, env):
         return True, ''
 
-    def run(self, env):
-        api_context = nimp.utils.git.initialize_gitea_api_context(env)
+    def run(self, env: Environment) -> bool:
+        api_context = git.initialize_gitea_api_context(env)
 
-        artifacts_source = env.artifact_repository_source
+        artifacts_source: str = env.artifact_repository_source
         if env.prefer_http:
             artifacts_http_source = getattr(env, 'artifact_http_repository_source', None)
             if artifacts_http_source:
                 artifacts_source = artifacts_http_source
             else:
                 logging.warning('prefer-http provided but no artifact_http_repository_source in configuration')
 
-        artifact_uri_pattern = artifacts_source.rstrip('/') + '/' + env.artifact_collection[env.fileset]
+        artifact_uri_pattern: str = artifacts_source.rstrip('/') + '/' + str(env.artifact_collection[env.fileset])
 
         install_directory = env.root_dir
         if env.destination:
             install_directory = str(PurePosixPath(install_directory) / env.format(env.destination))
 
         format_arguments = copy.deepcopy(vars(env))
-        format_arguments['revision'] = '*'
-        logging.info('Searching %s', artifact_uri_pattern.format(**format_arguments))
+        logging.info('Searching %s', artifact_uri_pattern.format_map({**format_arguments, 'revision': '*'}))
         all_artifacts = nimp.system.try_execute(
-            lambda: nimp.artifacts.list_artifacts(artifact_uri_pattern, format_arguments, api_context), OSError
+            lambda: nimp.artifacts.list_artifacts(artifact_uri_pattern, format_arguments, api_context),
+            OSError,
         )
         artifact_to_download = DownloadFileset._find_matching_artifact(
-            all_artifacts, env.revision, env.min_revision, env.max_revision, api_context
+            all_artifacts,
+            env.revision,
+            env.min_revision,
+            env.max_revision,
+            api_context,
         )
 
         logging.info('Downloading %s%s', artifact_to_download['uri'], ' (simulation)' if env.dry_run else '')
@@ -123,39 +136,140 @@ def run(self, env):
 
         return True
 
-    # TODO: Handle revision comparison when identified by a hash
     @staticmethod
-    def _find_matching_artifact(all_artifacts, exact_revision, minimum_revision, maximum_revision, api_context):
-        all_artifacts = sorted(all_artifacts, key=lambda artifact: int(artifact['sortable_revision'], 16), reverse=True)
-        has_revision_input = exact_revision or minimum_revision or maximum_revision
-
-        if api_context:
-            exact_revision = nimp.utils.git.get_gitea_commit_timestamp(api_context, exact_revision)
-            minimum_revision = nimp.utils.git.get_gitea_commit_timestamp(api_context, minimum_revision)
-            maximum_revision = nimp.utils.git.get_gitea_commit_timestamp(api_context, maximum_revision)
-            revision_not_found = not exact_revision and not minimum_revision and not maximum_revision
-            if has_revision_input and revision_not_found:
-                raise ValueError('Searched commit not found on gitea repo')
-
-        if not api_context and (has_revision_input is not None and not has_revision_input.isdigit()):
-            raise ValueError(
-                'Revision seems to be a git commit hash but missing gitea api information. Please check project_branches in project configuration.'
-            )
+    def _find_matching_artifact(
+        all_artifacts: list[nimp.artifacts.Artifact],
+        exact_revision: str | None,
+        minimum_revision: str | None,
+        maximum_revision: str | None,
+        api_context: git.GitApiContext | None,
+    ) -> nimp.artifacts.Artifact:
+        # fastpath for exact_revision
+        if exact_revision is not None:
+            if (artifact := next((a for a in all_artifacts if a['revision'] == exact_revision), None)) is not None:
+                return artifact
+            raise ValueError('Matching artifact not found')
+
+        # fastpath for maximum_revision
+        if maximum_revision is not None:
+            if (artifact := next((a for a in all_artifacts if a['revision'] == maximum_revision), None)) is not None:
+                return artifact
+
+        if (
+            any(git.maybe_git_revision(a['revision']) for a in all_artifacts)
+            or (minimum_revision is not None and git.maybe_git_revision(minimum_revision))
+            or (maximum_revision is not None and git.maybe_git_revision(maximum_revision))
+        ):
+            if (
+                newest_rev := DownloadFileset._get_newest_revision(
+                    revisions=[a['revision'] for a in all_artifacts],
+                    minimum_revision=minimum_revision,
+                    maximum_revision=maximum_revision,
+                    api_context=api_context,
+                )
+            ) is not None:
+                return next(a for a in all_artifacts if a['revision'] == newest_rev)
+
+        probably_p4_rev = all(a['revision'].isdigit() for a in all_artifacts)
+        if probably_p4_rev:
+            iter_: Iterator[int] = iter(int(a['revision']) for a in all_artifacts)
+            if minimum_revision:
+                minimum_revision_int = int(minimum_revision)
+                iter_ = filter(lambda rev: rev >= minimum_revision_int, iter_)
+
+            if maximum_revision:
+                maximum_revision_int = int(maximum_revision)
+                iter_ = filter(lambda rev: rev <= maximum_revision_int, iter_)
+
+            if (revision := max(iter_, default=None)) is not None:
+                revision_str = str(revision)
+                return next(a for a in all_artifacts if a['revision'] == revision_str)
+
+        raise ValueError('Matching artifact not found')
+
+    @staticmethod
+    def _get_newest_revision(
+        revisions: list[str],
+        minimum_revision: str | None,
+        maximum_revision: str | None,
+        api_context: git.GitApiContext | None,
+    ) -> str | None:
+        remote: str | None = None
+        if api_context is not None and (api_client := api_context['instance'].api_client) is not None:
+            remote = f"{api_client.configuration.host}/{api_context['repo_owner']}/{api_context['repo_name']}"
+
+        cwd_git_dir = git.get_git_dir()
+
+        if remote is not None:
+            with tempfile.TemporaryDirectory(prefix="nimp_git_") as tmp_git_dir:
+                Path(tmp_git_dir).mkdir(parents=True, exist_ok=True)
+                subprocess.check_call(['git', 'init', '--bare'], cwd=tmp_git_dir)
 
-        try:
-            if exact_revision is not None:
-                return next(a for a in all_artifacts if a['sortable_revision'] == exact_revision)
-            if minimum_revision is not None and maximum_revision is not None:
-                return next(
-                    a
-                    for a in all_artifacts
-                    if int(a['sortable_revision']) >= int(minimum_revision)
-                    and int(a['sortable_revision']) <= int(maximum_revision)
+                # if current workdir contains a git repo, use it as alternate to prevent unnecessary burden on remote
+                if cwd_git_dir is not None and git.is_shallow_repository(cwd_git_dir) is False:
+                    git.add_alternates(cwd_git_dir, cwd=tmp_git_dir)
+
+                return DownloadFileset._find_newest_revision(
+                    tmp_git_dir,
+                    revisions=revisions,
+                    minimum_revision=minimum_revision,
+                    maximum_revision=maximum_revision,
                 )
-            if minimum_revision is not None:
-                return next(a for a in all_artifacts if int(a['sortable_revision']) >= int(minimum_revision))
+
+        elif cwd_git_dir is not None:
+            # no remote, fallback to current git
+            return DownloadFileset._find_newest_revision(
+                cwd_git_dir,
+                revisions=revisions,
+                minimum_revision=minimum_revision,
+                maximum_revision=maximum_revision,
+            )
+
+        # no current git. Can't find revisions informations
+        return None
+
+    @staticmethod
+    def _find_newest_revision(
+        git_dir: str, revisions: list[str], minimum_revision: str | None, maximum_revision: str | None
+    ):
+        remotes = git.get_remotes(git_dir)
+
+        to_fetch = [*revisions]
+        if minimum_revision is not None:
+            to_fetch.append(minimum_revision)
+        if maximum_revision is not None:
+            to_fetch.append(maximum_revision)
+
+        for remote in remotes:
+            if subprocess.call(['git', 'fetch', '--no-recurse-submodules', remote, *to_fetch], cwd=git_dir) != 0:
+                # might have failed due to one (or more) unknown ref,
+                # try one-by-one and ignore failures
+                for rev in to_fetch:
+                    subprocess.call(['git', 'fetch', '--no-recurse-submodules', remote, rev], cwd=git_dir)
+
+        rev_list_cmd = ['git', 'rev-list', '--ignore-missing', *revisions]
+        if maximum_revision is not None:
+            rev_list_cmd.append(maximum_revision)
+
+        if minimum_revision is not None:
+            rev_list_cmd.append(minimum_revision)
+
+        process = subprocess.run(rev_list_cmd, text=True, stdout=subprocess.PIPE, cwd=git_dir)
+        assert process.returncode == 0
+        # read line by line, no need to split all
+        with io.StringIO(process.stdout) as buffer:
+            revisions_set = set(revisions)
             if maximum_revision is not None:
-                return next(a for a in all_artifacts if int(a['sortable_revision']) <= int(maximum_revision))
-            return next(a for a in all_artifacts)
-        except StopIteration:
-            raise ValueError('Matching artifact not found')
+                # should not happen, handled as a special case before
+                assert maximum_revision not in revisions_set
+                while revision := buffer.readline():
+                    if revision == maximum_revision:
+                        break
+
+            while revision := buffer.readline():
+                if revision in revisions_set:
+                    return revision
+                if revision == minimum_revision:
+                    break
+
+            return None

nimp/utils/git.py

@@ -25,9 +25,12 @@
 from __future__ import annotations
 
 import logging
+import os
+import subprocess
 import time
 from datetime import datetime
 from datetime import timezone
+from pathlib import Path
 from typing import TypedDict
 
 import giteapy
@@ -86,6 +89,10 @@ def get_commit_version(commit_hash):
     return output
 
 
+def maybe_git_revision(candidate: str) -> bool:
+    return candidate.isalnum() and candidate.islower()
+
+
 def is_full_sha1(string):
     '''cheap logic to check if we have full git commit sha1 string'''
     if len(string) != 40:
@@ -144,3 +151,60 @@ def get_gitea_commit_timestamp(gitea_context: GitApiContext, commit_sha):
         reason = str(e.reason).lower() if hasattr(e, 'reason') else ''
         logging.debug(f'[GITEA API] {gitea_context["repo_owner"]}@{gitea_context["repo_name"]}@{commit_sha} {reason}')
     return api_commit_timestamp
+
+
+def get_git_dir(cwd: os.PathLike[str] | str | None = None) -> str | None:
+    process = subprocess.run(
+        ["git", "rev-parse", "--git-dir"],
+        text=True,
+        cwd=cwd,
+        stdout=subprocess.PIPE,
+    )
+    if process.returncode == 0:
+        return process.stdout.strip()
+
+    return None
+
+
+def is_shallow_repository(cwd: os.PathLike[str] | str | None = None) -> bool | None:
+    process = subprocess.run(
+        ["git", "rev-parse", "--is-shallow-repository"],
+        text=True,
+        cwd=cwd,
+        stdout=subprocess.PIPE,
+    )
+    if process.returncode == 0:
+        return {"true": True, "false": False}.get(process.stdout.strip().lower())
+
+    return None
+
+
+def add_alternates(*alternates: str, cwd: os.PathLike[str] | str | None = None) -> None:
+    git_dir = get_git_dir(cwd)
+    if git_dir is None:
+        return
+
+    if not Path(git_dir).is_dir():
+        # git-dir might be a file pointing to the real git-dir
+        # Ignore this case for now
+        return
+
+    alternates_file = Path(git_dir, "objects/info/alternates")
+    alternates_file.parent.mkdir(parents=True, exist_ok=True)
+
+    current_alternates = alternates_file.read_text().splitlines(keepends=False)
+
+    new_alternates = set(alternates).difference(current_alternates)
+
+    current_alternates.extend(new_alternates)
+
+    alternates_file.write_text('\n'.join(current_alternates))
+
+
+def get_remotes(cwd: os.PathLike[str] | str) -> list[str]:
+    return subprocess.run(
+        ['git', 'remote'],
+        text=True,
+        cwd=cwd,
+        stdout=subprocess.PIPE,
+    ).stdout.splitlines(keepends=False)