INNER CODE UNIT · Python

_list_deployments

robusta-dev/krr · robusta_krr/core/integrations/kubernetes/__init__.py:419

    def _list_deployments(self) -> list[K8sObjectData]:
        return self._list_scannable_objects(
            kind="Deployment",
            all_namespaces_request=self.apps.list_deployment_for_all_namespaces,
            namespaced_request=self.apps.list_namespaced_deployment,
            extract_containers=lambda item: item.spec.template.spec.containers,
        )

    def _list_rollouts(self) -> list[K8sObjectData]:
        async def _extract_containers(item: Any) -> list[V1Container]:
            if item.spec.template is not None:
                return item.spec.template.spec.containers

            loop = asyncio.get_running_loop()

            logging.debug(
                f"Rollout has workloadRef, fetching template for {item.metadata.name} in {item.metadata.namespace}"
            )

View source record →