From e81a25f27b86f835c3a6b0ac302645a84120c244 Mon Sep 17 00:00:00 2001 From: Sangyoon Park Date: Fri, 10 Apr 2026 23:25:54 -0400 Subject: [PATCH 1/3] Introduce `ErrorRecord` dataclass for structured error files - Replace `write_error_file` helper with `ErrorRecord` dataclass that owns read, write, and capture-from-exception logic - Drop the `file` key from the on-disk `.err` schema; the stem is already recoverable from the filename - Write error details into `.setup-failed` sentinel files so setup failures are inspectable beyond just the sentinel's existence --- docs/mkdocs/guides/pipeline.md | 8 ++-- docs/mkdocs/guides/task.md | 3 +- src/tigerflow/models.py | 21 ++++++---- src/tigerflow/tasks/local.py | 6 +-- src/tigerflow/tasks/local_async.py | 6 +-- src/tigerflow/tasks/slurm.py | 8 ++-- src/tigerflow/tasks/utils.py | 22 ----------- src/tigerflow/utils.py | 49 ++++++++++++++++++++++++ tests/integration/test_slurm_task.py | 8 ++++ tests/unit/test_utils.py | 57 ++++++++++++++++++++++++++++ 10 files changed, 143 insertions(+), 45 deletions(-) diff --git a/docs/mkdocs/guides/pipeline.md b/docs/mkdocs/guides/pipeline.md index 8633576..e8fb86d 100644 --- a/docs/mkdocs/guides/pipeline.md +++ b/docs/mkdocs/guides/pipeline.md @@ -389,9 +389,9 @@ and errors: │ ingest █▄▂▅▁▄▃▁▇▄▄▁▄ 100ms – 101ms (101ms avg) │ │ │ │ Errors: 3 │ - │ transcribe 0016.mp4 ClientResponseError: 400, ... │ - │ transcribe 0004.mp4 ClientResponseError: 400, ... │ - │ transcribe 0028.mp4 ClientResponseError: 400, ... │ + │ transcribe 0016 ClientResponseError: 400, ... │ + │ transcribe 0004 ClientResponseError: 400, ... │ + │ transcribe 0028 ClientResponseError: 400, ... │ │ │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ``` @@ -409,7 +409,7 @@ The dashboard displays four sections: - **Metrics** — per-task sparklines with min, max, and average processing durations (only shown when metric data is available). - **Errors** — a count and summary of the first five errors, including the task name, - input file, and exception details. + input file ID, and exception details. ### Watch Mode diff --git a/docs/mkdocs/guides/task.md b/docs/mkdocs/guides/task.md index d36560f..c618c4d 100644 --- a/docs/mkdocs/guides/task.md +++ b/docs/mkdocs/guides/task.md @@ -131,8 +131,7 @@ For example, `path/to/data/4.txt` produces `path/to/results/4.txt`. If a task encounters an error, TigerFlow generates a structured error file alongside the expected output, e.g., `4.err` instead of `4.txt`. The file is JSON containing the - input file name, timestamp, exception type, message, and full traceback to assist with - debugging. + timestamp, exception type, message, and full traceback to assist with debugging. ## Lazy Imports diff --git a/src/tigerflow/models.py b/src/tigerflow/models.py index ed3a3d2..f26d639 100644 --- a/src/tigerflow/models.py +++ b/src/tigerflow/models.py @@ -12,7 +12,12 @@ from tigerflow.settings import settings from tigerflow.staging import StagingPipeline -from tigerflow.utils import is_process_running, read_pid_file, validate_file_ext +from tigerflow.utils import ( + ErrorRecord, + is_process_running, + read_pid_file, + validate_file_ext, +) class TaskStatusKind(Enum): @@ -600,18 +605,18 @@ def report(self) -> PipelineReport: stem = file.name.removesuffix(".err") failed_stems.add(stem) try: - data = json.loads(file.read_text()) + record = ErrorRecord.read(file) task_errors.append( FileError( - file=data.get("file", stem), + file=stem, path=str(file), - timestamp=datetime.fromisoformat(data["timestamp"]), - exception_type=data.get("exception_type", ""), - message=data.get("message", ""), - traceback=data.get("traceback", ""), + timestamp=datetime.fromisoformat(record.timestamp), + exception_type=record.exception_type, + message=record.message, + traceback=record.traceback, ) ) - except (OSError, json.JSONDecodeError, KeyError): + except (OSError, ValueError): task_errors.append(FileError(file=stem, path=str(file))) if task_errors: errors[task_dir.name] = task_errors diff --git a/src/tigerflow/tasks/local.py b/src/tigerflow/tasks/local.py index 65fad6b..060eab1 100644 --- a/src/tigerflow/tasks/local.py +++ b/src/tigerflow/tasks/local.py @@ -11,10 +11,10 @@ from tigerflow.logconfig import logger from tigerflow.models import LocalTaskConfig from tigerflow.settings import settings -from tigerflow.utils import SetupContext, atomic_write +from tigerflow.utils import ErrorRecord, SetupContext, atomic_write from ._base import Task -from .utils import log_metrics, write_error_file +from .utils import log_metrics class LocalTask(Task): @@ -52,7 +52,7 @@ def task(input_file: Path, output_file: Path): output_file.name.removesuffix(self.config.output_ext) + ".err" ) error_file = output_dir / error_fname - write_error_file(error_file, input_file.name) + ErrorRecord.from_exception().write(error_file) logger.error("Failed processing: {}", input_file.name) # Clean up incomplete temporary files left behind by a prior process instance diff --git a/src/tigerflow/tasks/local_async.py b/src/tigerflow/tasks/local_async.py index a6ea68b..2075e65 100644 --- a/src/tigerflow/tasks/local_async.py +++ b/src/tigerflow/tasks/local_async.py @@ -10,10 +10,10 @@ from tigerflow.logconfig import logger from tigerflow.models import LocalAsyncTaskConfig from tigerflow.settings import settings -from tigerflow.utils import SetupContext, atomic_write +from tigerflow.utils import ErrorRecord, SetupContext, atomic_write from ._base import Task -from .utils import log_metrics, write_error_file +from .utils import log_metrics class LocalAsyncTask(Task): @@ -53,7 +53,7 @@ async def task(input_file: Path, output_file: Path): output_file.name.removesuffix(self.config.output_ext) + ".err" ) error_file = self.config.output_dir / error_fname - write_error_file(error_file, input_file.name) + ErrorRecord.from_exception().write(error_file) logger.error("Failed processing: {}", input_file.name) async def worker(): diff --git a/src/tigerflow/tasks/slurm.py b/src/tigerflow/tasks/slurm.py index ab87596..ddc05e4 100644 --- a/src/tigerflow/tasks/slurm.py +++ b/src/tigerflow/tasks/slurm.py @@ -22,13 +22,14 @@ from tigerflow.settings import settings from tigerflow.utils import ( TEMP_FILE_PREFIX, + ErrorRecord, SetupContext, atomic_write, submit_to_slurm, ) from ._base import Task -from .utils import get_slurm_task_status, log_metrics, write_error_file +from .utils import get_slurm_task_status, log_metrics class SlurmTask(Task): @@ -79,7 +80,8 @@ async def setup(self, worker: Worker): logger.info("Task setup complete") except Exception: logger.exception("Task setup failed; aborting task") - setup_failed_sentinel.touch() + setup_failed_sentinel.touch() # ensure sentinel exists even if write below fails + ErrorRecord.from_exception().write(setup_failed_sentinel) async def teardown(self, worker: Worker): logger.info("Tearing down task") @@ -109,7 +111,7 @@ def task(input_file: Path, output_file: Path): metrics["status"] = "error" error_fname = output_file.name.removesuffix(output_ext) + ".err" error_file = output_dir / error_fname - write_error_file(error_file, input_file.name) + ErrorRecord.from_exception().write(error_file) logger.error("Failed processing: {}", input_file.name) # Define parameters for each Slurm job diff --git a/src/tigerflow/tasks/utils.py b/src/tigerflow/tasks/utils.py index 3fdfaf9..400434b 100644 --- a/src/tigerflow/tasks/utils.py +++ b/src/tigerflow/tasks/utils.py @@ -1,14 +1,10 @@ import json import subprocess -import sys -import traceback from contextlib import contextmanager from datetime import datetime, timezone -from pathlib import Path from tigerflow.logconfig import logger from tigerflow.models import TaskStatus, TaskStatusKind -from tigerflow.utils import atomic_write @contextmanager @@ -101,21 +97,3 @@ def get_slurm_task_status(client_job_id: int, worker_job_name: str) -> TaskStatu kind=TaskStatusKind.INACTIVE, detail=f"Reason: {reason.splitlines()[0].strip()}" if reason else None, ) - - -def write_error_file(error_path: Path, input_file: str) -> None: - """Write structured error JSON for a failed file. - - Must be called from within an exception handler. - """ - exc_type, exc_value, _ = sys.exc_info() - error_data = { - "file": input_file, - "timestamp": datetime.now(timezone.utc).isoformat(), - "exception_type": exc_type.__name__ if exc_type else "Unknown", - "message": str(exc_value) if exc_value else "", - "traceback": traceback.format_exc(), - } - with atomic_write(error_path) as temp_path: - with open(temp_path, "w") as f: - json.dump(error_data, f, indent=2) diff --git a/src/tigerflow/utils.py b/src/tigerflow/utils.py index 1f124d4..cac9111 100644 --- a/src/tigerflow/utils.py +++ b/src/tigerflow/utils.py @@ -1,11 +1,15 @@ +import dataclasses import importlib +import json import os import re import subprocess import sys import tempfile +import traceback from collections.abc import Callable from contextlib import contextmanager +from datetime import datetime, timezone from pathlib import Path from subprocess import TimeoutExpired from types import SimpleNamespace @@ -226,3 +230,48 @@ def atomic_write(filepath: str | os.PathLike[str]): raise else: temp_path.replace(filepath) + + +@dataclasses.dataclass(slots=True) +class ErrorRecord: + """Structured error record for JSON serialization. + + Represents the on-disk schema used by .err and .setup-failed files. + """ + + timestamp: str + exception_type: str + message: str + traceback: str + + @classmethod + def from_exception(cls) -> "ErrorRecord": + """Capture error details from the current exception context. + + Must be called from within an exception handler. + """ + exc_type, exc_value, _ = sys.exc_info() + return cls( + timestamp=datetime.now(timezone.utc).isoformat(), + exception_type=exc_type.__name__ if exc_type else "Unknown", + message=str(exc_value) if exc_value else "", + traceback=traceback.format_exc(), + ) + + def write(self, path: Path) -> None: + """Write error record as JSON to *path* using atomic write.""" + with atomic_write(path) as temp_path: + with open(temp_path, "w") as f: + json.dump(dataclasses.asdict(self), f, indent=2) + + @classmethod + def read(cls, path: Path) -> "ErrorRecord": + """Read error record from a JSON file. + + Raises `ValueError` if the file content is malformed or incomplete. + """ + try: + data = json.loads(path.read_text()) + return cls(**data) + except (json.JSONDecodeError, TypeError) as exc: + raise ValueError(f"invalid error record: {path}") from exc diff --git a/tests/integration/test_slurm_task.py b/tests/integration/test_slurm_task.py index 6a9d5b2..079d9df 100644 --- a/tests/integration/test_slurm_task.py +++ b/tests/integration/test_slurm_task.py @@ -9,6 +9,7 @@ Set SLURM_TEST_DIR to a shared filesystem path accessible by compute nodes. """ +import json import os import shutil import signal @@ -270,6 +271,13 @@ def test_setup_failure_aborts(self, task_dirs, input_files, tasks_dir): sentinel_files = list(log_base.rglob(".setup-failed")) assert len(sentinel_files) > 0, "Expected .setup-failed sentinel file" + # Sentinel should contain expected error details + error_data = json.loads(sentinel_files[0].read_text()) + assert error_data["exception_type"] == "RuntimeError" + assert "Intentional setup failure" in error_data["message"] + assert "timestamp" in error_data + assert "traceback" in error_data + # No output files should have been produced output_files = [ f diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 76ad121..98d7ead 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -1,3 +1,4 @@ +import json import os import textwrap from pathlib import Path @@ -6,6 +7,7 @@ from tigerflow.utils import ( TEMP_FILE_PREFIX, + ErrorRecord, atomic_write, has_running_pid, import_callable, @@ -205,3 +207,58 @@ def test_accepts_str_path(self, tmp_path: Path): assert temp.suffix == ".json" temp.write_text("{}") assert target.read_text() == "{}" + + +class TestErrorRecord: + def test_from_exception_captures_fields(self): + try: + raise ValueError("test error") + except ValueError: + record = ErrorRecord.from_exception() + + assert record.exception_type == "ValueError" + assert record.message == "test error" + assert "ValueError: test error" in record.traceback + assert record.timestamp # non-empty ISO string + + def test_from_exception_outside_handler(self): + record = ErrorRecord.from_exception() + assert record.exception_type == "Unknown" + assert record.message == "" + + def test_write_read_roundtrip(self, tmp_path: Path): + original = ErrorRecord( + timestamp="2026-01-01T00:00:00+00:00", + exception_type="RuntimeError", + message="boom", + traceback="Traceback ...", + ) + path = tmp_path / "error.err" + original.write(path) + loaded = ErrorRecord.read(path) + assert loaded == original + + def test_read_extra_keys_raises_value_error(self, tmp_path: Path): + path = tmp_path / "error.err" + data = { + "timestamp": "2026-01-01T00:00:00+00:00", + "exception_type": "RuntimeError", + "message": "boom", + "traceback": "Traceback ...", + "file": "unexpected.txt", + } + path.write_text(json.dumps(data)) + with pytest.raises(ValueError, match="invalid error record"): + ErrorRecord.read(path) + + def test_read_missing_keys_raises_value_error(self, tmp_path: Path): + path = tmp_path / "error.err" + path.write_text(json.dumps({"timestamp": "2026-01-01T00:00:00+00:00"})) + with pytest.raises(ValueError, match="invalid error record"): + ErrorRecord.read(path) + + def test_read_malformed_json_raises_value_error(self, tmp_path: Path): + path = tmp_path / "error.err" + path.write_text("not json") + with pytest.raises(ValueError, match="invalid error record"): + ErrorRecord.read(path) From ee08cc9b6827670c6b38ac97462eb958fb446ba0 Mon Sep 17 00:00:00 2001 From: Sangyoon Park Date: Mon, 13 Apr 2026 16:29:42 -0400 Subject: [PATCH 2/3] Track input filename on `ErrorRecord` for display attribution The `.err` stem is derived from the output filename, which does not always recover the original input filename. Store it on the record so `PipelineOutput` can display the input the user recognizes --- docs/mkdocs/guides/pipeline.md | 8 +++---- docs/mkdocs/guides/task.md | 3 ++- src/tigerflow/models.py | 2 +- src/tigerflow/tasks/local.py | 2 +- src/tigerflow/tasks/local_async.py | 2 +- src/tigerflow/tasks/slurm.py | 2 +- src/tigerflow/utils.py | 8 +++++-- tests/unit/test_utils.py | 37 +++++++++++++++++++++++++++++- 8 files changed, 52 insertions(+), 12 deletions(-) diff --git a/docs/mkdocs/guides/pipeline.md b/docs/mkdocs/guides/pipeline.md index e8fb86d..8633576 100644 --- a/docs/mkdocs/guides/pipeline.md +++ b/docs/mkdocs/guides/pipeline.md @@ -389,9 +389,9 @@ and errors: │ ingest █▄▂▅▁▄▃▁▇▄▄▁▄ 100ms – 101ms (101ms avg) │ │ │ │ Errors: 3 │ - │ transcribe 0016 ClientResponseError: 400, ... │ - │ transcribe 0004 ClientResponseError: 400, ... │ - │ transcribe 0028 ClientResponseError: 400, ... │ + │ transcribe 0016.mp4 ClientResponseError: 400, ... │ + │ transcribe 0004.mp4 ClientResponseError: 400, ... │ + │ transcribe 0028.mp4 ClientResponseError: 400, ... │ │ │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ``` @@ -409,7 +409,7 @@ The dashboard displays four sections: - **Metrics** — per-task sparklines with min, max, and average processing durations (only shown when metric data is available). - **Errors** — a count and summary of the first five errors, including the task name, - input file ID, and exception details. + input file, and exception details. ### Watch Mode diff --git a/docs/mkdocs/guides/task.md b/docs/mkdocs/guides/task.md index c618c4d..d36560f 100644 --- a/docs/mkdocs/guides/task.md +++ b/docs/mkdocs/guides/task.md @@ -131,7 +131,8 @@ For example, `path/to/data/4.txt` produces `path/to/results/4.txt`. If a task encounters an error, TigerFlow generates a structured error file alongside the expected output, e.g., `4.err` instead of `4.txt`. The file is JSON containing the - timestamp, exception type, message, and full traceback to assist with debugging. + input file name, timestamp, exception type, message, and full traceback to assist with + debugging. ## Lazy Imports diff --git a/src/tigerflow/models.py b/src/tigerflow/models.py index f26d639..037d878 100644 --- a/src/tigerflow/models.py +++ b/src/tigerflow/models.py @@ -608,7 +608,7 @@ def report(self) -> PipelineReport: record = ErrorRecord.read(file) task_errors.append( FileError( - file=stem, + file=record.file or stem, path=str(file), timestamp=datetime.fromisoformat(record.timestamp), exception_type=record.exception_type, diff --git a/src/tigerflow/tasks/local.py b/src/tigerflow/tasks/local.py index 060eab1..fe1d18e 100644 --- a/src/tigerflow/tasks/local.py +++ b/src/tigerflow/tasks/local.py @@ -52,7 +52,7 @@ def task(input_file: Path, output_file: Path): output_file.name.removesuffix(self.config.output_ext) + ".err" ) error_file = output_dir / error_fname - ErrorRecord.from_exception().write(error_file) + ErrorRecord.from_exception(file=input_file.name).write(error_file) logger.error("Failed processing: {}", input_file.name) # Clean up incomplete temporary files left behind by a prior process instance diff --git a/src/tigerflow/tasks/local_async.py b/src/tigerflow/tasks/local_async.py index 2075e65..e81eba3 100644 --- a/src/tigerflow/tasks/local_async.py +++ b/src/tigerflow/tasks/local_async.py @@ -53,7 +53,7 @@ async def task(input_file: Path, output_file: Path): output_file.name.removesuffix(self.config.output_ext) + ".err" ) error_file = self.config.output_dir / error_fname - ErrorRecord.from_exception().write(error_file) + ErrorRecord.from_exception(file=input_file.name).write(error_file) logger.error("Failed processing: {}", input_file.name) async def worker(): diff --git a/src/tigerflow/tasks/slurm.py b/src/tigerflow/tasks/slurm.py index ddc05e4..ec5bfc9 100644 --- a/src/tigerflow/tasks/slurm.py +++ b/src/tigerflow/tasks/slurm.py @@ -111,7 +111,7 @@ def task(input_file: Path, output_file: Path): metrics["status"] = "error" error_fname = output_file.name.removesuffix(output_ext) + ".err" error_file = output_dir / error_fname - ErrorRecord.from_exception().write(error_file) + ErrorRecord.from_exception(file=input_file.name).write(error_file) logger.error("Failed processing: {}", input_file.name) # Define parameters for each Slurm job diff --git a/src/tigerflow/utils.py b/src/tigerflow/utils.py index cac9111..a665400 100644 --- a/src/tigerflow/utils.py +++ b/src/tigerflow/utils.py @@ -243,12 +243,15 @@ class ErrorRecord: exception_type: str message: str traceback: str + file: str | None = None @classmethod - def from_exception(cls) -> "ErrorRecord": + def from_exception(cls, file: str | None = None) -> "ErrorRecord": """Capture error details from the current exception context. - Must be called from within an exception handler. + Must be called from within an exception handler. *file* is the + name of the input file being processed, if any; omit for errors + not associated with a specific file (e.g. task setup failures). """ exc_type, exc_value, _ = sys.exc_info() return cls( @@ -256,6 +259,7 @@ def from_exception(cls) -> "ErrorRecord": exception_type=exc_type.__name__ if exc_type else "Unknown", message=str(exc_value) if exc_value else "", traceback=traceback.format_exc(), + file=file, ) def write(self, path: Path) -> None: diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index 98d7ead..b80b90d 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -225,6 +225,14 @@ def test_from_exception_outside_handler(self): record = ErrorRecord.from_exception() assert record.exception_type == "Unknown" assert record.message == "" + assert record.file is None + + def test_from_exception_captures_file(self): + try: + raise ValueError("test error") + except ValueError: + record = ErrorRecord.from_exception(file="input.txt") + assert record.file == "input.txt" def test_write_read_roundtrip(self, tmp_path: Path): original = ErrorRecord( @@ -232,12 +240,26 @@ def test_write_read_roundtrip(self, tmp_path: Path): exception_type="RuntimeError", message="boom", traceback="Traceback ...", + file="input.txt", ) path = tmp_path / "error.err" original.write(path) loaded = ErrorRecord.read(path) assert loaded == original + def test_write_read_roundtrip_without_file(self, tmp_path: Path): + original = ErrorRecord( + timestamp="2026-01-01T00:00:00+00:00", + exception_type="RuntimeError", + message="boom", + traceback="Traceback ...", + ) + path = tmp_path / "error.err" + original.write(path) + loaded = ErrorRecord.read(path) + assert loaded == original + assert loaded.file is None + def test_read_extra_keys_raises_value_error(self, tmp_path: Path): path = tmp_path / "error.err" data = { @@ -245,12 +267,25 @@ def test_read_extra_keys_raises_value_error(self, tmp_path: Path): "exception_type": "RuntimeError", "message": "boom", "traceback": "Traceback ...", - "file": "unexpected.txt", + "file": "input.txt", + "unexpected": "value", } path.write_text(json.dumps(data)) with pytest.raises(ValueError, match="invalid error record"): ErrorRecord.read(path) + def test_read_without_file_key(self, tmp_path: Path): + path = tmp_path / "error.err" + data = { + "timestamp": "2026-01-01T00:00:00+00:00", + "exception_type": "RuntimeError", + "message": "boom", + "traceback": "Traceback ...", + } + path.write_text(json.dumps(data)) + loaded = ErrorRecord.read(path) + assert loaded.file is None + def test_read_missing_keys_raises_value_error(self, tmp_path: Path): path = tmp_path / "error.err" path.write_text(json.dumps({"timestamp": "2026-01-01T00:00:00+00:00"})) From c9294a4eb9a4504410447c76ac3d4d56525f3a0d Mon Sep 17 00:00:00 2001 From: Sangyoon Park Date: Thu, 20 Aug 2026 17:23:37 -0400 Subject: [PATCH 3/3] Wrap ErrorRecord in FileError instead of copying fields `FileError` duplicated every `ErrorRecord` field, so reading an .err file meant unpacking the record and rebuilding it field by field. It is now a slotted dataclass holding the record plus `path`, the file's location. Timestamps stay raw strings instead of being parsed, so a malformed one no longer discards the whole record. --- src/tigerflow/cli/report.py | 26 +++++++++++++------------- src/tigerflow/models.py | 36 ++++++++++++++++++------------------ 2 files changed, 31 insertions(+), 31 deletions(-) diff --git a/src/tigerflow/cli/report.py b/src/tigerflow/cli/report.py index 10e07cf..4aba0c8 100644 --- a/src/tigerflow/cli/report.py +++ b/src/tigerflow/cli/report.py @@ -168,12 +168,12 @@ def fmt_duration(ms: float) -> str: if shown >= 5: break err_detail = ( - f"{err.exception_type}: {err.message}" - if err.exception_type - else err.message or "Unknown error" + f"{err.record.exception_type}: {err.record.message}" + if err.record.exception_type + else err.record.message or "Unknown error" ) lines.append( - f" [dim]{task_name}[/dim] {err.file} [red]{err_detail}[/red]" + f" [dim]{task_name}[/dim] {err.record.file} [red]{err_detail}[/red]" ) shown += 1 if shown >= 5: @@ -322,18 +322,18 @@ def report( } if "errors" in sections: result["errors"] = { - name: [ + task: [ { - "file": e.file, - "path": e.path, - "timestamp": e.timestamp.isoformat() if e.timestamp else None, - "exception_type": e.exception_type, - "message": e.message, - "traceback": e.traceback, + "file": err.record.file, + "path": err.path, + "timestamp": err.record.timestamp or None, + "exception_type": err.record.exception_type, + "message": err.record.message, + "traceback": err.record.traceback, } - for e in errs + for err in errors ] - for name, errs in pipeline_report.errors.items() + for task, errors in pipeline_report.errors.items() } print(json.dumps(result, indent=2, default=str)) diff --git a/src/tigerflow/models.py b/src/tigerflow/models.py index f6e849d..029bedb 100644 --- a/src/tigerflow/models.py +++ b/src/tigerflow/models.py @@ -2,6 +2,7 @@ import re import shlex import textwrap +from dataclasses import dataclass from datetime import datetime from enum import Enum from pathlib import Path @@ -462,15 +463,16 @@ class TaskProgress(BaseModel): failed: int = 0 -class FileError(BaseModel): - """Error information for a failed file.""" +@dataclass(slots=True) +class FileError: + """A failed file's error record paired with where it was found. + + `path` is the location of the .err file itself, which the record does + not carry: it is known only to the reader that discovers the file. + """ - file: str path: str - timestamp: datetime | None = None - exception_type: str = "" - message: str = "" - traceback: str = "" + record: ErrorRecord class TaskMeta(BaseModel): @@ -624,18 +626,16 @@ def report(self) -> PipelineReport: failed_stems.add(stem) try: record = ErrorRecord.read(file) - task_errors.append( - FileError( - file=record.file or stem, - path=str(file), - timestamp=datetime.fromisoformat(record.timestamp), - exception_type=record.exception_type, - message=record.message, - traceback=record.traceback, - ) - ) + record.file = record.file or stem except (OSError, ValueError): - task_errors.append(FileError(file=stem, path=str(file))) + record = ErrorRecord( + timestamp="", + exception_type="", + message="", + traceback="", + file=stem, + ) + task_errors.append(FileError(path=str(file), record=record)) if task_errors: errors[task_dir.name] = task_errors