INNER CODE UNIT · Python

_is_job_owned_by_cronjob

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

    def _is_job_owned_by_cronjob(self, job: V1Job) -> bool:
        """Check if a job is owned by a CronJob."""
        return any(owner.kind == "CronJob" for owner in job.metadata.owner_references or [])

    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,

View source record →