diff --git a/trustgraph-cli/trustgraph/cli/show_flow_classes.py b/trustgraph-cli/trustgraph/cli/show_flow_classes.py index 4cf6fc2f..f28459be 100644 --- a/trustgraph-cli/trustgraph/cli/show_flow_classes.py +++ b/trustgraph-cli/trustgraph/cli/show_flow_classes.py @@ -5,38 +5,90 @@ Shows all defined flow classes. import argparse import os import tabulate -from trustgraph.api import Api +from trustgraph.api import Api, ConfigKey import json default_url = os.getenv("TRUSTGRAPH_URL", 'http://localhost:8088/') +def format_parameters(params_metadata, config_api): + """ + Format parameter metadata for display + + Args: + params_metadata: Parameter definitions from flow class + config_api: API client to get parameter type information + + Returns: + Formatted string describing parameters + """ + if not params_metadata: + return "None" + + param_list = [] + + # Sort parameters by order if available + sorted_params = sorted( + params_metadata.items(), + key=lambda x: x[1].get("order", 999) + ) + + for param_name, param_meta in sorted_params: + description = param_meta.get("description", param_name) + param_type = param_meta.get("type", "unknown") + + # Get type information if available + type_info = param_type + if config_api: + try: + key = ConfigKey("parameter-types", param_type) + type_def_value = config_api.get([key])[0].value + param_type_def = json.loads(type_def_value) + + # Add default value if available + default = param_type_def.get("default") + if default is not None: + type_info = f"{param_type} (default: {default})" + + except: + # If we can't get type definition, just show the type name + pass + + param_list.append(f" {param_name}: {description} [{type_info}]") + + return "\n".join(param_list) + def show_flow_classes(url): - api = Api(url).flow() + api = Api(url) + flow_api = api.flow() + config_api = api.config() - class_names = api.list_classes() + class_names = flow_api.list_classes() if len(class_names) == 0: - print("No flows.") + print("No flow classes.") return - classes = [] - for class_name in class_names: - cls = api.get_class(class_name) - classes.append(( - class_name, - cls.get("description", ""), - ", ".join(cls.get("tags", [])), - )) + cls = flow_api.get_class(class_name) - print(tabulate.tabulate( - classes, - tablefmt="pretty", - maxcolwidths=[None, 40, 20], - stralign="left", - headers = ["flow class", "description", "tags"], - )) + print(f"Flow Class: {class_name}") + print(f"Description: {cls.get('description', 'No description')}") + + tags = cls.get("tags", []) + if tags: + print(f"Tags: {', '.join(tags)}") + + # Show parameters if they exist + parameters = cls.get("parameters", {}) + if parameters: + print("Parameters:") + param_str = format_parameters(parameters, config_api) + print(param_str) + else: + print("Parameters: None") + + print() # Blank line between flow classes def main():