diff --git a/src/ndi/cloud/download.py b/src/ndi/cloud/download.py index da0f382..59cb2bf 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..39756e7 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,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 "" + 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..cdc7620 100644 --- a/src/ndi/dataset/_dataset.py +++ b/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", "") + 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", "") + 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: