Streamline manifest output format (#22)

Update the manifest data model to include `HelmRelease` `values`. The manifest is compacted when persisted, and now the same object can be used for a raw release, or just persisted manifest. Increase a breaking change to update all objects to be pyndantic models to simplify how exclusion works so that it can work on nested objects.
allenporter · Feb 6, 2023 · bb9b1eb · bb9b1eb
1 parent f5565cc
commit bb9b1eb
Show file tree

Hide file tree

Showing 2 changed files with 79 additions and 46 deletions.
diff --git a/flux_local/manifest.py b/flux_local/manifest.py
@@ -5,14 +5,12 @@
 e.g. such as writing management plan for resources.
 """
 
-import dataclasses
-import datetime
 from pathlib import Path
-from typing import Any, cast
+from typing import Any, Optional, cast
 
 import aiofiles
 import yaml
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 __all__ = [
     "read_manifest",
@@ -27,8 +25,7 @@
 ]
 
 
-@dataclasses.dataclass
-class HelmChart:
+class HelmChart(BaseModel):
     """A representation of an instantiation of a chart for a HelmRelease."""
 
     name: str
@@ -60,16 +57,20 @@ def from_doc(cls, doc: dict[str, Any]) -> "HelmChart":
             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(chart, version, source_ref["name"], source_ref["namespace"])
+        return cls(
+            name=chart,
+            version=version,
+            repo_name=source_ref["name"],
+            repo_namespace=source_ref["namespace"],
+        )
 
     @property
     def chart_name(self) -> str:
         """Identifier for the HelmChart."""
         return f"{self.repo_namespace}-{self.repo_name}/{self.name}"
 
 
-@dataclasses.dataclass
-class HelmRelease:
+class HelmRelease(BaseModel):
     """A representation of a Flux HelmRelease."""
 
     name: str
@@ -81,6 +82,9 @@ class HelmRelease:
     chart: HelmChart
     """A mapping to a specific helm chart for this HelmRelease."""
 
+    values: Optional[dict[str, Any]] = None
+    """The values to install in the chart."""
+
     @classmethod
     def from_doc(cls, doc: dict[str, Any]) -> "HelmRelease":
         """Parse a HelmRelease from a kubernetes resource object."""
@@ -91,16 +95,20 @@ def from_doc(cls, doc: dict[str, Any]) -> "HelmRelease":
         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)
+        return cls(
+            name=name,
+            namespace=namespace,
+            chart=chart,
+            values=doc["spec"].get("values"),
+        )
 
     @property
     def release_name(self) -> str:
         """Identifier for the HelmRelease."""
         return f"{self.namespace}-{self.name}"
 
 
-@dataclasses.dataclass
-class HelmRepository:
+class HelmRepository(BaseModel):
     """A representation of a flux HelmRepository."""
 
     name: str
@@ -125,16 +133,15 @@ def from_doc(cls, doc: dict[str, Any]) -> "HelmRepository":
             raise ValueError(f"Invalid {cls} missing spec: {doc}")
         if not (url := spec.get("url")):
             raise ValueError(f"Invalid {cls} missing spec.url: {doc}")
-        return cls(name, namespace, url)
+        return cls(name=name, namespace=namespace, url=url)
 
     @property
     def repo_name(self) -> str:
         """Identifier for the HelmRepository."""
         return f"{self.namespace}-{self.name}"
 
 
-@dataclasses.dataclass
-class Kustomization:
+class Kustomization(BaseModel):
     """A Kustomization is a set of declared cluster artifacts.
 
     This represents a flux Kustomization that points to a path that
@@ -148,20 +155,32 @@ class Kustomization:
     path: str
     """The local repo path to the kustomization."""
 
-    helm_repos: list[HelmRepository]
+    helm_repos: list[HelmRepository] = Field(default_factory=list)
     """The set of HelmRepositories represented in this kustomization."""
 
-    helm_releases: list[HelmRelease]
+    helm_releases: list[HelmRelease] = Field(default_factory=list)
     """The set of HelmRelease represented in this kustomization."""
 
+    @classmethod
+    def from_doc(cls, doc: dict[str, Any]) -> "Kustomization":
+        """Parse a partial Kustomization from a kubernetes resource."""
+        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 (spec := doc.get("spec")):
+            raise ValueError(f"Invalid {cls} missing spec: {doc}")
+        if not (path := spec.get("path")):
+            raise ValueError(f"Invalid {cls} missing spec.path: {doc}")
+        return Kustomization(name=name, path=path)
+
     @property
     def id_name(self) -> str:
         """Identifier for the Kustomization in tests"""
         return f"{self.path}"
 
 
-@dataclasses.dataclass
-class Cluster:
+class Cluster(BaseModel):
     """A set of nodes that run containerized applications.
 
     Many flux git repos will only have a single flux cluster, though
@@ -174,28 +193,21 @@ class Cluster:
     path: str
     """The local git repo path to the Kustomization objects for the cluster."""
 
-    kustomizations: list[Kustomization]
+    kustomizations: list[Kustomization] = Field(default_factory=list)
     """A list of flux Kustomizations for the cluster."""
 
-    def helm_repo_config(self) -> dict[str, Any]:
-        """Return a synthetic HelmRepoistory config."""
-        now = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0)
-        repos = []
-        for kustomize in self.kustomizations:
-            repos.extend(
-                [
-                    {
-                        "name": f"{repo.namespace}-{repo.name}",
-                        "url": repo.url,
-                    }
-                    for repo in kustomize.helm_repos
-                ]
-            )
-        return {
-            "apiVersion": "",
-            "generated": now.isoformat(),
-            "repositories": repos,
-        }
+    @classmethod
+    def from_doc(cls, doc: dict[str, Any]) -> "Cluster":
+        """Parse a partial Kustomization from a kubernetes resource."""
+        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 (spec := doc.get("spec")):
+            raise ValueError(f"Invalid {cls} missing spec: {doc}")
+        if not (path := spec.get("path")):
+            raise ValueError(f"Invalid {cls} missing spec.path: {doc}")
+        return Cluster(name=name, path=path)
 
     @property
     def id_name(self) -> str:
@@ -218,14 +230,35 @@ def parse_yaml(content: str) -> "Manifest":
         return Manifest(clusters=doc["spec"])
 
     def yaml(self) -> str:
-        """Serialize the manifest as a yaml file."""
+        """Serialize the manifest as a yaml file.
+
+        The contents of the cluster will be compacted to remove values that
+        exist in the live cluster but do not make sense to be persisted in the
+        manifest on disk.
+        """
+        data = self.dict(
+            exclude={
+                "clusters": {
+                    "__all__": {
+                        "kustomizations": {
+                            "__all__": {
+                                "helm_releases": {
+                                    "__all__": {
+                                        "values": True,
+                                        "chart": {
+                                            "version": True,
+                                        },
+                                    },
+                                },
+                            },
+                        },
+                    },
+                },
+            }
+        )
         return cast(
             str,
-            yaml.dump(
-                {"spec": [dataclasses.asdict(cluster) for cluster in self.clusters]},
-                sort_keys=False,
-                explicit_start=True,
-            ),
+            yaml.dump({"spec": data["clusters"]}, sort_keys=False, explicit_start=True),
         )
 
 

diff --git a/setup.cfg b/setup.cfg
@@ -1,6 +1,6 @@
 [metadata]
 name = flux-local
-version = 0.0.3
+version = 0.0.4
 description = flux-local is a python library and set of tools for managing a flux gitops repository, with validation steps to help improve quality of commits, PRs, and general local testing.
 long_description = file: README.md
 long_description_content_type = text/markdown