(log: BoundLogger, tasks: List["Task"])
| 81 | |
| 82 | |
| 83 | def get_runner_class(log: BoundLogger, tasks: List["Task"]) -> Type[BaseRunner]: |
| 84 | runner_class_paths = {task.serialized_runner_class for task in tasks} |
| 85 | if len(runner_class_paths) > 1: |
| 86 | log.error( |
| 87 | "cannot mix multiple runner classes", |
| 88 | runner_class_paths=", ".join(str(p) for p in runner_class_paths), |
| 89 | ) |
| 90 | raise ValueError("Found multiple runner classes in batch task.") |
| 91 | |
| 92 | runner_class_path = runner_class_paths.pop() |
| 93 | if runner_class_path: |
| 94 | try: |
| 95 | return import_attribute(runner_class_path) |
| 96 | except TaskImportError: |
| 97 | log.error( |
| 98 | "could not import runner class", |
| 99 | runner_class_path=runner_class_path, |
| 100 | ) |
| 101 | raise |
| 102 | return DefaultRunner |
no test coverage detected