(config, task, group=None, yaml_path=None)
| 137 | yaml_path: str = None, |
| 138 | ) -> ConfigurableTask: |
| 139 | def load_task(config, task, group=None, yaml_path=None): |
| 140 | if "include" in config: |
| 141 | assert yaml_path is not None |
| 142 | config.update( |
| 143 | utils.load_yaml_config( |
| 144 | yaml_path, |
| 145 | yaml_config={"include": config.pop("include")}, |
| 146 | mode="full", |
| 147 | ) |
| 148 | ) |
| 149 | if self._config_is_python_task(config): |
| 150 | task_object = config["class"]() |
| 151 | else: |
| 152 | config = self._process_alias(config, group=group) |
| 153 | task_object = ConfigurableTask(config=config) |
| 154 | if group is not None: |
| 155 | task_object = (group, task_object) |
| 156 | return {task: task_object} |
| 157 | |
| 158 | if isinstance(name_or_config, str): |
| 159 | if update_config is not None: |
nothing calls this directly
no test coverage detected