Get task output schema. Args: task_name (str): The task name.
(task_name: str)
| 167 | |
| 168 | |
| 169 | def get_output_schema(task_name: str): |
| 170 | """Get task output schema. |
| 171 | |
| 172 | Args: |
| 173 | task_name (str): The task name. |
| 174 | """ |
| 175 | task_outputs = TASK_OUTPUTS[task_name] |
| 176 | output_schema = {'type': 'object', 'properties': {}} |
| 177 | if not isinstance(task_outputs, list): |
| 178 | raise ValueError('TASK_OUTPUTS for %s is not list.' % task_name) |
| 179 | else: |
| 180 | for output_key in task_outputs: |
| 181 | output_schema['properties'][output_key] = OutputTypeSchema[ |
| 182 | output_key] |
| 183 | return output_schema |
| 184 | |
| 185 | |
| 186 | def get_input_info(task_name: str): |
no outgoing calls
no test coverage detected
searching dependent graphs…