Get task input schema. Args: task_name (str): The task name. input_type (type): The input type
(task_name: str, input_type: type)
| 117 | |
| 118 | |
| 119 | def get_input_schema(task_name: str, input_type: type): |
| 120 | """Get task input schema. |
| 121 | |
| 122 | Args: |
| 123 | task_name (str): The task name. |
| 124 | input_type (type): The input type |
| 125 | """ |
| 126 | if input_type is None: |
| 127 | task_inputs = TASK_INPUTS[task_name] |
| 128 | if isinstance(task_inputs, |
| 129 | str): # only one input field, key is task_inputs |
| 130 | return { |
| 131 | 'type': 'object', |
| 132 | 'properties': { |
| 133 | task_inputs: INPUT_TYPE_SCHEMA[task_inputs] |
| 134 | } |
| 135 | } |
| 136 | else: |
| 137 | task_inputs = input_type |
| 138 | |
| 139 | if isinstance(task_inputs, str): # no input key |
| 140 | return INPUT_TYPE_SCHEMA[task_inputs] |
| 141 | elif input_type is None and isinstance(task_inputs, list): |
| 142 | for item in task_inputs: |
| 143 | # for list, server only support dict format. |
| 144 | if isinstance(item, dict): |
| 145 | return get_input_schema(None, item) |
| 146 | elif isinstance(task_inputs, tuple) or isinstance(task_inputs, list): |
| 147 | input_schema = {'type': 'array', 'items': {}} |
| 148 | for item in task_inputs: |
| 149 | if isinstance(item, dict): |
| 150 | item_schema = get_input_schema(None, item) |
| 151 | input_schema['items']['type'] = item_schema |
| 152 | return input_schema |
| 153 | else: |
| 154 | input_schema['items'] = INPUT_TYPE_SCHEMA[item] |
| 155 | return input_schema |
| 156 | |
| 157 | elif isinstance(task_inputs, dict): |
| 158 | input_schema = { |
| 159 | 'type': 'object', |
| 160 | 'properties': {} |
| 161 | } # key input key, value input type |
| 162 | for k, v in task_inputs.items(): |
| 163 | input_schema['properties'][k] = get_input_schema(None, v) |
| 164 | return input_schema |
| 165 | else: |
| 166 | raise ValueError(f'invalid input_type definition {task_inputs}') |
| 167 | |
| 168 | |
| 169 | def get_output_schema(task_name: str): |
no test coverage detected
searching dependent graphs…