Add OCIRepository as a parsed object

flux_local/git_repo.py

@@ -58,6 +58,7 @@
     Secret,
     SECRET_KIND,
     CONFIG_MAP_KIND,
+    OCIRepository,
 )
 from .exceptions import InputException
 from .context import trace_context
@@ -204,7 +205,11 @@ class ResourceVisitor:
     func: Callable[
         [
             Path,
-            Kustomization | HelmRelease | HelmRepository | ClusterPolicy,
+            Kustomization
+            | HelmRelease
+            | HelmRepository
+            | ClusterPolicy
+            | OCIRepository,
             kustomize.Kustomize | None,
         ],
         Awaitable[None],
@@ -265,11 +270,20 @@ class MetadataSelector:
     @property
     def predicate(
         self,
-    ) -> Callable[[Kustomization | HelmRelease | HelmRepository | ClusterPolicy], bool]:
+    ) -> Callable[
+        [Kustomization | HelmRelease | HelmRepository | ClusterPolicy | OCIRepository],
+        bool,
+    ]:
         """A predicate that selects Kustomization objects."""
 
         def predicate(
-            obj: Kustomization | HelmRelease | HelmRepository | ClusterPolicy,
+            obj: (
+                Kustomization
+                | HelmRelease
+                | HelmRepository
+                | ClusterPolicy
+                | OCIRepository
+            ),
         ) -> bool:
             if not self.enabled:
                 return False
@@ -323,6 +337,9 @@ class ResourceSelector:
     helm_release: MetadataSelector = field(default_factory=MetadataSelector)
     """HelmRelease objects to return."""
 
+    oci_repo: MetadataSelector = field(default_factory=MetadataSelector)
+    """OCIRepository objects to return."""
+
     cluster_policy: MetadataSelector = field(default_factory=MetadataSelector)
     """ClusterPolicy objects to return."""
 
@@ -568,11 +585,13 @@ async def build_kustomization(
     root: Path = selector.path.root
     kustomization_selector: MetadataSelector = selector.kustomization
     helm_repo_selector: MetadataSelector = selector.helm_repo
+    oci_repo_selector: MetadataSelector = selector.oci_repo
     helm_release_selector: MetadataSelector = selector.helm_release
     cluster_policy_selector: MetadataSelector = selector.cluster_policy
     if (
         not kustomization_selector.enabled
         and not helm_repo_selector.enabled
+        and not oci_repo_selector.visitor
         and not helm_release_selector.enabled
         and not cluster_policy_selector.enabled
         and not selector.doc_visitor
@@ -608,6 +627,8 @@ async def build_kustomization(
         kinds.append(CONFIG_MAP_KIND)
         if helm_repo_selector.enabled:
             kinds.append(HELM_REPO_KIND)
+        if oci_repo_selector.enabled:
+            kinds.append(OCI_REPO_KIND)
         if helm_release_selector.enabled:
             kinds.append(HELM_RELEASE_KIND)
             # Needed for expanding value references
@@ -640,6 +661,16 @@ async def build_kustomization(
                 ],
             )
         )
+        kustomization.oci_repos = list(
+            filter(
+                oci_repo_selector.predicate,
+                [
+                    OCIRepository.parse_doc(doc)
+                    for doc in docs
+                    if doc.get("kind") == OCI_REPO_KIND
+                ],
+            )
+        )
         kustomization.helm_releases = list(
             filter(
                 helm_release_selector.predicate,
@@ -793,6 +824,15 @@ async def update_kustomization(cluster: Cluster) -> None:
                             None,
                         )
 
+            if selector.oci_repo.visitor:
+                for kustomization in cluster.kustomizations:
+                    for oci_repo in kustomization.oci_repos:
+                        await selector.oci_repo.visitor.func(
+                            Path(kustomization.path),
+                            oci_repo,
+                            None,
+                        )
+
             if selector.helm_release.visitor:
                 for kustomization in cluster.kustomizations:
                     for helm_release in kustomization.helm_releases:

flux_local/manifest.py

@@ -50,6 +50,7 @@
 VALUE_B64_PLACEHOLDER = base64.b64encode(VALUE_PLACEHOLDER.encode())
 HELM_REPOSITORY = "HelmRepository"
 GIT_REPOSITORY = "GitRepository"
+GIT_REPOSITORY_DOMAIN = "source.toolkit.fluxcd.io"
 
 REPO_TYPE_DEFAULT = "default"
 REPO_TYPE_OCI = "oci"
@@ -275,6 +276,45 @@ def repo_name(self) -> str:
         return f"{self.namespace}-{self.name}"
 
 
+@dataclass
+class OCIRepository(BaseManifest):
+    """A representation of a flux OCIRepository."""
+
+    name: str
+    """The name of the OCIRepository."""
+
+    namespace: str
+    """The namespace of owning the OCIRepository."""
+
+    url: str
+    """The URL to the repository."""
+
+    @classmethod
+    def parse_doc(cls, doc: dict[str, Any]) -> "OCIRepository":
+        """Parse a HelmRepository from a kubernetes resource."""
+        _check_version(doc, GIT_REPOSITORY_DOMAIN)
+        if not (metadata := doc.get("metadata")):
+            raise InputException(f"Invalid {cls} missing metadata: {doc}")
+        if not (name := metadata.get("name")):
+            raise InputException(f"Invalid {cls} missing metadata.name: {doc}")
+        if not (namespace := metadata.get("namespace")):
+            raise InputException(f"Invalid {cls} missing metadata.namespace: {doc}")
+        if not (spec := doc.get("spec")):
+            raise InputException(f"Invalid {cls} missing spec: {doc}")
+        if not (url := spec.get("url")):
+            raise InputException(f"Invalid {cls} missing spec.url: {doc}")
+        return cls(
+            name=name,
+            namespace=namespace,
+            url=url,
+        )
+
+    @property
+    def repo_name(self) -> str:
+        """Identifier for the OCIRepository."""
+        return f"{self.namespace}-{self.name}"
+
+
 @dataclass
 class ClusterPolicy(BaseManifest):
     """A kyverno policy object."""
@@ -412,6 +452,9 @@ class Kustomization(BaseManifest):
     helm_repos: list[HelmRepository] = field(default_factory=list)
     """The set of HelmRepositories represented in this kustomization."""
 
+    oci_repos: list[OCIRepository] = field(default_factory=list)
+    """The set of OCIRepositories represented in this kustomization."""
+
     helm_releases: list[HelmRelease] = field(default_factory=list)
     """The set of HelmRelease represented in this kustomization."""
 
@@ -564,6 +607,15 @@ def helm_repos(self) -> list[HelmRepository]:
             for repo in kustomization.helm_repos
         ]
 
+    @property
+    def oci_repos(self) -> list[OCIRepository]:
+        """Return the list of OCIRepository objects from all Kustomizations."""
+        return [
+            repo
+            for kustomization in self.kustomizations
+            for repo in kustomization.oci_repos
+        ]
+
     @property
     def helm_releases(self) -> list[HelmRelease]:
         """Return the list of HelmRelease objects from all Kustomizations."""

flux_local/tool/get.py

@@ -68,16 +68,17 @@ async def run(  # type: ignore[no-untyped-def]
         results: list[dict[str, str]] = []
         cols = ["name", "path"]
         if output == "wide":
-            cols.extend(["helmrepos", "releases"])
+            cols.extend(["helmrepos", "ocirepos", "releases"])
         if query.kustomization.namespace is None:
             cols.insert(0, "namespace")
         if len(manifest.clusters) > 1:
             cols.insert(0, "cluster")
         for cluster in manifest.clusters:
             for ks in cluster.kustomizations:
-                value = { k: v for k, v in ks.compact_dict().items() if k in cols }
+                value = {k: v for k, v in ks.compact_dict().items() if k in cols}
                 if output == "wide":
                     value["helmrepos"] = len(ks.helm_repos)
+                    value["ocirepos"] = len(ks.oci_repos)
                     value["releases"] = len(ks.helm_releases)
                 value["cluster"] = cluster.path
                 results.append(value)
@@ -126,7 +127,9 @@ async def run(  # type: ignore[no-untyped-def]
         results: list[dict[str, Any]] = []
         for cluster in manifest.clusters:
             for helmrelease in cluster.helm_releases:
-                value = { k: v for k, v in helmrelease.compact_dict().items() if k in cols }
+                value = {
+                    k: v for k, v in helmrelease.compact_dict().items() if k in cols
+                }
                 value["revision"] = str(helmrelease.chart.version)
                 value["chart"] = f"{helmrelease.namespace}-{helmrelease.chart.name}"
                 value["source"] = helmrelease.chart.repo_name
@@ -230,7 +233,7 @@ async def run(  # type: ignore[no-untyped-def]
         cols = ["path", "kustomizations"]
         results: list[dict[str, Any]] = []
         for cluster in manifest.clusters:
-            value = { k: v for k, v in cluster.compact_dict().items() if k in cols }
+            value = {k: v for k, v in cluster.compact_dict().items() if k in cols}
             value["kustomizations"] = len(cluster.kustomizations)
             results.append(value)
 

flux_local/tool/visitor.py

@@ -18,6 +18,7 @@
     HelmRepository,
     ClusterPolicy,
     Manifest,
+    OCIRepository,
 )
 
 
@@ -32,7 +33,9 @@
 ]
 
 
-ResourceType = Kustomization | HelmRelease | HelmRepository | ClusterPolicy
+ResourceType = (
+    Kustomization | HelmRelease | HelmRepository | ClusterPolicy | OCIRepository
+)
 
 
 @dataclass(frozen=True, order=True)