INNER CODE UNIT · Python

result

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

        result = [item for request_result in await asyncio.gather(*requests) for item in request_result.items]

        logger.debug(f"Found {len(result)} {kind} in {self.cluster}")
        return result

    async def _list_scannable_objects(
        self,
        kind: KindLiteral,
        all_namespaces_request: Callable,
        namespaced_request: Callable,
        extract_containers: Callable[[Any], Union[Iterable[V1Container], Awaitable[Iterable[V1Container]]]],
        filter_workflows: Optional[Callable[[Any], bool]] = None,
    ) -> list[K8sObjectData]:
        if not self._should_list_resource(kind):
            logger.debug(f"Skipping {kind}s in {self.cluster}")
            return []

        if not self.__kind_available[kind]:

View source record →