get_operations_collection.py

async def get_operations_collection(
    module_list: List[ModuleType] = [inputs, processors, outputs]
) -> List[Dict[str, Any]]:
    operations_collection = []
    for module in module_list:
        importlib.invalidate_caches()
        module = importlib.reload(module)
        for _class in get_classes_from_module(module):
            if hasattr(_class, "_node_subtype"):
                args = [arg for arg in get_init_args_from_class(_class)]
                operations_collection.append(
                    {
                        "operation_id": f"{_class._node_type}.{_class._node_subtype}",
                        "nodeType": _class._node_type,
                        "nodeSubtype": _class._node_subtype,
                        "name": _class._node_name,
                        "args": args,
                    }
                )
    return operations_collection