(prompt_id, prompt, item, validated, visiting=None)
| 832 | |
| 833 | |
| 834 | async def validate_inputs(prompt_id, prompt, item, validated, visiting=None): |
| 835 | if visiting is None: |
| 836 | visiting = [] |
| 837 | |
| 838 | unique_id = item |
| 839 | if unique_id in validated: |
| 840 | return validated[unique_id] |
| 841 | |
| 842 | if unique_id in visiting: |
| 843 | cycle_path_nodes = visiting[visiting.index(unique_id):] + [unique_id] |
| 844 | cycle_nodes = list(dict.fromkeys(cycle_path_nodes)) |
| 845 | cycle_path = " -> ".join(f"{node_id} ({prompt[node_id]['class_type']})" for node_id in cycle_path_nodes) |
| 846 | for node_id in cycle_nodes: |
| 847 | validated[node_id] = (False, [{ |
| 848 | "type": "dependency_cycle", |
| 849 | "message": "Dependency cycle detected", |
| 850 | "details": cycle_path, |
| 851 | "extra_info": { |
| 852 | "node_id": node_id, |
| 853 | "cycle_nodes": cycle_nodes, |
| 854 | } |
| 855 | }], node_id) |
| 856 | return validated[unique_id] |
| 857 | |
| 858 | inputs = prompt[unique_id]['inputs'] |
| 859 | class_type = prompt[unique_id]['class_type'] |
| 860 | obj_class = nodes.NODE_CLASS_MAPPINGS[class_type] |
| 861 | |
| 862 | errors = [] |
| 863 | valid = True |
| 864 | |
| 865 | v3_data = None |
| 866 | validate_function_inputs = [] |
| 867 | validate_has_kwargs = False |
| 868 | if issubclass(obj_class, _ComfyNodeInternal): |
| 869 | obj_class: _io._ComfyNodeBaseInternal |
| 870 | class_inputs = obj_class.INPUT_TYPES() |
| 871 | class_inputs, _, v3_data = _io.get_finalized_class_inputs(class_inputs, inputs) |
| 872 | validate_function_name = "validate_inputs" |
| 873 | validate_function = first_real_override(obj_class, validate_function_name) |
| 874 | else: |
| 875 | class_inputs = obj_class.INPUT_TYPES() |
| 876 | validate_function_name = "VALIDATE_INPUTS" |
| 877 | validate_function = getattr(obj_class, validate_function_name, None) |
| 878 | if validate_function is not None: |
| 879 | argspec = inspect.getfullargspec(validate_function) |
| 880 | validate_function_inputs = argspec.args |
| 881 | validate_has_kwargs = argspec.varkw is not None |
| 882 | received_types = {} |
| 883 | |
| 884 | valid_inputs = set(class_inputs.get('required',{})).union(set(class_inputs.get('optional',{}))) |
| 885 | |
| 886 | for x in valid_inputs: |
| 887 | input_type, input_category, extra_info = get_input_info(obj_class, x, class_inputs) |
| 888 | assert extra_info is not None |
| 889 | if x not in inputs: |
| 890 | if input_category == "required": |
| 891 | details = f"{x}" if not v3_data else x.split(".")[-1] |
no test coverage detected