Raise errors when documents fail to add to dataset

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", "<unknown>") if isinstance(dj, dict) else "<invalid>"
+            )
+            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
 
 

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,30 @@ 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 +453,28 @@ 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 "<unknown>"
+            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}
 

src/ndi/dataset/_dataset.py

@@ -163,12 +163,30 @@ 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", "<unknown>")
+                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 +839,26 @@ 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", "<unknown>")
+                        if isinstance(doc, dict)
+                        else "<unknown>"
+                    )
+                    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: