[Integration][Argocd] - Add support for managed resources (#430)

integrations/argocd/.port/spec.yaml

@@ -10,6 +10,7 @@ features:
       - kind: project
       - kind: application
       - kind: deployment-history
+      - kind: managed-resource
 configurations:
   - name: token
     required: true

integrations/argocd/CHANGELOG.md

@@ -6,6 +6,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 <!-- towncrier release notes start -->
+# Port_Ocean 0.1.30 (2024-03-18)
+
+### Improvements
+
+- Added support for Application managed resources kind
+
+
 # Port_Ocean 0.1.29 (2024-03-18)
 
 ### Improvements

integrations/argocd/client.py

@@ -15,6 +15,7 @@ class ObjectKind(StrEnum):
 
 class ResourceKindsWithSpecialHandling(StrEnum):
     DEPLOYMENT_HISTORY = "deployment-history"
+    MANAGED_RESOURCE = "managed-resource"
 
 
 class ArgocdClient:
@@ -52,6 +53,7 @@ async def _send_api_request(
             raise
 
     async def get_resources(self, resource_kind: ObjectKind) -> list[dict[str, Any]]:
+        logger.info(f"Fetching ArgoCD resource: {resource_kind}")
         url = f"{self.api_url}/{resource_kind}s"
         response_data = (await self._send_api_request(url=url))["items"]
         return response_data
@@ -63,11 +65,18 @@ async def get_application_by_name(self, name: str) -> dict[str, Any]:
 
     async def get_deployment_history(self) -> list[dict[str, Any]]:
         """The ArgoCD application route returns a history of all deployments. This function reuses the output of the application endpoint"""
-        logger.info("fetching Argocd deployment history from applications endpoint")
         applications = await self.get_resources(resource_kind=ObjectKind.APPLICATION)
         all_history = [
             {**history_item, "__applicationId": application["metadata"]["uid"]}
             for application in applications
             for history_item in application["status"].get("history", [])
         ]
         return all_history
+
+    async def get_managed_resources(
+        self, application_name: str
+    ) -> list[dict[str, Any]]:
+        logger.info(f"Fetching managed resources for application: {application_name}")
+        url = f"{self.api_url}/{ObjectKind.APPLICATION}s/{application_name}/managed-resources"
+        managed_resources = (await self._send_api_request(url=url))["items"]
+        return managed_resources

integrations/argocd/main.py

@@ -1,6 +1,6 @@
 from loguru import logger
 from port_ocean.context.ocean import ocean
-from port_ocean.core.ocean_types import RAW_RESULT
+from port_ocean.core.ocean_types import RAW_RESULT, ASYNC_GENERATOR_RESYNC_TYPE
 from client import ArgocdClient, ObjectKind, ResourceKindsWithSpecialHandling
 from fastapi import Request
 
@@ -14,8 +14,6 @@ def init_client() -> ArgocdClient:
 
 @ocean.on_resync()
 async def on_resources_resync(kind: str) -> RAW_RESULT:
-    logger.info(f"Listing ArgoCD resource: {kind}")
-
     if kind in iter(ResourceKindsWithSpecialHandling):
         logger.info(f"Kind {kind} has a special handling. Skipping...")
         return []
@@ -26,12 +24,29 @@ async def on_resources_resync(kind: str) -> RAW_RESULT:
 
 @ocean.on_resync(kind=ResourceKindsWithSpecialHandling.DEPLOYMENT_HISTORY)
 async def on_history_resync(kind: str) -> RAW_RESULT:
-    logger.info("Listing ArgoCD deployment history")
     argocd_client = init_client()
 
     return await argocd_client.get_deployment_history()
 
 
+@ocean.on_resync(kind=ResourceKindsWithSpecialHandling.MANAGED_RESOURCE)
+async def on_managed_resources_resync(kind: str) -> ASYNC_GENERATOR_RESYNC_TYPE:
+    argocd_client = init_client()
+
+    applications = await argocd_client.get_resources(
+        resource_kind=ObjectKind.APPLICATION
+    )
+    for application in applications:
+        managed_resources = await argocd_client.get_managed_resources(
+            application_name=application["metadata"]["name"]
+        )
+        application_resource = [
+            {**managed_resource, "__applicationId": application["metadata"]["uid"]}
+            for managed_resource in managed_resources
+        ]
+        yield application_resource
+
+
 @ocean.router.post("/webhook")
 async def on_application_event_webhook_handler(request: Request) -> None:
     data = await request.json()

integrations/argocd/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "argocd"
-version = "0.1.29"
+version = "0.1.30"
 description = "Argo CD integration powered by Ocean"
 authors = ["Isaac Coffie <[email protected]>"]