INNER CODE UNIT · Python

_list_jobs_for_cronjobs

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

    async def _list_jobs_for_cronjobs(self, namespace: str) -> list[V1Job]:
        if namespace not in self.__jobs_for_cronjobs:
            loop = asyncio.get_running_loop()

            async with self.__jobs_loading_locks[namespace]:
                logging.debug(f"Loading jobs for cronjobs in {namespace}")
                ret = await loop.run_in_executor(
                    self.executor,
                    lambda: self.batch.list_namespaced_job(namespace=namespace),
                )
                self.__jobs_for_cronjobs[namespace] = ret.items

        return self.__jobs_for_cronjobs[namespace]

    async def list_pods(self, object: K8sObjectData) -> list[PodData]:
        loop = asyncio.get_running_loop()

        if object.kind == "CronJob":

View source record →