Add clear_cache methods and post extraction calls.

Author: nealkruis

cachebin/cachebin.py

@@ -2,6 +2,7 @@
 import zipfile
 from pathlib import Path
 from platform import machine, system
+from shutil import rmtree
 from stat import S_IXGRP, S_IXOTH, S_IXUSR
 from subprocess import PIPE, Popen
 from typing import Callable
@@ -28,15 +29,14 @@ def download_file(url: str, directory_path: Path | str, force: bool = False) ->
         Path of the downloaded file.
     """
     directory_path = Path(directory_path)  # Ensure directory_path is a Path object
-    response = requests.get(url, stream=True)
-    response.raise_for_status()  # Raise an error for bad responses
-
     directory_path.mkdir(parents=True, exist_ok=True)
-
     filename = url.split("/")[-1]  # Extract the filename from the URL
     file_path = directory_path / filename
 
     if not file_path.exists() or force:
+        response = requests.get(url, stream=True)
+        response.raise_for_status()  # Raise an error for bad responses
+
         print(f"Downloading {url} to {file_path}...")
         with open(file_path, "wb") as file:
             for chunk in response.iter_content(chunk_size=8192):
@@ -45,7 +45,7 @@ def download_file(url: str, directory_path: Path | str, force: bool = False) ->
     return file_path
 
 
-def extract_archive(archive_path: Path | str, extract_path: Path | str) -> Path:  # noqa: PLR0912
+def extract_archive(archive_path: Path | str, extract_path: Path | str) -> tuple[Path, bool]:  # noqa: PLR0912
     """
     Extracts a compressed archive to the specified directory.
 
@@ -94,10 +94,15 @@ def extract_archive(archive_path: Path | str, extract_path: Path | str) -> Path:
         if top_item.is_directory:
             extracted_parent_directory = extract_path / top_item.filename
 
+    extracted = False
     if not extracted_parent_directory.exists() or not any(extracted_parent_directory.iterdir()):
         print(f"Extracting {archive_path} to {extract_path}...")
-        archive.extractall(path=extract_path)
-    return extracted_parent_directory
+        if isinstance(archive, tarfile.TarFile):
+            archive.extractall(path=extract_path, filter="data")
+        else:
+            archive.extractall(path=extract_path)
+        extracted = True
+    return extracted_parent_directory, extracted
 
 
 def make_executable(file_path: str | Path) -> None:
@@ -107,6 +112,12 @@ def make_executable(file_path: str | Path) -> None:
     file_path.chmod(current_permissions | S_IXUSR | S_IXGRP | S_IXOTH)
 
 
+def remove_directory(directory_path: str | Path) -> None:
+    directory_path = Path(directory_path)
+    if directory_path.exists():
+        rmtree(directory_path)
+
+
 class BinaryVersion:
     def __init__(self, version: str, parent: "BinaryManager"):
         self.parent = parent
@@ -118,23 +129,27 @@ def __init__(self, version: str, parent: "BinaryManager"):
             package_name=self.parent.package_name,
         )
         self.archive_name = self.url.split("/")[-1]
-        self.archive_path = download_file(self.url, self.parent._downloads_directory)
-        self.binary_directory_path = (
-            extract_archive(self.archive_path, self.parent._package_directory / self.version)
-            / self.parent._extracted_bin_path
-        )
-
-    def call(self, command: str, *args: str) -> str:
+        self.archive_path = download_file(self.url, self.parent._archive_directory)
+        self.extraction_directory_path = self.parent._package_directory / self.version
+        extracted_path, extracted = extract_archive(self.archive_path, self.extraction_directory_path)
+        self.binary_directory_path = extracted_path / self.parent._extracted_bin_path
+        if not self.binary_directory_path.exists():
+            raise FileNotFoundError(f"Binary directory path '{self.binary_directory_path}' does not exist.")
+        if extracted:
+            for call in self.parent._post_extraction_calls:
+                command, args = call
+                print(self.call(command, args, self.binary_directory_path))
+
+    def get_binary_path(self, command: str | None = None) -> Path:
         """
-        Calls the binary with the specified command and arguments.
+        Returns the path to the binary for the specified command.
 
         Args:
-            command (str): The command to execute.
-            *args (str): Additional arguments for the command.
+            command (str): The command to get the binary path for.
 
-        Returns:
-            str: The output of the command.
         """
+        if command is None:
+            command = self.parent.package_name
         if self.parent._system == "windows":
             binary_path_exe = self.binary_directory_path / f"{command}.exe"
             binary_path_bat = self.binary_directory_path / f"{command}.bat"
@@ -144,33 +159,74 @@ def call(self, command: str, *args: str) -> str:
                 command = f"{command}.bat"
         binary_path = self.binary_directory_path / command
         if not binary_path.exists():
-            raise FileNotFoundError(f"Binary {binary_path} does not exist.")
+            raise FileNotFoundError(f"Binary '{binary_path}' does not exist.")
+        return binary_path
+
+    def call(
+        self, command: str | None = None, arguments: list[str] | None = None, working_directory: Path | str = Path(".")
+    ) -> str:
+        """
+        Calls the binary with the specified command and arguments.
+
+        Args:
+            command (str): The command to execute.
+            *args (str): Additional arguments for the command.
+
+        Returns:
+            str: The output of the command.
+        """
+        working_directory = Path(working_directory)
+        if command is None:
+            command = self.parent.package_name
+        binary_path = self.get_binary_path(command)
 
         make_executable(binary_path)
 
         creation_flag = (
             0x08000000 if self.parent._system == "windows" else 0
         )  # set creation flag to not open in new console on windows
-        process = Popen([str(binary_path), *args], stdout=PIPE, stderr=PIPE, creationflags=creation_flag)
+
+        if arguments is None:
+            arguments = []
+        process = Popen(
+            [str(binary_path)] + arguments, stdout=PIPE, stderr=PIPE, creationflags=creation_flag, cwd=working_directory
+        )
         stdout, stderr = process.communicate()
         if process.returncode != 0:
-            raise RuntimeError(f"Command failed with error: {stderr.decode('utf-8')}")
+            raise RuntimeError(
+                f"Command '{binary_path} {' '.join(arguments)}' failed with error:\n"
+                f"{stderr.decode('utf-8')}\n{stdout.decode('utf-8')}"
+            )
         return stdout.decode("utf-8")
 
-    # TODO: def clear_cache(self) -> None:
+    def clear_cache(self) -> None:
+        """
+        Clears the cache for the version.
+        """
+        remove_directory(self.extraction_directory_path)
+        self.archive_path.unlink(missing_ok=True)
+
+
+def default_platform_string(system: str, architecture: str) -> str:
+    return f"{system}-{architecture}"
+
+
+def default_extracted_bin_path(system: str, architecture: str) -> str:
+    return "bin"
 
 
 class BinaryManager:
-    def __init__(
+    def __init__(  # noqa: PLR0913
         self,
         package_name: str,
         url_pattern: str,
         get_archive_extension: Callable[[str], str],  # returns archive extension based on system
-        get_platform_string: Callable[[str, str], str] = lambda system,
-        architecture: f"{system}-{architecture}",  # returns platform string used in url_pattern
-        get_extracted_bin_path: Callable[[str, str], str] = lambda system,
-        architecture: "bin",  # returns extracted bin path
+        get_platform_string: Callable[
+            [str, str], str
+        ] = default_platform_string,  # returns platform string used in url_pattern
+        get_extracted_bin_path: Callable[[str, str], str] = default_extracted_bin_path,  # returns extracted bin path
         cache_directory: Path | str | None = None,
+        post_extraction_calls: list[tuple[str, list[str]]] | None = None,
     ):
         self.package_name = package_name
         self.url_pattern = url_pattern
@@ -179,6 +235,10 @@ def __init__(
         self._platform_string = get_platform_string(self._system, self._architecture)
         self._extension = get_archive_extension(self._system)
         self._extracted_bin_path = get_extracted_bin_path(self._system, self._architecture)
+        if post_extraction_calls is None:
+            self._post_extraction_calls = []
+        else:
+            self._post_extraction_calls = post_extraction_calls
 
         self._cache_directory: Path
         if cache_directory is None:
@@ -189,7 +249,7 @@ def __init__(
         else:
             self._cache_directory = Path(cache_directory)
         self._downloads_directory = self._cache_directory / "downloads"
-        self._archive_directory = self._cache_directory / self.package_name
+        self._archive_directory = self._downloads_directory / self.package_name
         self._packages_directory = self._cache_directory / "packages"
         self._package_directory = self._packages_directory / self.package_name
         self._versions: dict[str, BinaryVersion] = {}
@@ -208,4 +268,19 @@ def get_version(self, version: str) -> BinaryVersion:
             self._versions[version] = BinaryVersion(version, self)
         return self._versions[version]
 
-    # TODO: def clear_cache(self) -> None:
+    def add_post_extraction_call(self, command: str, args: list[str]) -> None:
+        """
+        Adds a post-extraction call to the manager.
+
+        Args:
+            command (str): The command to execute.
+            args (list[str]): Additional arguments for the command.
+        """
+        self._post_extraction_calls.append((command, args))
+
+    def clear_cache(self) -> None:
+        """
+        Clears the cache for the package manager.
+        """
+        remove_directory(self._archive_directory)
+        remove_directory(self._package_directory)

cachebin/recipies.py

@@ -37,6 +37,7 @@ def process_map(map: dict[str, str], key_description: str, key: str) -> str:
     else f"bin/{system}"
     if system == "windows"
     else f"bin/{architecture}-{system}",
+    post_extraction_calls=[("tlmgr", ["update", "--self"])],
 )
 
 pandoc_crossref_manager = BinaryManager(

test/test_cachebin.py

@@ -1,22 +1,33 @@
 def test_pandoc():
     from cachebin.recipies import pandoc_manager
 
+    pandoc_manager.clear_cache()
+
     version = "3.6.4"
     pandoc = pandoc_manager.get_version(version)
-    assert version in pandoc.call("pandoc", "--version")
+    assert version in pandoc.call("pandoc", ["--version"])
+    pandoc.clear_cache()
 
 
 def test_tinytex():
     from cachebin.recipies import tinytex_manager
 
+    tinytex_manager.clear_cache()
+
     version = "v2025.05"
     tinytex = tinytex_manager.get_version(version)
-    assert version in tinytex.call("tlmgr", "--version")
+    assert version in tinytex.call("tlmgr", ["--version"])
+    tlmgr_list = tinytex.call("tlmgr", ["info", "--only-installed"]).splitlines()
+    tlmgr_list = [line[2 : line.find(": ")] for line in tlmgr_list]
+    tinytex.clear_cache()
 
 
 def test_pandoc_crossref():
     from cachebin.recipies import pandoc_crossref_manager
 
+    pandoc_crossref_manager.clear_cache()
+
     version = "v0.3.18.2"
     pandoc_crossref = pandoc_crossref_manager.get_version(version)
-    assert version in pandoc_crossref.call("pandoc-crossref", "--version")
+    assert version in pandoc_crossref.call("pandoc-crossref", ["--version"])
+    pandoc_crossref.clear_cache()