chore(exporter): add ErrorReporter

Author: Michał Sośnicki

src/neptune_exporter/export_manager.py

@@ -20,6 +20,7 @@
 from tqdm import tqdm
 import pyarrow as pa
 import pyarrow.compute as pc
+from neptune_exporter.exporters.error_reporter import ErrorReporter
 from neptune_exporter.exporters.exporter import NeptuneExporter
 from neptune_exporter.storage.parquet_writer import ParquetWriter, RunWriterContext
 from neptune_exporter.storage.parquet_reader import ParquetReader
@@ -33,13 +34,15 @@ def __init__(
         exporter: NeptuneExporter,
         reader: ParquetReader,
         writer: ParquetWriter,
+        error_reporter: ErrorReporter,
         files_destination: Path,
         batch_size: int = 16,
         progress_bar: bool = True,
     ):
         self._exporter = exporter
         self._reader = reader
         self._writer = writer
+        self._error_reporter = error_reporter
         self._files_destination = files_destination
         self._batch_size = batch_size
         self._progress_bar = progress_bar
@@ -185,10 +188,10 @@ def run(
                     # Update progress bar for completed batch
                     runs_pbar.update(len(batch_run_ids))
 
-        exception_infos = self._exporter.get_exception_infos()
-        if exception_infos:
+        exception_summary = self._error_reporter.get_summary()
+        if exception_summary.exception_count > 0:
             self._logger.error(
-                f"{len(exception_infos)} exceptions occurred during export. See the logs for details."
+                f"{exception_summary.exception_count} exceptions occurred during export. See the logs for details."
             )
 
         return total_runs

src/neptune_exporter/exporters/error_reporter.py

@@ -0,0 +1,110 @@
+#
+# Copyright (c) 2025, Neptune Labs Sp. z o.o.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from dataclasses import dataclass
+import json
+from pathlib import Path
+import threading
+from typing import Optional
+
+from neptune_exporter.types import ProjectId, SourceRunId
+
+
+@dataclass
+class ErrorSummary:
+    exception_count: int
+
+
+class ErrorReporter:
+    def __init__(
+        self,
+        path: Path,
+    ) -> None:
+        self.path = path
+        self._lock = threading.Lock()
+        self._summary = ErrorSummary(exception_count=0)
+
+    def get_summary(self) -> ErrorSummary:
+        with self._lock:
+            return ErrorSummary(exception_count=self._summary.exception_count)
+
+    def record_exception(
+        self,
+        project_id: ProjectId,
+        run_id: SourceRunId,
+        attribute_path: Optional[str],
+        attribute_type: Optional[str],
+        exception: Exception,
+    ) -> None:
+        with self._lock:
+            self._summary.exception_count += 1
+            self._write_record(
+                project_id=project_id,
+                run_id=run_id,
+                attribute_path=attribute_path,
+                attribute_type=attribute_type,
+                exception=exception,
+            )
+
+    def record_batch_exception(
+        self,
+        project_id: ProjectId,
+        run_ids: list[SourceRunId],
+        attribute_paths: Optional[list[str]],
+        exception: Exception,
+    ) -> None:
+        with self._lock:
+            for run_id in run_ids:
+                if attribute_paths is None:
+                    self._summary.exception_count += 1
+                    self._write_record(
+                        project_id=project_id,
+                        run_id=run_id,
+                        attribute_path=None,
+                        attribute_type=None,
+                        exception=exception,
+                    )
+                else:
+                    for attribute_path in attribute_paths:
+                        self._summary.exception_count += 1
+                        self._write_record(
+                            project_id=project_id,
+                            run_id=run_id,
+                            attribute_path=attribute_path,
+                            attribute_type=None,
+                            exception=exception,
+                        )
+
+    def _write_record(
+        self,
+        project_id: ProjectId,
+        run_id: SourceRunId,
+        attribute_path: Optional[str],
+        attribute_type: Optional[str],
+        exception: Exception,
+    ) -> None:
+        with open(self.path, "a") as f:
+            f.write(
+                json.dumps(
+                    {
+                        "project_id": project_id,
+                        "run_id": run_id,
+                        "attribute_path": attribute_path,
+                        "attribute_type": attribute_type,
+                        "exception": exception.__class__.__name__,
+                    }
+                )
+                + "\n"
+            )

src/neptune_exporter/exporters/exporter.py

@@ -16,23 +16,13 @@
 """Core exporter abstract base class and types."""
 
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
 from typing import Generator, Optional, Sequence
 from pathlib import Path
 import pyarrow as pa
 
 from neptune_exporter.types import ProjectId, SourceRunId
 
 
-@dataclass
-class ExceptionInfo:
-    project_id: ProjectId
-    run_id: SourceRunId
-    attribute_path: Optional[str]
-    attribute_type: Optional[str]
-    exception: Exception
-
-
 class NeptuneExporter(ABC):
     """Abstract base class for Neptune data exporters."""
 
@@ -89,11 +79,6 @@ def download_files(
         """Download files from Neptune runs."""
         pass
 
-    @abstractmethod
-    def get_exception_infos(self) -> list[ExceptionInfo]:
-        """Get list of exceptions that occurred during export."""
-        pass
-
     def close(self) -> None:
         """Close the exporter."""
         pass

src/neptune_exporter/exporters/neptune2.py

@@ -32,7 +32,8 @@
 from neptune import management
 
 from neptune_exporter import model
-from neptune_exporter.exporters.exporter import ExceptionInfo, NeptuneExporter
+from neptune_exporter.exporters.exporter import NeptuneExporter
+from neptune_exporter.exporters.error_reporter import ErrorReporter
 from neptune_exporter.types import ProjectId, SourceRunId
 
 _ATTRIBUTE_TYPE_MAP = {
@@ -72,19 +73,20 @@
 class Neptune2Exporter(NeptuneExporter):
     def __init__(
         self,
+        error_reporter: ErrorReporter,
         api_token: Optional[str] = None,
         max_workers: int = 16,
         show_client_logs: bool = False,
         include_trashed_runs: bool = False,
     ):
+        self._error_reporter = error_reporter
         self._include_trashed_runs = include_trashed_runs
         self._api_token = api_token
         self._quantize_base = Decimal("1.000000")
         self._executor = ThreadPoolExecutor(max_workers=max_workers)
         self._logger = logging.getLogger(__name__)
         self._show_client_logs = show_client_logs
         self._initialize_client(show_client_logs=show_client_logs)
-        self._exception_infos: list[ExceptionInfo] = []
 
     def _initialize_client(self, show_client_logs: bool) -> None:
         if show_client_logs:
@@ -696,28 +698,6 @@ def _should_include_attribute(
 
         return True
 
-    def get_exception_infos(self) -> list[ExceptionInfo]:
-        """Get list of exceptions that occurred during export."""
-        return self._exception_infos
-
-    def _record_exception(
-        self,
-        project_id: ProjectId,
-        run_id: SourceRunId,
-        attribute_path: Optional[str],
-        attribute_type: Optional[str],
-        exception: Exception,
-    ) -> None:
-        self._exception_infos.append(
-            ExceptionInfo(
-                project_id=project_id,
-                run_id=run_id,
-                attribute_path=attribute_path,
-                attribute_type=attribute_type,
-                exception=exception,
-            )
-        )
-
     def _handle_run_exception(
         self, project_id: ProjectId, run_id: SourceRunId, exception: Exception
     ) -> None:
@@ -740,7 +720,7 @@ def _handle_run_exception(
                 f"Skipping project {project_id}, run {run_id} because of unexpected error.",
                 exc_info=True,
             )
-        self._record_exception(
+        self._error_reporter.record_exception(
             project_id=project_id,
             run_id=run_id,
             attribute_path=None,
@@ -786,7 +766,7 @@ def _handle_attribute_exception(
                 f"Skipping project {project_id}, run {run_id}, attribute {attribute_path} ({attribute_type}) because of unexpected error.",
                 exc_info=True,
             )
-        self._record_exception(
+        self._error_reporter.record_exception(
             project_id=project_id,
             run_id=run_id,
             attribute_path=attribute_path,

src/neptune_exporter/exporters/neptune3.py

@@ -27,7 +27,8 @@
 import logging
 
 from neptune_exporter import model
-from neptune_exporter.exporters.exporter import ExceptionInfo, NeptuneExporter
+from neptune_exporter.exporters.exporter import NeptuneExporter
+from neptune_exporter.exporters.error_reporter import ErrorReporter
 from neptune_exporter.types import ProjectId, SourceRunId
 
 
@@ -51,6 +52,7 @@
 class Neptune3Exporter(NeptuneExporter):
     def __init__(
         self,
+        error_reporter: ErrorReporter,
         api_token: Optional[str] = None,
         series_attribute_batch_size: int = 128,
         file_attribute_batch_size: int = 16,
@@ -67,7 +69,7 @@ def __init__(
         self._executor = ThreadPoolExecutor(max_workers=max_workers)
         self._logger = logging.getLogger(__name__)
         self._initialize_client(api_token=api_token, show_client_logs=show_client_logs)
-        self._exception_infos: list[ExceptionInfo] = []
+        self._error_reporter = error_reporter
 
     def _initialize_client(
         self, api_token: Optional[str], show_client_logs: bool
@@ -693,40 +695,6 @@ def _convert_files_to_schema(
             }
         )
 
-    def get_exception_infos(self) -> list[ExceptionInfo]:
-        """Get list of exceptions that occurred during export."""
-        return self._exception_infos
-
-    def _record_exception(
-        self,
-        project_id: ProjectId,
-        run_ids: list[SourceRunId],
-        attribute_paths: Optional[list[str]],
-        exception: Exception,
-    ) -> None:
-        for run_id in run_ids:
-            if attribute_paths is None:
-                self._exception_infos.append(
-                    ExceptionInfo(
-                        project_id=project_id,
-                        run_id=run_id,
-                        attribute_path=None,
-                        attribute_type=None,
-                        exception=exception,
-                    )
-                )
-            else:
-                for attribute_path in attribute_paths:
-                    self._exception_infos.append(
-                        ExceptionInfo(
-                            project_id=project_id,
-                            run_id=run_id,
-                            attribute_path=attribute_path,
-                            attribute_type=None,
-                            exception=exception,
-                        )
-                    )
-
     def _handle_batch_exception(
         self,
         project_id: ProjectId,
@@ -755,7 +723,7 @@ def _handle_batch_exception(
                 exc_info=True,
             )
 
-        self._record_exception(
+        self._error_reporter.record_batch_exception(
             project_id=project_id,
             run_ids=run_ids,
             attribute_paths=attribute_batch,
@@ -789,7 +757,7 @@ def _handle_runs_exception(
                 exc_info=True,
             )
 
-        self._record_exception(
+        self._error_reporter.record_batch_exception(
             project_id=project_id,
             run_ids=run_ids,
             attribute_paths=None,