Implement tag fetching using git ls-remote --tags

This allows us to skip over using the Github API which can hit the rate
limit. We also introduce a cache to avoid the Git server's rate limit

Author: woojiahao

app/cli.py

@@ -10,8 +10,8 @@
 from app.utils.gitmastery import (
     find_exercise_root,
     find_gitmastery_root,
-    read_gitmastery_config,
     read_exercise_config,
+    read_gitmastery_config,
 )
 from app.utils.version import Version
 from app.version import __version__
@@ -45,6 +45,10 @@ def cli(ctx, verbose) -> None:
         ctx.obj[CliContextKey.GITMASTERY_EXERCISE_CONFIG] = exercise_root_config
 
     ctx.obj[CliContextKey.VERBOSE] = verbose
+    # We make the assumption that within a single command run, the "state of the world"
+    # is immutable, allowing us to cache things
+    ctx.obj[CliContextKey.WEB_CACHE] = {}
+    ctx.obj[CliContextKey.TAG_CACHE] = []
 
     current_version = Version.parse_version_string(__version__)
     ctx.obj[CliContextKey.VERSION] = current_version

app/commands/setup_folder.py

@@ -6,7 +6,7 @@
 from app.commands.check.git import git
 from app.commands.progress.constants import PROGRESS_LOCAL_FOLDER_NAME
 from app.utils.click import error, info, invoke_command, prompt
-from app.utils.version import get_latest_release_exercise_version
+from app.utils.exercises import get_latest_release_exercise_version
 
 
 @click.command("setup")

app/utils/click.py

@@ -1,7 +1,7 @@
 import logging
 import sys
 from enum import StrEnum
-from typing import Any, NoReturn, Optional
+from typing import Any, Dict, List, NoReturn, Optional
 
 import click
 
@@ -16,6 +16,8 @@ class CliContextKey(StrEnum):
     GITMASTERY_EXERCISE_CONFIG = "GITMASTERY_EXERCISE_CONFIG"
     VERBOSE = "VERBOSE"
     VERSION = "VERSION"
+    WEB_CACHE = "WEB_CACHE"
+    TAG_CACHE = "TAG_CACHE"
 
 
 class ClickColor(StrEnum):
@@ -109,6 +111,14 @@ def get_exercise_root_config() -> Optional[ExerciseConfig]:
     )
 
 
+def get_web_cache() -> Dict[str, str | bytes | Any]:
+    return click.get_current_context().obj.get(CliContextKey.WEB_CACHE, {})
+
+
+def get_tag_cache() -> List[str]:
+    return click.get_current_context().obj.get(CliContextKey.TAG_CACHE, [])
+
+
 def invoke_command(command: click.Command) -> None:
     ctx = click.get_current_context()
     ctx.invoke(command)

app/utils/exercises.py

@@ -0,0 +1,76 @@
+from typing import List, Optional
+
+from app.utils.click import get_tag_cache, get_verbose
+from app.utils.command import run
+from app.utils.version import Version
+
+
+def get_exercises_tags() -> List[str]:
+    tag_cache = get_tag_cache()
+    if len(tag_cache) != 0:
+        if get_verbose():
+            print("Fetching tags from cache")
+        # Use the in-memory, per command cache for tags to avoid re-querying
+        return tag_cache
+
+    result = run(
+        [
+            "git",
+            "ls-remote",
+            "--tags",
+            "--refs",
+            "https://github.com/git-mastery/exercises",
+        ]
+    )
+    versions = []
+    if result.is_success() and result.stdout:
+        lines = result.stdout.split("\n")
+        for line in lines:
+            tag_raw = line.split()[1]
+            if tag_raw.startswith("refs/tags/"):
+                tag = tag_raw[len("refs/tags/") :]
+                versions.append(tag)
+    tag_cache = versions
+    if get_verbose():
+        print("Queried for tags to store in cache")
+    return versions
+
+
+def get_all_exercise_tags() -> List[Version]:
+    tags = get_exercises_tags()
+    return list(sorted([Version.parse_version_string(t) for t in tags], reverse=True))
+
+
+def get_latest_release_exercise_version() -> Optional[Version]:
+    all_tags = get_all_exercise_tags()
+    if len(all_tags) == 0:
+        # Although this should not be happening, we will let the callsite handle this
+        return None
+
+    # These should always ignore the development versions, just focus on the release
+    # versions
+    for tag in all_tags:
+        if tag.prerelease is None:
+            return tag
+
+    return None
+
+
+def get_latest_development_exercise_version() -> Optional[Version]:
+    all_tags = get_all_exercise_tags()
+    if len(all_tags) == 0:
+        # Although this should not be happening, we will let the callsite handle this
+        return None
+
+    for tag in all_tags:
+        if tag.build is None:
+            return tag
+    return None
+
+
+def get_latest_exercise_version_within_pin(pin_version: Version) -> Optional[Version]:
+    all_tags = get_all_exercise_tags()
+    for tag in all_tags:
+        if tag.within_pin(pin_version):
+            return tag
+    return None

app/utils/git.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import List, Optional
 
 from app.utils.command import run
 

app/utils/gitmastery.py

@@ -10,12 +10,20 @@
 
 from app.exercise_config import ExerciseConfig
 from app.gitmastery_config import GitMasteryConfig
-from app.utils.click import error, get_exercise_root_config, get_gitmastery_root_config
-from app.utils.general import ensure_str
-from app.utils.version import (
-    Version,
+from app.utils.click import (
+    error,
+    get_exercise_root_config,
+    get_gitmastery_root_config,
+    get_verbose,
+    get_web_cache,
+)
+from app.utils.exercises import (
+    get_latest_development_exercise_version,
     get_latest_exercise_version_within_pin,
+    get_latest_release_exercise_version,
 )
+from app.utils.general import ensure_str
+from app.utils.version import Version
 
 GITMASTERY_CONFIG_NAME = ".gitmastery.json"
 GITMASTERY_EXERCISE_CONFIG_NAME = ".gitmastery-exercise.json"
@@ -26,22 +34,26 @@
 
 def _construct_gitmastery_exercises_url(filepath: str, version: Version) -> str:
     if version.release:
-        ref = "heads/main"
+        latest_release = get_latest_release_exercise_version()
+        if latest_release is None:
+            raise ValueError("This should not happen. Contact the Git-Mastery team.")
+        ref = f"tags/v{latest_release.to_version_string()}"
     elif version.development:
-        latest_development = get_latest_exercise_version_within_pin(version)
-        ref = f"tags/{latest_development}"
+        latest_development = get_latest_development_exercise_version()
+        if latest_development is None:
+            raise ValueError("This should not happen. Contact the Git-Mastery team.")
+        ref = f"tags/v{latest_development.to_version_string()}"
     elif not version.pinned:
-        ref = f"tags/{version.to_version_string()}"
+        ref = f"tags/v{version.to_version_string()}"
     else:
         # If pinned, we need to basically search for all the available tags within the
         # range
         latest_within_pin = get_latest_exercise_version_within_pin(version)
-        ref = f"tags/{latest_within_pin}"
+        ref = f"tags/v{latest_within_pin}"
 
     url = (
         f"https://raw.githubusercontent.com/git-mastery/exercises/refs/{ref}/{filepath}"
     )
-    print(url)
     return url
 
 
@@ -156,12 +168,22 @@ def get_gitmastery_file_path(path: str):
 
 
 def fetch_file_contents(url: str, is_binary: bool) -> str | bytes:
+    web_cache = get_web_cache()
+    if url in web_cache:
+        if get_verbose():
+            print(f"Fetching {url} contents from WEB_CACHE")
+        return web_cache[url]
     response = requests.get(url)
 
+    if get_verbose():
+        print(f"Querying {url} for contents")
+
     if response.status_code == 200:
         if is_binary:
-            return response.content
-        return response.text
+            web_cache[url] = response.content
+        else:
+            web_cache[url] = response.text
+        return web_cache[url]
     else:
         error(
             f"Failed to fetch resource {click.style(url, bold=True, italic=True)}. Inform the Git-Mastery team."
@@ -171,12 +193,22 @@ def fetch_file_contents(url: str, is_binary: bool) -> str | bytes:
 def fetch_file_contents_or_none(
     url: str, is_binary: bool
 ) -> Optional[Union[str, bytes]]:
+    web_cache = get_web_cache()
+    if url in web_cache:
+        if get_verbose():
+            print(f"Fetching {url} contents from WEB_CACHE")
+        return web_cache[url]
     response = requests.get(url)
 
+    if get_verbose():
+        print(f"Querying {url} for contents")
+
     if response.status_code == 200:
         if is_binary:
-            return response.content
-        return response.text
+            web_cache[url] = response.content
+        else:
+            web_cache[url] = response.text
+        return web_cache[url]
     return None
 
 
@@ -208,10 +240,14 @@ def get_variable_from_url(
 
 def exercise_exists(exercise: str, timeout: int = 5) -> bool:
     try:
+        exercise_url = get_gitmastery_file_path(
+            f"{exercise.replace('-', '_')}/.gitmastery-exercise.json"
+        )
+        if get_verbose():
+            print(exercise_url)
+
         response = requests.head(
-            get_gitmastery_file_path(
-                f"{exercise.replace('-', '_')}/.gitmastery-exercise.json"
-            ),
+            exercise_url,
             allow_redirects=True,
             timeout=timeout,
         )

app/utils/version.py

@@ -1,8 +1,7 @@
 from dataclasses import dataclass
-from typing import ClassVar, List, Optional, Self, Type
+from typing import ClassVar, Optional, Self, Type
 
-import requests
-import semver  # type: ignore
+import semver
 
 
 @dataclass
@@ -160,49 +159,3 @@ def __repr__(self) -> str:
     release=False,
     development=True,
 )
-
-
-def get_all_exercise_tags() -> List[Version]:
-    url = "https://api.github.com/repos/git-mastery/exercises/tags"
-    tags = requests.get(url, timeout=10).json()
-    return list(
-        sorted(
-            [Version.parse_version_string(t["name"].lstrip("v")) for t in tags],
-            reverse=True,
-        )
-    )
-
-
-def get_latest_release_exercise_version() -> Optional[Version]:
-    all_tags = get_all_exercise_tags()
-    if len(all_tags) == 0:
-        # Although this should not be happening, we will let the callsite handle this
-        return None
-
-    # These should always ignore the development versions, just focus on the release
-    # versions
-    for tag in all_tags:
-        if tag.prerelease is None:
-            return tag
-
-    return None
-
-
-def get_latest_development_exercise_version() -> Optional[Version]:
-    all_tags = get_all_exercise_tags()
-    if len(all_tags) == 0:
-        # Although this should not be happening, we will let the callsite handle this
-        return None
-
-    for tag in all_tags:
-        if tag.build is None:
-            return tag
-    return None
-
-
-def get_latest_exercise_version_within_pin(pin_version: Version) -> Optional[Version]:
-    all_tags = get_all_exercise_tags()
-    for tag in all_tags:
-        if tag.within_pin(pin_version):
-            return tag
-    return None

tests/utils/test_exercises.py

@@ -0,0 +1,56 @@
+from typing import List
+from unittest import mock
+
+import pytest
+from app.utils.exercises import (
+    get_latest_exercise_version_within_pin,
+    get_latest_release_exercise_version,
+)
+from app.utils.version import Version
+
+
+def test_get_latest_release_exercise_version():
+    exercises = [
+        "0.0.1-beta.0",
+        "0.9.9",
+        "v1.0.0",
+        "1.1.1",
+        "12.1.1-beta.1",
+        "12.1.1-beta.4",
+    ]
+    with mock.patch(
+        "app.utils.version.get_all_exercise_tags", return_value=_get_versions(exercises)
+    ):
+        assert get_latest_release_exercise_version() == Version(
+            0, 9, 9, None, None, False, False, False
+        )
+
+
+@pytest.mark.parametrize(
+    "pin,expected",
+    [
+        ("^1.1.0", Version(1, 1, 1, None, None, False, False, False)),
+        ("^12.0.0", None),
+    ],
+)
+def test_get_latest_exercise_version_within_pin(pin, expected):
+    exercises = [
+        "0.0.1-beta.0",
+        "0.9.9",
+        "v1.0.0",
+        "1.1.1",
+        "12.1.1-beta.1",
+        "12.1.1-beta.4",
+    ]
+    with mock.patch(
+        "app.utils.version.get_all_exercise_tags",
+        return_value=_get_versions(exercises),
+    ):
+        version = get_latest_exercise_version_within_pin(
+            Version.parse_version_string(pin)
+        )
+        assert version == expected
+
+
+def _get_versions(version_strings: List[str]) -> List[Version]:
+    return [Version.parse_version_string(v) for v in version_strings]