(hook_objs, phase: str, *phase_args)
| 189 | |
| 190 | |
| 191 | def run_hooks(hook_objs, phase: str, *phase_args): |
| 192 | if phase not in ('post_hook', 'on_error', 'finalize'): |
| 193 | raise ValueError(f'Invalid hook phase: {phase}') |
| 194 | strict_errors = [] |
| 195 | ordered_hooks = hook_objs[::-1] |
| 196 | for hook_obj in ordered_hooks: |
| 197 | try: |
| 198 | getattr(hook_obj, phase)(*phase_args) |
| 199 | except Exception as e: |
| 200 | if _hook_error_mode(hook_obj) == 'raise': |
| 201 | strict_errors.append((hook_obj, e)) |
| 202 | else: |
| 203 | LOG.warning(f'Hook `{type(hook_obj).__name__}` {phase} failed and will be skipped: {e}') |
| 204 | _raise_hook_phase_errors(phase, strict_errors) |
| 205 | |
| 206 | |
| 207 | @contextmanager |
no test coverage detected