(stages: list["Stage"])
| 92 | |
| 93 | @contextmanager |
| 94 | def translate_graph_error(stages: list["Stage"]) -> Iterator[None]: |
| 95 | try: |
| 96 | yield |
| 97 | except OverlappingOutputPathsError as exc: |
| 98 | if exc.parent in [o for s in stages for o in s.outs]: |
| 99 | msg = OVERLAPPING_PARENT_FMT.format( |
| 100 | out=exc.overlapping_out, |
| 101 | parent=exc.parent, |
| 102 | out_stage=exc.overlapping_out.stage.addressing, |
| 103 | ) |
| 104 | else: |
| 105 | msg = OVERLAPPING_CHILD_FMT.format( |
| 106 | out=exc.overlapping_out, |
| 107 | parent=exc.parent, |
| 108 | parent_stage=exc.parent.stage.addressing, |
| 109 | ) |
| 110 | raise OverlappingOutputPathsError( # noqa: B904 |
| 111 | exc.parent, exc.overlapping_out, msg |
| 112 | ) |
| 113 | except OutputDuplicationError as exc: |
| 114 | raise OutputDuplicationError( # noqa: B904 |
| 115 | exc.output, set(exc.stages) - set(stages) |
| 116 | ) |
| 117 | |
| 118 | |
| 119 | def progress_iter(stages: dict[str, StageInfo]) -> Iterator[tuple[str, StageInfo]]: |
no test coverage detected