Konfigurowalna semantyka błędów importu (on_error) + adopcja download_to_model

src/bpp/newsfragments/pbn-pobierz-mongodb-on-error.feature.rst

@@ -0,0 +1,7 @@
+``pobierz_mongodb`` przyjmuje teraz parametr ``on_error``: ``"raise"`` (domyślne,
+dotychczasowe fail-fast — błąd zapisu jednego rekordu przerywa cały import) lub
+``"skip"`` (deleguje do pakietowego ``download_to_model`` — zły rekord jest
+logowany i liczony, a import reszty listy kończy się; zwraca
+``DownloadResult(processed, errored)``). Przydatne przy masowych synchronizacjach,
+gdzie pojedynczy zepsuty rekord nie powinien przerywać całego przebiegu.
+Zachowanie domyślne bez zmian.

src/pbn_integrator/tests/test_pobierz_mongodb_on_error.py

@@ -0,0 +1,58 @@
+"""Knob ``on_error`` w ``pobierz_mongodb``: fail-fast (default) vs skip-and-continue.
+
+Błąd dotyczy zapisu POJEDYNCZEGO rekordu do lokalnego lustra BPP (nie PBN).
+``"raise"`` (default) zachowuje historyczne fail-fast; ``"skip"`` deleguje do
+pakietowego ``download_to_model`` (log + licznik + kontynuacja).
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def _fun_failing_on(bad_id):
+    """Save-fun rzucający na elemencie o danym ``id``, resztę „zapisujący”."""
+    saved = []
+
+    def fun(elem, klass, client=None, **extra):
+        if elem["id"] == bad_id:
+            raise ValueError(f"boom {bad_id}")
+        saved.append(elem["id"])
+
+    fun.saved = saved
+    return fun
+
+
+def test_on_error_raise_jest_domyslny_fail_fast():
+    """Default = fail-fast: pierwszy błąd przerywa batch, reszta nietknięta."""
+    from pbn_integrator.utils.mongodb_ops import pobierz_mongodb
+
+    elems = [{"id": "a"}, {"id": "b"}, {"id": "c"}]
+    fun = _fun_failing_on("b")
+    with pytest.raises(ValueError, match="boom b"):
+        pobierz_mongodb(elems, MagicMock(), fun=fun, disable_progress_bar=True)
+    # „c” (po zepsutym „b”) NIE zostało zapisane — batch przerwany.
+    assert fun.saved == ["a"]
+
+
+def test_on_error_skip_kontynuuje_i_liczy_bledy():
+    """``skip`` = kontynuuj mimo błędu; zwraca DownloadResult z licznikami."""
+    from pbn_integrator.utils.mongodb_ops import pobierz_mongodb
+
+    elems = [{"id": "a"}, {"id": "b"}, {"id": "c"}]
+    fun = _fun_failing_on("b")
+    result = pobierz_mongodb(
+        elems, MagicMock(), fun=fun, disable_progress_bar=True, on_error="skip"
+    )
+    # „a” i „c” zapisane mimo błędu na „b”.
+    assert fun.saved == ["a", "c"]
+    assert result.processed == 2
+    assert result.errored == 1
+
+
+def test_on_error_nieznana_wartosc_to_valueerror():
+    """Literówka w ``on_error`` fail-uje głośno, nie po cichu wybiera tryb."""
+    from pbn_integrator.utils.mongodb_ops import pobierz_mongodb
+
+    with pytest.raises(ValueError, match="on_error"):
+        pobierz_mongodb([], MagicMock(), fun=lambda *a, **k: None, on_error="whatever")

src/pbn_integrator/utils/mongodb_ops.py

@@ -7,6 +7,7 @@
 from typing import TYPE_CHECKING
 
 import rollbar
+from django_pbn_client.download import download_to_model
 from django_pbn_client.persistence import (
     download_pbn_objects,
     get_or_download,
@@ -284,6 +285,7 @@ def pobierz_mongodb(
     client=None,
     disable_progress_bar=False,
     callback=None,
+    on_error="raise",
 ):
     """Fetch and save elements from PBN API.
 
@@ -295,24 +297,50 @@ def pobierz_mongodb(
         client: PBN client.
         disable_progress_bar: Whether to disable progress bar.
         callback: Optional callback for progress tracking.
+        on_error: Zachowanie przy błędzie zapisu POJEDYNCZEGO rekordu do
+            lokalnego lustra BPP (``IntegrityError``, zły kształt danych itp.):
+
+            - ``"raise"`` (default) — fail-fast: pierwszy błąd propaguje i
+              przerywa cały batch (zachowanie historyczne). Zwraca ``None``.
+            - ``"skip"`` — skip-and-log-and-continue: zły rekord jest logowany
+              (pełny traceback) i liczony, import reszty listy kończy się.
+              Deleguje do pakietowego ``download_to_model`` i zwraca
+              ``DownloadResult(processed, errored)``. Przydatne przy masowych
+              synchronizacjach (tysiące rekordów), gdzie jeden zepsuty rekord
+              nie powinien wywalać całego przebiegu.
     """
     if fun is None:
         fun = zapisz_mongodb
 
-    def progress(elements, total, label):
+    def progress(elements, total, _label):
+        # Używamy ``pbar_label`` (nie ``_label`` z delegata) — ``download_to_model``
+        # nie forwarduje etykiety, więc trzymamy ją stałą w obu trybach.
         return pbar(
             elements,
             total,
-            label,
+            pbar_label,
             disable_progress_bar=disable_progress_bar,
             callback=callback,
         )
 
-    return download_pbn_objects(
-        elems,
-        klass,
-        label=pbar_label,
-        save=fun,
-        client=client,
-        progress=progress,
-    )
+    if on_error == "raise":
+        return download_pbn_objects(
+            elems,
+            klass,
+            label=pbar_label,
+            save=fun,
+            client=client,
+            progress=progress,
+        )
+    if on_error == "skip":
+        # ``elems`` jest już zbudowanym paginatorem/iteratorem — fasada oczekuje
+        # fabryki (zero-arg), więc oddajemy gotowy zasób. Domyślne
+        # ``concurrency=None`` → ścieżka sekwencyjna (bez ponownego requestu).
+        return download_to_model(
+            lambda: elems,
+            klass,
+            save=fun,
+            client=client,
+            progress=progress,
+        )
+    raise ValueError(f"on_error musi być 'raise' albo 'skip', otrzymano {on_error!r}")