Update tg-show-flow-classes CLI

This commit is contained in:
Cyber MacGeddon 2025-09-26 10:25:17 +01:00
parent 9a7d1f29fb
commit a029cc8094

View file

@ -5,38 +5,90 @@ Shows all defined flow classes.
import argparse import argparse
import os import os
import tabulate import tabulate
from trustgraph.api import Api from trustgraph.api import Api, ConfigKey
import json import json
default_url = os.getenv("TRUSTGRAPH_URL", 'http://localhost:8088/') 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): 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: if len(class_names) == 0:
print("No flows.") print("No flow classes.")
return return
classes = []
for class_name in class_names: for class_name in class_names:
cls = api.get_class(class_name) cls = flow_api.get_class(class_name)
classes.append((
class_name,
cls.get("description", ""),
", ".join(cls.get("tags", [])),
))
print(tabulate.tabulate( print(f"Flow Class: {class_name}")
classes, print(f"Description: {cls.get('description', 'No description')}")
tablefmt="pretty",
maxcolwidths=[None, 40, 20], tags = cls.get("tags", [])
stralign="left", if tags:
headers = ["flow class", "description", "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(): def main():