INNER CODE UNIT · Python

_list_namespaced_or_global_objects

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

    async def _list_namespaced_or_global_objects(
        self, kind: KindLiteral, all_namespaces_request: Callable, namespaced_request: Callable
    ) -> list[Any]:
        logger.debug(f"Listing {kind}s in {self.cluster}")
        loop = asyncio.get_running_loop()

        if self.namespaces == "*":
            requests = [
                loop.run_in_executor(
                    self.executor,
                    lambda: all_namespaces_request(
                        watch=False,
                        label_selector=settings.selector,
                    ),
                )
            ]
        else:
            requests = [

View source record →