(
self,
run: FlowRun,
*,
status: str,
message: str,
details: dict[str, Any] | None = None,
level: int | None = None,
)
| 121 | return run |
| 122 | |
| 123 | def finish_run( |
| 124 | self, |
| 125 | run: FlowRun, |
| 126 | *, |
| 127 | status: str, |
| 128 | message: str, |
| 129 | details: dict[str, Any] | None = None, |
| 130 | level: int | None = None, |
| 131 | ) -> None: |
| 132 | resolved_level = level if level is not None else (logging.ERROR if status == "failed" else logging.INFO) |
| 133 | self.log_event( |
| 134 | run, |
| 135 | stage="run", |
| 136 | status=status, |
| 137 | message=message, |
| 138 | details=details, |
| 139 | level=resolved_level, |
| 140 | ) |
| 141 | |
| 142 | def log_event( |
| 143 | self, |
no test coverage detected