From 9a57c72205b2bdd4f4d17e15273c9399fe29fda2 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 25 Mar 2026 15:36:21 +0000 Subject: [PATCH 1/2] Raise errors instead of silently dropping documents during dataset download Four locations were silently swallowing exceptions (except: pass) when documents failed to convert or be added to the dataset database. This caused downloaded datasets to have fewer documents than expected with no indication of what went wrong. Changes: - jsons2documents(): warn with document IDs and error details for each conversion failure - ndi_dataset_dir.__init__(): raise RuntimeError when documents fail to add to the database (FileExistsError for duplicates is still silently skipped) - downloadDataset(): raise RuntimeError at the end if any documents were lost during conversion or database insertion - _sync_download_new(): raise RuntimeError when sync loses documents - add_ingested_session(): raise RuntimeError instead of DEBUG-only logging https://claude.ai/code/session_015oe6ucKwCFznpu1PN1nAfz --- src/ndi/cloud/download.py | 20 ++++++++++-- src/ndi/cloud/orchestration.py | 60 +++++++++++++++++++++++++++++++--- src/ndi/dataset/_dataset.py | 33 +++++++++++++++++-- 3 files changed, 104 insertions(+), 9 deletions(-) diff --git a/src/ndi/cloud/download.py b/src/ndi/cloud/download.py index da0f382..fba2321 100644 --- a/src/ndi/cloud/download.py +++ b/src/ndi/cloud/download.py @@ -8,6 +8,7 @@ import json import logging +import warnings from collections.abc import Callable from pathlib import Path from typing import TYPE_CHECKING, Any @@ -255,15 +256,30 @@ def jsons2documents( """Convert a list of raw JSON dicts into ndi.ndi_document objects. MATLAB equivalent: downloadDataset.m conversion step. + + Raises a warning if any documents fail to convert, listing the + document IDs and error messages for each failure. """ from ndi.document import ndi_document documents = [] + failures: list[tuple[str, str]] = [] for dj in doc_jsons: try: documents.append(ndi_document(dj)) - except Exception: - pass + except Exception as exc: + doc_id = dj.get("base", {}).get("id", "") if isinstance(dj, dict) else "" + failures.append((doc_id, str(exc))) + if failures: + failure_details = "\n".join( + f" - {doc_id}: {err}" for doc_id, err in failures[:20] + ) + extra = f"\n ... and {len(failures) - 20} more" if len(failures) > 20 else "" + warnings.warn( + f"Failed to convert {len(failures)} of {len(doc_jsons)} " + f"documents from JSON to ndi_document:\n{failure_details}{extra}", + stacklevel=2, + ) return documents diff --git a/src/ndi/cloud/orchestration.py b/src/ndi/cloud/orchestration.py index 3b681ce..02409c1 100644 --- a/src/ndi/cloud/orchestration.py +++ b/src/ndi/cloud/orchestration.py @@ -10,6 +10,7 @@ from __future__ import annotations +import warnings from pathlib import Path from typing import TYPE_CHECKING, Any @@ -83,6 +84,7 @@ def downloadDataset( from ndi.dataset import ndi_dataset_dir documents = jsons2documents(doc_jsons) + conversion_lost = len(doc_jsons) - len(documents) dataset = ndi_dataset_dir("", target, documents=documents) # Create remote link document if not already present @@ -93,8 +95,13 @@ def downloadDataset( remote_doc = createRemoteDatasetDoc(cloud_dataset_id, dataset) try: dataset._session._database.add(remote_doc) - except Exception: - pass + except FileExistsError: + pass # Already exists, safe to skip + except Exception as exc: + warnings.warn( + f"Failed to add remote dataset link document: {exc}", + stacklevel=2, + ) # Store cloud client for on-demand file fetching dataset.cloud_client = client @@ -106,9 +113,34 @@ def downloadDataset( if verbose: print(f' Files downloaded: {report["downloaded"]}, failed: {report["failed"]}') + # Check how many documents actually made it into the dataset + from ndi.query import ndi_query as _ndi_query + + db_docs = dataset.database_search(_ndi_query("").isa("base")) + db_count = len(db_docs) + db_lost = len(documents) - db_count + total_lost = conversion_lost + db_lost + if verbose: print("Download complete.") + if total_lost > 0: + parts = [] + if conversion_lost > 0: + parts.append( + f"{conversion_lost} failed to convert from JSON to ndi_document" + ) + if db_lost > 0: + parts.append( + f"{db_lost} failed to add to the dataset database" + ) + raise RuntimeError( + f"Downloaded {len(doc_jsons)} documents but only {db_count} " + f"were added to the dataset. {total_lost} documents lost: " + + "; ".join(parts) + + ". See preceding warnings for details on each failed document." + ) + return dataset @@ -425,12 +457,32 @@ def _sync_download_new( documents = jsons2documents(new_docs) added = 0 + failures: list[tuple[str, str]] = [] for doc in documents: try: dataset.session.database_add(doc) added += 1 - except Exception: - pass + except Exception as exc: + doc_id = getattr(doc, "id", None) or "" + failures.append((str(doc_id), str(exc))) + conversion_lost = len(new_docs) - len(documents) + total_lost = conversion_lost + len(failures) + if total_lost > 0: + failure_details = "\n".join( + f" - {doc_id}: {err}" for doc_id, err in failures[:20] + ) + extra = f"\n ... and {len(failures) - 20} more" if len(failures) > 20 else "" + parts = [] + if conversion_lost > 0: + parts.append(f"{conversion_lost} failed JSON-to-document conversion") + if failures: + parts.append( + f"{len(failures)} failed to add to database:\n{failure_details}{extra}" + ) + raise RuntimeError( + f"Sync downloaded {len(new_docs)} documents but only {added} " + f"were added. {total_lost} documents lost: " + "; ".join(parts) + ) return {"downloaded": added} diff --git a/src/ndi/dataset/_dataset.py b/src/ndi/dataset/_dataset.py index e1dfb0c..4aa7a9e 100644 --- a/src/ndi/dataset/_dataset.py +++ b/src/ndi/dataset/_dataset.py @@ -163,12 +163,26 @@ def add_ingested_session(self, session: Any) -> ndi_dataset: # their *original* session_id so we can tell which session they came from. # Binary files are also copied from the source session. all_docs = session.database_search(ndi_query("").isa("base")) + ingestion_failures: list[tuple[str, str]] = [] for doc in all_docs: try: self._session._database.add(doc) self._copy_binary_files(session, doc) + except FileExistsError: + pass # Duplicates are expected and safe to skip except Exception as exc: - logger.debug("Skipping document %s during ingestion: %s", doc.id, exc) + doc_id = getattr(doc, "id", "") + ingestion_failures.append((str(doc_id), str(exc))) + logger.debug("Skipping document %s during ingestion: %s", doc_id, exc) + if ingestion_failures: + failure_details = "\n".join( + f" - {doc_id}: {err}" for doc_id, err in ingestion_failures[:20] + ) + extra = f"\n ... and {len(ingestion_failures) - 20} more" if len(ingestion_failures) > 20 else "" + raise RuntimeError( + f"Failed to add {len(ingestion_failures)} of {len(all_docs)} " + f"documents during session ingestion:\n{failure_details}{extra}" + ) session_info_here = self._make_session_info(session, is_linked=False) # For ingested sessions, clear the path arg (matches MATLAB kludge) @@ -821,11 +835,24 @@ def __init__( session_id=dataset_session_id, ) # Bulk-add all documents to the database + failures: list[tuple[str, str]] = [] for doc in documents: try: self._session._database.add(doc) - except Exception: - pass + except FileExistsError: + pass # Duplicates are expected and safe to skip + except Exception as exc: + doc_id = getattr(doc, "id", None) or doc.get("base", {}).get("id", "") if isinstance(doc, dict) else "" + failures.append((str(doc_id), str(exc))) + if failures: + failure_details = "\n".join( + f" - {doc_id}: {err}" for doc_id, err in failures[:20] + ) + extra = f"\n ... and {len(failures) - 20} more" if len(failures) > 20 else "" + raise RuntimeError( + f"Failed to add {len(failures)} of {len(documents)} " + f"documents to dataset database:\n{failure_details}{extra}" + ) # Re-create session without forced ID (reads from database) self._session = ndi_session_dir(ref or "temp", self._path) elif path_or_ref is None and not ref: From 5b8eaa3084b1403786dd09f7f540dfd1fa86dbaa Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 25 Mar 2026 15:38:27 +0000 Subject: [PATCH 2/2] Apply black formatting https://claude.ai/code/session_015oe6ucKwCFznpu1PN1nAfz --- src/ndi/cloud/download.py | 8 ++++---- src/ndi/cloud/orchestration.py | 16 ++++------------ src/ndi/dataset/_dataset.py | 16 +++++++++++----- 3 files changed, 19 insertions(+), 21 deletions(-) diff --git a/src/ndi/cloud/download.py b/src/ndi/cloud/download.py index fba2321..59cb2bf 100644 --- a/src/ndi/cloud/download.py +++ b/src/ndi/cloud/download.py @@ -268,12 +268,12 @@ def jsons2documents( try: documents.append(ndi_document(dj)) except Exception as exc: - doc_id = dj.get("base", {}).get("id", "") if isinstance(dj, dict) else "" + doc_id = ( + dj.get("base", {}).get("id", "") if isinstance(dj, dict) else "" + ) failures.append((doc_id, str(exc))) if failures: - failure_details = "\n".join( - f" - {doc_id}: {err}" for doc_id, err in failures[:20] - ) + failure_details = "\n".join(f" - {doc_id}: {err}" for doc_id, err in failures[:20]) extra = f"\n ... and {len(failures) - 20} more" if len(failures) > 20 else "" warnings.warn( f"Failed to convert {len(failures)} of {len(doc_jsons)} " diff --git a/src/ndi/cloud/orchestration.py b/src/ndi/cloud/orchestration.py index 02409c1..39756e7 100644 --- a/src/ndi/cloud/orchestration.py +++ b/src/ndi/cloud/orchestration.py @@ -127,13 +127,9 @@ def downloadDataset( if total_lost > 0: parts = [] if conversion_lost > 0: - parts.append( - f"{conversion_lost} failed to convert from JSON to ndi_document" - ) + parts.append(f"{conversion_lost} failed to convert from JSON to ndi_document") if db_lost > 0: - parts.append( - f"{db_lost} failed to add to the dataset database" - ) + parts.append(f"{db_lost} failed to add to the dataset database") raise RuntimeError( f"Downloaded {len(doc_jsons)} documents but only {db_count} " f"were added to the dataset. {total_lost} documents lost: " @@ -468,17 +464,13 @@ def _sync_download_new( conversion_lost = len(new_docs) - len(documents) total_lost = conversion_lost + len(failures) if total_lost > 0: - failure_details = "\n".join( - f" - {doc_id}: {err}" for doc_id, err in failures[:20] - ) + failure_details = "\n".join(f" - {doc_id}: {err}" for doc_id, err in failures[:20]) extra = f"\n ... and {len(failures) - 20} more" if len(failures) > 20 else "" parts = [] if conversion_lost > 0: parts.append(f"{conversion_lost} failed JSON-to-document conversion") if failures: - parts.append( - f"{len(failures)} failed to add to database:\n{failure_details}{extra}" - ) + parts.append(f"{len(failures)} failed to add to database:\n{failure_details}{extra}") raise RuntimeError( f"Sync downloaded {len(new_docs)} documents but only {added} " f"were added. {total_lost} documents lost: " + "; ".join(parts) diff --git a/src/ndi/dataset/_dataset.py b/src/ndi/dataset/_dataset.py index 4aa7a9e..cdc7620 100644 --- a/src/ndi/dataset/_dataset.py +++ b/src/ndi/dataset/_dataset.py @@ -178,7 +178,11 @@ def add_ingested_session(self, session: Any) -> ndi_dataset: failure_details = "\n".join( f" - {doc_id}: {err}" for doc_id, err in ingestion_failures[:20] ) - extra = f"\n ... and {len(ingestion_failures) - 20} more" if len(ingestion_failures) > 20 else "" + extra = ( + f"\n ... and {len(ingestion_failures) - 20} more" + if len(ingestion_failures) > 20 + else "" + ) raise RuntimeError( f"Failed to add {len(ingestion_failures)} of {len(all_docs)} " f"documents during session ingestion:\n{failure_details}{extra}" @@ -842,12 +846,14 @@ def __init__( except FileExistsError: pass # Duplicates are expected and safe to skip except Exception as exc: - doc_id = getattr(doc, "id", None) or doc.get("base", {}).get("id", "") if isinstance(doc, dict) else "" + doc_id = ( + getattr(doc, "id", None) or doc.get("base", {}).get("id", "") + if isinstance(doc, dict) + else "" + ) failures.append((str(doc_id), str(exc))) if failures: - failure_details = "\n".join( - f" - {doc_id}: {err}" for doc_id, err in failures[:20] - ) + failure_details = "\n".join(f" - {doc_id}: {err}" for doc_id, err in failures[:20]) extra = f"\n ... and {len(failures) - 20} more" if len(failures) > 20 else "" raise RuntimeError( f"Failed to add {len(failures)} of {len(documents)} "