(self, args, config, plugin_registry: PluginRegistry)
Entrypoint for the inspect command
Entrypoint for the inspect command
Entrypoint for the inspect command
def handle(self, args, config, plugin_registry: PluginRegistry):
        """Entrypoint for the inspect command"""
        command_utilities.setup_logging(args.log_level or "warning", config)
        bus_module, bus = command_utilities.import_bus(args)
        api_names: List[str]

        block(bus.client.lazy_load_now())

        # Locally registered APIs
        api_names = [api.meta.name for api in bus.client.api_registry.all()]

        # APIs registered to other services on the bus
        for api_name in bus.client.schema.api_names:
            if api_name not in api_names:
                api_names.append(api_name)

        if not args.internal and not args.api:
            # Hide internal APIs if we don't want them
            api_names = [api_name for api_name in api_names if not api_name.startswith("internal.")]

        if args.api and args.api not in api_names:
            sys.stderr.write(
                f"Specified API was not found locally or within the schema on the bus.\n"
                f"Ensure a Lightbus worker is running for this API.\n"
                f"Cannot continue.\n"
            )
            sys.exit(1)

        api_names_to_inspect = []
        for api_name in api_names:
            if not args.api or self.wildcard_match(args.api, api_name):
                api_names_to_inspect.append(api_name)

        if len(api_names_to_inspect) != 1 and args.follow:
            sys.stderr.write(
                f"The --follow option is only available when following a single API.\n"
                f"Please specify the --api option to select a single API to follow.\n"
            )
            sys.exit(1)

        try:
            for api_name in api_names_to_inspect:
                if not args.api or self.wildcard_match(args.api, api_name):
                    logger.debug(f"Inspecting {api_name}")
                    block(self.search_in_api(args, api_name, bus))
                else:
                    logger.debug(f"API {api_name} did not match {args.api}. Skipping")
        except KeyboardInterrupt:
            logger.info("Stopped by user")
def handle ( self , args , config , plugin_registry : PluginRegistry ) : command_utilities . setup_logging ( args . log_level or "warning" , config ) bus_module , bus = command_utilities . import_bus ( args ) api_names : List [ str ] block ( bus . client . lazy_load_now ( ) ) # Locally registered APIs api_names = [ api . meta . name for api in bus . client . api_registry . all ( ) ] # APIs registered to other services on the bus for api_name in bus . client . schema . api_names : if api_name not in api_names : api_names . append ( api_name ) if not args . internal and not args . api : # Hide internal APIs if we don't want them api_names = [ api_name for api_name in api_names if not api_name . startswith ( "internal." ) ] if args . api and args . api not in api_names : sys . stderr . write ( f"Specified API was not found locally or within the schema on the bus.\n" f"Ensure a Lightbus worker is running for this API.\n" f"Cannot continue.\n" ) sys . exit ( 1 ) api_names_to_inspect = [ ] for api_name in api_names : if not args . api or self . wildcard_match ( args . api , api_name ) : api_names_to_inspect . append ( api_name ) if len ( api_names_to_inspect ) != 1 and args . follow : sys . stderr . write ( f"The --follow option is only available when following a single API.\n" f"Please specify the --api option to select a single API to follow.\n" ) sys . exit ( 1 ) try : for api_name in api_names_to_inspect : if not args . api or self . wildcard_match ( args . api , api_name ) : logger . debug ( f"Inspecting {api_name}" ) block ( self . search_in_api ( args , api_name , bus ) ) else : logger . debug ( f"API {api_name} did not match {args.api}. Skipping" ) except KeyboardInterrupt : logger . info ( "Stopped by user" )