INNER CODE UNIT · Python

_list_namespaced_or_global_objects_batched

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

    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()

        try:
            if namespace == "*":
                requests = [
                    loop.run_in_executor(
                        self.executor,
                        lambda: all_namespaces_request(

View source record →