INNER CODE UNIT · Python

_is_job_grouped

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

    def _is_job_grouped(self, job: V1Job) -> bool:
        """Check if a job has any of the grouping labels."""
        if not settings.job_grouping_labels or not job.metadata.labels:
            return False
        return any(label in job.metadata.labels for label in settings.job_grouping_labels)

    async def _list_namespaced_or_global_objects_batched(
        self,
        kind: KindLiteral,
        all_namespaces_request: Callable,
        namespaced_request: Callable,
        namespace: str,
        limit: Optional[int] = None,
        continue_ref: Optional[str] = None,
    ) -> tuple[list[Any], Optional[str]]:
        logger.debug("Listing %s in %s with batching (limit=%d)", kind, self.cluster, limit)
        loop = asyncio.get_running_loop()

View source record →