Add commands for evaluating HelmReleases

.pre-commit-config.yaml

@@ -19,28 +19,27 @@ repos:
     - -c
     - ".yaml-lint.yaml"
 - repo: https://github.com/pycqa/isort
-  rev: 5.9.3
+  rev: 5.12.0
   hooks:
   - id: isort
     args: ["--profile", "black"]
 - repo: https://github.com/psf/black
   rev: 22.3.0
   hooks:
   - id: black
-- repo: https://gitlab.com/pycqa/flake8
-  rev: 4.0.1
+- repo: https://github.com/PyCQA/flake8
+  rev: 6.0.0
   hooks:
   - id: flake8
-- repo: https://github.com/pre-commit/mirrors-mypy
-  rev: v0.991
+- repo: local
   hooks:
   - id: mypy
+    name: mypy
+    entry: script/run-in-env.sh mypy
+    language: script
+    types: [python]
+    require_serial: true
     files: ^(flux_local/|tests/)
-    additional_dependencies:
-    - "pydantic>=1.10.4"
-    - "types-PyYAML>=6.0.12.4"
-- repo: local
-  hooks:
   - id: pylint
     name: pylint
     entry: pylint

flux_local/helm.py

@@ -1,9 +1,132 @@
-"""Library for running `helm template` to produce local items in the cluster."""
+"""Library for running `helm template` to produce local items in the cluster.
 
-import subprocess
+You can instantiate a helm template with the following:
+- A HelmRepository which is a url that contains charts
+- A HelmRelease which is an instance of a HelmChart in a HelmRepository
 
+This is an example that prepares the helm repository:
+```
+from flux_local.kustomize import Kustomize
+from flux_local.helm import Helm
+from flux_local.manifest import HelmRepository
 
-def render(chart, values):
-    """Runs `helm template` to produce local items in the cluster."""
-    cmd = ["helm", "template", chart, "--values", values]
-    return subprocess.check_output(cmd)
+kustomize = Kustomize.build(TESTDATA_DIR)
+repos = await kustomize.grep("kind=^HelmRepository$").objects()
+helm = Helm("/tmp/path/helm", "/tmp/path/cache")
+for repo in repos:
+  helm.add_repo(HelmRepository.from_doc(repo))
+await helm.update()
+```
+
+Then to actually instantiate a template from a HelmRelease:
+```
+from flux_local.manifest import HelmRelease
+
+releases = await kustomize.grep("kind=^HelmRelease$").objects()
+if not len(releases) == 1:
+    raise ValueError("Expected only one HelmRelease")
+tmpl = helm.template(
+    HelmRelease.from_doc(releases[0]),
+    releases[0]["spec"].get("values"))
+objects = await tmpl.objects()
+for object in objects:
+    print(f"Found object {object['apiVersion']} {object['kind']}")
+```
+"""
+
+import datetime
+import logging
+from pathlib import Path
+from typing import Any
+
+import aiofiles
+import yaml
+
+from . import command
+from .kustomize import Kustomize
+from .manifest import HelmRelease, HelmRepository
+
+__all__ = [
+    "Helm",
+]
+
+_LOGGER = logging.getLogger(__name__)
+
+
+HELM_BIN = "helm"
+
+
+class RepositoryConfig:
+    """Generates a helm repository configuration from flux HelmRepository objects."""
+
+    def __init__(self, repos: list[HelmRepository]) -> None:
+        """Initialize RepositoryConfig."""
+        self._repos = repos
+
+    @property
+    def config(self) -> dict[str, Any]:
+        """Return a synthetic repository config object."""
+        now = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0)
+        return {
+            "apiVersion": "",
+            "generated": now.isoformat(),
+            "repositories": [
+                {
+                    "name": f"{repo.namespace}-{repo.name}",
+                    "url": repo.url,
+                }
+                for repo in self._repos
+            ],
+        }
+
+
+class Helm:
+    """Manages local HelmRepository state."""
+
+    def __init__(self, tmp_dir: Path, cache_dir: Path) -> None:
+        """Initialize Helm."""
+        self._tmp_dir = tmp_dir
+        self._repo_config_file = self._tmp_dir / "repository-config.yaml"
+        self._flags = [
+            "--registry-config",
+            "/dev/null",
+            "--repository-cache",
+            str(cache_dir),
+            "--repository-config",
+            str(self._repo_config_file),
+        ]
+        self._repos: list[HelmRepository] = []
+
+    def add_repo(self, repo: HelmRepository) -> None:
+        """Add the specified HelmRepository to the local config."""
+        self._repos.append(repo)
+
+    async def update(self) -> None:
+        """Return command line arguments to update the local repo.
+
+        Typically the repository must be updated before doing any chart templating.
+        """
+        content = yaml.dump(RepositoryConfig(self._repos).config)
+        async with aiofiles.open(str(self._repo_config_file), mode="w") as config_file:
+            await config_file.write(content)
+        await command.run([HELM_BIN, "repo", "update"] + self._flags)
+
+    async def template(self, release: HelmRelease, values: dict[str, Any]) -> Kustomize:
+        """Return command line arguments to template the specified chart."""
+        args = [
+            HELM_BIN,
+            "template",
+            release.name,
+            release.chart.chart_name,
+            "--namespace",
+            release.namespace,
+            "--skip-crds",  # Reduce size of output
+            "--version",
+            release.chart.version,
+        ]
+        if values:
+            values_path = self._tmp_dir / f"{release.release_name}-values.yaml"
+            async with aiofiles.open(values_path, mode="w") as values_file:
+                await values_file.write(yaml.dump(values))
+            args.extend(["--values", str(values_path)])
+        return Kustomize([args + self._flags])

flux_local/kustomize.py

@@ -3,30 +3,75 @@
 Kustomize build can be used to apply overlays and output a set of resources or artifacts
 for the cluster that can be either be parsed directly or piped into additional
 commands for processing and filtering by kustomize grep.
+
+This example returns the objects inside a Kustomization using `kustomize build`:
+```
+from flux_local.kustomize import Kustomize
+
+objects = await Kustomize.build('/path/to/objects').objects()
+for object in objects:
+    print(f"Found object {object['apiVersion']} {object['kind']}")
+```
+
+You can also filter documents to specific resource types or other fields:
+```
+from flux_local.kustomize import Kustomize
+
+objects = await Kustomize.build('/path/to/objects').grep('kind=ConfigMap').objects()
+for object in objects:
+    print(f"Found ConfigMap: {object['metadata']['name']}")
+```
+
 """
 
 from pathlib import Path
+from typing import Any, AsyncGenerator
+
+import yaml
+
+from . import command
 
 KUSTOMIZE_BIN = "kustomize"
 
 
-def build(path: Path) -> list[str]:
-    """Generate a kustomize build command for the specified path."""
-    return [KUSTOMIZE_BIN, "build", str(path)]
+class Kustomize:
+    """Library for issuing a kustomize command."""
+
+    def __init__(self, cmds: list[list[str]]) -> None:
+        """Initialize Kustomize."""
+        self._cmds = cmds
+
+    @classmethod
+    def build(cls, path: Path) -> "Kustomize":
+        """Build cluster artifacts from the specified path."""
+        return Kustomize(cmds=[[KUSTOMIZE_BIN, "build", str(path)]])
+
+    def grep(
+        self, expr: str, path: Path | None = None, invert: bool = False
+    ) -> "Kustomize":
+        """Filter resources based on an expression.
 
+        Example expressions:
+          `kind=HelmRelease`
+          `metadata.name=redis`
+        """
+        out = [KUSTOMIZE_BIN, "cfg", "grep", expr]
+        if invert:
+            out.append("--invert-match")
+        if path:
+            out.append(str(path))
+        return Kustomize(self._cmds + [out])
 
-def grep(expr: str, path: Path | None = None, invert: bool = False) -> list[str]:
-    """Generate a kustomize grep command to filter resources based on an expression.
+    async def run(self) -> str:
+        """Run the kustomize command and return the output as a string."""
+        return await command.run_piped(self._cmds)
 
-    Example expressions:
-      `kind=HelmRelease`
-      `metadata.name=redis`
+    async def _docs(self) -> AsyncGenerator[dict[str, Any], None]:
+        """Run the kustomize command and return the result documents."""
+        out = await self.run()
+        for doc in yaml.safe_load_all(out):
+            yield doc
 
-    The return value is a set of command args.
-    """
-    out = [KUSTOMIZE_BIN, "cfg", "grep", expr]
-    if invert:
-        out.append("--invert-match")
-    if path:
-        out.append(str(path))
-    return out
+    async def objects(self) -> list[dict[str, Any]]:
+        """Run the kustomize command and return the result cluster objects as a list."""
+        return [doc async for doc in self._docs()]

flux_local/manifest.py

@@ -17,18 +17,19 @@
     "Kustomization",
     "HelmRepository",
     "HelmRelease",
+    "HelmChart",
 ]
 
 
 @dataclass
-class HelmRelease:
-    """A representation of a Flux HelmRelease."""
+class HelmChart:
+    """A representation of an instantiation of a chart for a HelmRelease."""
 
     name: str
-    """The name of the HelmRelease."""
+    """The name of the chart within the HelmRepository."""
 
-    namespace: str
-    """The namespace that owns the HelmRelease."""
+    version: str
+    """The version of the chart."""
 
     repo_name: str
     """The name of the HelmRepository."""
@@ -37,29 +38,58 @@ class HelmRelease:
     """The namespace of the HelmRepository."""
 
     @classmethod
-    def from_doc(cls, doc: dict[str, Any]) -> "HelmRelease":
-        """Parse a HelmRelease from a kubernetes resource object."""
-        if not (metadata := doc.get("metadata")):
-            raise ValueError(f"Invalid {cls} missing metadata: {doc}")
-        if not (name := metadata.get("name")):
-            raise ValueError(f"Invalid {cls} missing metadata.name: {doc}")
-        if not (namespace := metadata.get("namespace")):
-            raise ValueError(f"Invalid {cls} missing metadata.namespace: {doc}")
+    def from_doc(cls, doc: dict[str, Any]) -> "HelmChart":
+        """Parse a HelmChart from a HelmRelease resource object."""
         if not (spec := doc.get("spec")):
             raise ValueError(f"Invalid {cls} missing spec: {doc}")
         if not (chart := spec.get("chart")):
             raise ValueError(f"Invalid {cls} missing spec.chart: {doc}")
         if not (chart_spec := chart.get("spec")):
             raise ValueError(f"Invalid {cls} missing spec.chart.spec: {doc}")
+        if not (chart := chart_spec.get("chart")):
+            raise ValueError(f"Invalid {cls} missing spec.chart.spec.chart: {doc}")
+        if not (version := chart_spec.get("version")):
+            raise ValueError(f"Invalid {cls} missing spec.chart.spec.version: {doc}")
         if not (source_ref := chart_spec.get("sourceRef")):
             raise ValueError(f"Invalid {cls} missing spec.chart.spec.sourceRef: {doc}")
         if "namespace" not in source_ref or "name" not in source_ref:
             raise ValueError(f"Invalid {cls} missing sourceRef fields: {doc}")
-        return cls(name, namespace, source_ref["name"], source_ref["namespace"])
+        return cls(chart, version, source_ref["name"], source_ref["namespace"])
 
     @property
-    def id_name(self) -> str:
-        """Identifier for the HelmRelease in tests."""
+    def chart_name(self) -> str:
+        """Identifier for the HelmChart."""
+        return f"{self.repo_namespace}-{self.repo_name}/{self.name}"
+
+
+@dataclass
+class HelmRelease:
+    """A representation of a Flux HelmRelease."""
+
+    name: str
+    """The name of the HelmRelease."""
+
+    namespace: str
+    """The namespace that owns the HelmRelease."""
+
+    chart: HelmChart
+    """A mapping to a specific helm chart for this HelmRelease."""
+
+    @classmethod
+    def from_doc(cls, doc: dict[str, Any]) -> "HelmRelease":
+        """Parse a HelmRelease from a kubernetes resource object."""
+        if not (metadata := doc.get("metadata")):
+            raise ValueError(f"Invalid {cls} missing metadata: {doc}")
+        if not (name := metadata.get("name")):
+            raise ValueError(f"Invalid {cls} missing metadata.name: {doc}")
+        if not (namespace := metadata.get("namespace")):
+            raise ValueError(f"Invalid {cls} missing metadata.namespace: {doc}")
+        chart = HelmChart.from_doc(doc)
+        return cls(name, namespace, chart)
+
+    @property
+    def release_name(self) -> str:
+        """Identifier for the HelmRelease."""
         return f"{self.namespace}-{self.name}"
 
 
@@ -92,8 +122,8 @@ def from_doc(cls, doc: dict[str, Any]) -> "HelmRepository":
         return cls(name, namespace, url)
 
     @property
-    def id_name(self) -> str:
-        """Identifier for the HelmRepository in tests."""
+    def repo_name(self) -> str:
+        """Identifier for the HelmRepository."""
         return f"{self.namespace}-{self.name}"
 
 

requirements.txt

@@ -16,6 +16,7 @@ pytest-asyncio==0.20.3
 pytest-cov==3.0.0
 python-slugify==8.0.0
 PyYAML==6.0
+types-aiofiles==22.1.0.6
 types-PyYAML==6.0.12.4
 typing-extensions==4.4.0
 wheel==0.37.1