Split incus specific code to incuslib

image_builder.py

@@ -1,103 +1,15 @@
 #!/usr/bin/env python3
 
 import argparse
-import os
 import logging
-import platform
-import subprocess
 from datetime import datetime
 from pathlib import Path
 from typing import Optional
-
-import yaml
+from incuslib import Incus, SimpleStreams
 
 
 SCRIPT_DIR = Path(__file__).resolve().parent
 
-
-class Incus:
-    def __init__(self) -> None:
-        pass
-
-    def arch(self) -> str:
-        plat = platform.machine()
-        if plat in ["x86_64", "amd64"]:
-            return "amd64"
-        if plat in ["arm64", "aarch64"]:
-            return "arm64"
-        if plat in ["armhf"]:
-            return "armhf"
-        raise RuntimeError(f"Unknown platform {plat}!")
-
-    def _run(self, *args: str, **kwargs) -> str:
-        command = ["incus"] + [*args]
-        return subprocess.check_output(command, **kwargs).decode("utf-8")
-
-    def _run_logged_prefixed(self, *args: str, prefix: str = "", **kwargs) -> None:
-        command = ["incus"] + [*args]
-
-        process = subprocess.Popen(
-            command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kwargs
-        )
-        assert process.stdout
-        with process.stdout:
-            for line in iter(process.stdout.readline, b""):  # b'\n'-separated lines
-                logging.debug("%s%s", prefix, line.decode("utf-8").rstrip("\n"))
-        exitcode = process.wait()  # 0 means success
-        if exitcode:
-            raise RuntimeError(f"Could not run {' '.join(command)}")
-
-    def instance_stopped(self, name: str) -> bool:
-        assert self.instance_exists(name)
-        res = yaml.safe_load(self._run("info", name))
-        return res["Status"] == "STOPPED"
-
-    def instance_exists(self, name: str) -> bool:
-        res = yaml.safe_load(self._run("list", "-f", "yaml"))
-        instance_names = [instance["name"] for instance in res]
-        return name in instance_names
-
-    def instance_start(self, name: str) -> None:
-        self._run("start", name)
-
-    def instance_stop(self, name: str) -> None:
-        self._run("stop", name)
-
-    def instance_delete(self, name: str) -> None:
-        self._run("delete", name)
-
-    def launch(self, image_name: str, instance_name: str) -> None:
-        self._run("launch", image_name, instance_name)
-
-    def push_file(self, instance_name: str, file: Path, target: str) -> None:
-        self._run("file", "push", str(file), f"{instance_name}{target}")
-        os.sync()
-
-    def execute(self, instance_name: str, *args: str) -> None:
-        self._run_logged_prefixed(
-            "exec", instance_name, "--", *args, prefix=" In container |\t"
-        )
-
-    def publish(
-        self, instance_name: str, image_alias: str, properties: dict[str, str]
-    ) -> None:
-        properties_list = [f"{key}={value}" for key, value in properties.items()]
-        self._run("publish", instance_name, "--alias", image_alias, *properties_list)
-
-    def image_export(
-        self, image_alias: str, image_target: str, target_dir: Path
-    ) -> None:
-        self._run("image", "export", image_alias, image_target, cwd=target_dir)
-
-    def image_exists(self, alias: str) -> bool:
-        res = yaml.safe_load(self._run("image", "list", "-f", "yaml"))
-        image_aliases = [alias["name"] for image in res for alias in image["aliases"]]
-        return alias in image_aliases
-
-    def image_delete(self, alias: str) -> None:
-        self._run("image", "delete", alias)
-
-
 incus = Incus()
 
 
@@ -162,17 +74,8 @@ def publish(self, short_name: str) -> None:
         incus.publish(self.instance_name, image_alias, properties)
 
         if self.ss_repo:
-            images_path = SCRIPT_DIR / "images"
-            images_path.mkdir(exist_ok=True)
-            image_alias_underscorified = image_alias.replace("/", "_")
-            image_file = images_path / f"{image_alias_underscorified}.tar.gz"
-            incus.image_export(image_alias, image_alias_underscorified, images_path)
-
-            subprocess.run(
-                ["incus-simplestreams", "add", image_file],
-                cwd=self.ss_repo,
-            )
-            image_file.unlink()
+            ss = SimpleStreams(incus, self.ss_repo, SCRIPT_DIR / "images")
+            ss.import_from_incus(image_alias, image_alias)
 
         if should_restart:
             incus.instance_start(self.instance_name)

incuslib/__init__.py

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+
+from .incus import Incus
+from .simplestreams import SimpleStreams
+
+__all__ = ["Incus", "SimpleStreams"]

incuslib/incus.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+
+import logging
+import os
+import platform
+import subprocess
+from pathlib import Path
+
+import yaml
+
+
+class Incus:
+    def __init__(self) -> None:
+        pass
+
+    def arch(self) -> str:
+        plat = platform.machine()
+        if plat in ["x86_64", "amd64"]:
+            return "amd64"
+        if plat in ["arm64", "aarch64"]:
+            return "arm64"
+        if plat in ["armhf"]:
+            return "armhf"
+        raise RuntimeError(f"Unknown platform {plat}!")
+
+    def _run(self, *args: str, **kwargs) -> str:
+        command = ["incus"] + [*args]
+        return subprocess.check_output(command, **kwargs).decode("utf-8")
+
+    def _run_logged_prefixed(self, *args: str, prefix: str = "", **kwargs) -> None:
+        command = ["incus"] + [*args]
+
+        process = subprocess.Popen(
+            command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kwargs
+        )
+        assert process.stdout
+        with process.stdout:
+            for line in iter(process.stdout.readline, b""):  # b'\n'-separated lines
+                logging.debug("%s%s", prefix, line.decode("utf-8").rstrip("\n"))
+        exitcode = process.wait()  # 0 means success
+        if exitcode:
+            raise RuntimeError(f"Could not run {' '.join(command)}")
+
+    def instance_stopped(self, name: str) -> bool:
+        assert self.instance_exists(name)
+        res = yaml.safe_load(self._run("info", name))
+        return res["Status"] == "STOPPED"
+
+    def instance_exists(self, name: str) -> bool:
+        res = yaml.safe_load(self._run("list", "-f", "yaml"))
+        instance_names = [instance["name"] for instance in res]
+        return name in instance_names
+
+    def instance_start(self, name: str) -> None:
+        self._run("start", name)
+
+    def instance_stop(self, name: str) -> None:
+        self._run("stop", name)
+
+    def instance_delete(self, name: str) -> None:
+        self._run("delete", name)
+
+    def launch(self, image_name: str, instance_name: str) -> None:
+        self._run("launch", image_name, instance_name)
+
+    def push_file(self, instance_name: str, file: Path, target: str) -> None:
+        self._run("file", "push", str(file), f"{instance_name}{target}")
+        os.sync()
+
+    def execute(self, instance_name: str, *args: str) -> None:
+        self._run_logged_prefixed(
+            "exec", instance_name, "--", *args, prefix=" In container |\t"
+        )
+
+    def publish(
+        self, instance_name: str, image_alias: str, properties: dict[str, str]
+    ) -> None:
+        properties_list = [f"{key}={value}" for key, value in properties.items()]
+        self._run("publish", instance_name, "--alias", image_alias, *properties_list)
+
+    def image_export(
+        self, image_alias: str, image_target: str, target_dir: Path
+    ) -> None:
+        self._run("image", "export", image_alias, image_target, cwd=target_dir)
+
+    def image_exists(self, alias: str) -> bool:
+        res = yaml.safe_load(self._run("image", "list", "-f", "yaml"))
+        image_aliases = [alias["name"] for image in res for alias in image["aliases"]]
+        return alias in image_aliases
+
+    def image_delete(self, alias: str) -> None:
+        self._run("image", "delete", alias)

incuslib/simplestreams.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+
+import json
+import logging
+import subprocess
+from pathlib import Path
+
+from .incus import Incus
+
+
+class SimpleStreams:
+    def __init__(self, incus: Incus, path: Path, cachedir: Path) -> None:
+        self.incus = incus
+        self.path = path
+        self.path.mkdir(exist_ok=True)
+        self.cachedir = cachedir
+        self.cachedir.mkdir(exist_ok=True)
+
+    def import_from_incus(self, name: str, alias: str) -> None:
+        image_alias_underscorified = alias.replace("/", "_")
+        image_file = self.cachedir / f"{image_alias_underscorified}.tar.gz"
+
+        self.incus.image_export(name, image_alias_underscorified, self.cachedir)
+
+        subprocess.run(
+            ["incus-simplestreams", "add", image_file],
+            cwd=self.path,
+        )
+        image_file.unlink()
+
+    def images_paths(self) -> list[str]:
+        images_data = self.images_data()
+        images = [
+            self.path / item["path"]
+            for product in images_data["products"].values()
+            for version in product["versions"].values()
+            for item in version["items"].values()
+        ]
+        return images
+
+    def images_data(self) -> dict:
+        images_file = self.path / "streams" / "v1" / "images.json"
+        images_data = json.load(images_file.open(encoding="utf-8"))
+        return images_data
+
+    def prune_images(self) -> None:
+        images_dir: Path = self.path / "images"
+
+        images = self.images_paths()
+        for file in images_dir.iterdir():
+            if file not in images:
+                logging.info(f"Pruning {file.name}...")
+                file.unlink()
+
+    def clean_previous_versions(self) -> None:
+        for product_name, product in self.images_data()["products"].items():
+            versions = sorted(product["versions"].keys())
+            for version in versions[:-1]:
+                for item in product["versions"][version]["items"].values():
+                    sha = item["sha256"]
+                    print(f"Pruning {product_name} / {sha}...")
+                    subprocess.run(
+                        ["incus-simplestreams", "remove", sha], cwd=self.path
+                    )

prune_incus_simplestreams.py

@@ -1,30 +1,11 @@
 #!/usr/bin/env python3
 
 import argparse
-import subprocess
 from pathlib import Path
-import json
 
+from incuslib import Incus, SimpleStreams
 
-def images_paths(repo_root: Path, images_data: dict) -> list[str]:
-    images = [
-        repo_root / item["path"]
-        for product in images_data["products"].values()
-        for version in product["versions"].values()
-        for item in version["items"].values()
-    ]
-    return images
-
-
-def clean_old_versions(repo_root: Path, images_data: dict) -> None:
-    for product_name, product in images_data["products"].items():
-        versions = sorted(product["versions"].keys())
-        for version in versions[:-1]:
-            for item in product["versions"][version]["items"].values():
-                sha = item["sha256"]
-                print(f"Pruning {product_name} / {sha}...")
-                subprocess.run(["incus-simplestreams", "remove", sha], cwd=repo_root)
-
+SCRIPT_DIR = Path(__file__).resolve().parent
 
 def main() -> None:
     parser = argparse.ArgumentParser()
@@ -37,18 +18,10 @@ def main() -> None:
     )
     args = parser.parse_args()
 
-    images_file = args.repository / "streams" / "v1" / "images.json"
-    images_data = json.load(images_file.open(encoding="utf-8"))
-
-    images = images_paths(args.repository, images_data)
-
-    images_dir: Path = args.repository / "images"
-    for file in images_dir.iterdir():
-        if file not in images:
-            print(f"Pruning {file.name}...")
-            file.unlink()
-
-    clean_old_versions(args.repository, images_data)
+    incus = Incus()
+    ss = SimpleStreams(incus, args.repository, SCRIPT_DIR / "images")
+    ss.clean_previous_versions()
+    ss.prune_images()
 
 
 if __name__ == "__main__":