rebase; refactor; tests

Author: vitali-yanushchyk-valor

docs/src/flows/rdp_matrix.md

@@ -3,14 +3,12 @@
     create_rdp_core,
     dedup_existing_rdp_core,
     push_existing_rdp_core,
-    dedup_engine_status_or_error,
 )
 
 __all__ = [
     "CreateRdpConfig",
     "PushExistingRdpConfig",
     "create_rdp_core",
-    "dedup_engine_status_or_error",
     "dedup_existing_rdp_core",
     "push_existing_rdp_core",
 ]

src/country_workspace/contrib/hope/push/__init__.py

@@ -1,30 +1,25 @@
 from collections.abc import Callable, Iterator
 from functools import partial
 from typing import Any
-from requests.exceptions import RequestException
-from uuid import UUID
 from django.db import transaction, IntegrityError
 
 from country_workspace.contrib.hope.exceptions import HopePushError
-from country_workspace.contrib.hope.constants import IMAGES_TO_DEDUPLICATE_BULK_BATCH_SIZE
 from country_workspace.models import AsyncJob, Rdp
 from country_workspace.workspaces.models import CountryIndividual
-from country_workspace.contrib.dedup_engine.client import Status as DedupClientStatus, make_client
 from country_workspace.contrib.dedup_engine.response import Status as DedupResponseStatus
 
 
-from .processor import PushProcessor
+from .processor import HopeProcessor, DedupProcessor
 from .config import CreateRdpConfig, PushWorkflowConfig
 from .repository import (
     individuals_by_pks,
-    individuals_for_rdp,
     individuals_by_household_pks,
     households,
     rdp_for_push,
-    rdp_for_dedup,
     preflight_errors,
     workflow_config_for_rdp,
 )
+from .transport import dedup_api
 
 
 def create_rdp_records(config: CreateRdpConfig, job_id: int) -> Rdp:
@@ -64,7 +59,7 @@ def mark_rdp_beneficiaries_removed(rdp: Rdp, is_master_detail: bool) -> None:
         rdp.individuals.update(removed=True)
 
 
-def steps(processor: PushProcessor, config: PushWorkflowConfig) -> Iterator[Callable[[], None]]:
+def steps(processor: HopeProcessor, config: PushWorkflowConfig) -> Iterator[Callable[[], None]]:
     """Yield the ordered workflow callables; each step appends errors to processor.total."""
     pks = config["pks"]
 
@@ -100,99 +95,13 @@ def create_rdp_core(job: AsyncJob) -> dict[str, Any]:
     return {"rdp_id": rdp.id, "rdp_str": str(rdp)}
 
 
-def dedup_engine_status_or_error(program_code: str) -> DedupClientStatus:
-    """Fetch DedupEngine status or raise HopePushError."""
-    try:
-        with make_client(program_code) as client:
-            return client.status()
-    except (RequestException, ValueError) as e:
-        raise HopePushError({"errors": [f"DedupEngine: status() failed: {e}"]}) from e
-
-
-def dedup_engine_approve_or_error(program_code: str) -> None:
-    """Approve DedupEngine results or raise HopePushError."""
-    try:
-        with make_client(program_code) as client:
-            client.approve()
-    except (RequestException, ValueError) as e:
-        raise HopePushError({"errors": [f"DedupEngine: approve() failed: {e}"]}) from e
-
-
-def _parse_uuid(value: object) -> UUID | None:
-    """Return UUID parsed from value or None if parsing fails."""
-    if value is None:
-        return None
-    try:
-        return UUID(str(value))
-    except (TypeError, ValueError):
-        return None
-
-
-def _collect_dedup_images(*, rdp: Rdp) -> tuple[list[dict[str, str]], int]:
-    """Collect DedupEngine images payload (reference_pk, filename) from RDP individuals."""
-    rows = individuals_for_rdp(rdp=rdp).values_list("pk", "flex_fields__photo")
-
-    images: list[dict[str, str]] = []
-    skipped = 0
-    for pk, photo in rows.iterator(chunk_size=IMAGES_TO_DEDUPLICATE_BULK_BATCH_SIZE * 5):
-        if isinstance(photo, str) and (photo := photo.strip()):
-            images.append({"reference_pk": str(pk), "filename": photo})
-        else:
-            skipped += 1
-    return images, skipped
-
-
-def _dedup_engine_create_set_or_error(*, program_code: str) -> UUID:
-    """Create a DedupEngine deduplication set and return its UUID."""
-    try:
-        with make_client(program_code) as client:
-            ds_id = client.create_deduplication_set(settings={})
-    except (RequestException, ValueError, KeyError, TypeError) as e:
-        raise HopePushError({"errors": [f"DedupEngine: create_deduplication_set() failed: {e}"]}) from e
-
-    if (dedup_uuid := _parse_uuid(ds_id)) is None:
-        raise HopePushError({"errors": [f"DedupEngine: invalid deduplication_set_id={ds_id!r}"]})
-    return dedup_uuid
-
-
-def _dedup_engine_upload_and_process_or_error(*, program_code: str, images: list[dict[str, str]]) -> None:
-    """Upload images to DedupEngine and start processing."""
-    try:
-        with make_client(program_code) as client:
-            client.create_images(images)
-            client.process()
-    except (RequestException, ValueError, KeyError, TypeError) as e:
-        raise HopePushError({"errors": [f"DedupEngine: create_images()/process() failed: {e}"]}) from e
-
-
 def dedup_existing_rdp_core(job: AsyncJob) -> dict[str, Any]:
     """Run DedupEngine deduplication for an existing RDP."""
-    rdp = rdp_for_dedup(pk=job.config["rdp_id"])
-    if rdp.status != Rdp.PushStatus.PENDING:
-        raise HopePushError({"errors": [f"RDP: can not run dedup in status={rdp.status}"]})
-    if rdp.dedup_run_state == rdp.DedupRunState.APPROVED:
-        raise HopePushError({"errors": ["RDP: can not run dedup after approval"]})
-
-    images, skipped = _collect_dedup_images(rdp=rdp)
-    total = {
-        "rdp_id": rdp.pk,
-        "program": rdp.program.code,
-        "skipped_no_photo": skipped,
-        "images_sent": len(images),
-    }
-    if not images:
-        return total | {"deduplication_set_id": None}
-
-    deduplication_set_id = _dedup_engine_create_set_or_error(program_code=rdp.program.code)
-
-    Rdp.objects.filter(pk=rdp.pk).update(
-        deduplication_set_id=deduplication_set_id,
-        dedup_run_state=rdp.DedupRunState.IN_PROGRESS,
-    )
-
-    _dedup_engine_upload_and_process_or_error(program_code=rdp.program.code, images=images)
-
-    return total | {"deduplication_set_id": str(deduplication_set_id)}
+    processor = DedupProcessor(rdp_id=job.config["rdp_id"])
+    processor.run()
+    if processor.has_errors:
+        raise HopePushError(processor.total)
+    return processor.total
 
 
 def push_existing_rdp_core(job: AsyncJob) -> dict[str, Any]:
@@ -203,25 +112,24 @@ def push_existing_rdp_core(job: AsyncJob) -> dict[str, Any]:
     imported_by_email = getattr(job.owner, "email", "") or getattr(rdp.pushed_by, "email", "")
 
     config: PushWorkflowConfig = workflow_config_for_rdp(rdp=rdp, imported_by_email=imported_by_email)
-    processor = PushProcessor(config)
-    for step in steps(processor, config):
+    hope_processor = HopeProcessor(config)
+    for step in steps(hope_processor, config):
         step()
-        if processor.total["errors"]:
-            complete_rdp(rdp.id, Rdp.PushStatus.FAILURE, processor.hope_rdi_id or "N/A")
-            raise HopePushError(processor.total)
-
-    program_code = rdp.program.code
-    if rdp.dedup_run_state == Rdp.DedupRunState.IN_PROGRESS:
-        try:
-            if dedup_engine_status_or_error(program_code).status == DedupResponseStatus.SUCCESS:
-                dedup_engine_approve_or_error(program_code)
+        if hope_processor.has_errors:
+            complete_rdp(rdp.id, Rdp.PushStatus.FAILURE, hope_processor.hope_rdi_id or "N/A")
+            raise HopePushError(hope_processor.total)
+
+    if rdp.program.biometric_deduplication_enabled:
+        with dedup_api(rdp.program.code, hope_processor._err) as de:
+            if (de_status := de.status()) is None:
+                raise HopePushError(hope_processor.total)
+            if de_status.status == DedupResponseStatus.SUCCESS:
+                if not de.approve():
+                    raise HopePushError(hope_processor.total)
                 Rdp.objects.filter(pk=rdp.pk).update(dedup_run_state=Rdp.DedupRunState.APPROVED)
-        except HopePushError:
-            complete_rdp(rdp.id, Rdp.PushStatus.FAILURE, processor.hope_rdi_id or "N/A")
-            raise
 
     with transaction.atomic():
-        updated = complete_rdp(rdp.id, Rdp.PushStatus.SUCCESS, processor.hope_rdi_id)
+        updated = complete_rdp(rdp.id, Rdp.PushStatus.SUCCESS, hope_processor.hope_rdi_id)
         mark_rdp_beneficiaries_removed(updated, config["master_detail"])
 
-    return processor.total
+    return hope_processor.total

src/country_workspace/contrib/hope/push/orchestration.py

@@ -3,22 +3,53 @@
 from functools import cached_property
 from itertools import batched
 from typing import Any
+from uuid import UUID
 
 from django.db.models import QuerySet
 
 from country_workspace.contrib.hope.constants import PUSH_BATCH_SIZE
 from country_workspace.workspaces.models import CountryHousehold, CountryIndividual
 
+from country_workspace.contrib.hope.constants import IMAGES_TO_DEDUPLICATE_BULK_BATCH_SIZE
+from country_workspace.models import Rdp
+from .repository import individuals_for_rdp, rdp_for_dedup
+
+
 from .config import ROLE_FIELDS, Serializer, ERROR_CONFIG, PushWorkflowConfig
 from .mappings import load_mapping_from_api, map_members, map_role_value
 from .repository import serializer_for_program, preflight_errors
-from .transport import HopeApi
+from .transport import HopeApi, dedup_api
 
 
-class PushProcessor:
+class ProcessorBase:
+    """Shared processor primitives."""
+
+    def __init__(self) -> None:
+        self.total: dict[str, Any] = {"errors": []}
+
+    @property
+    def has_errors(self) -> bool:
+        """Return True when at least one error was collected."""
+        return bool(self.total["errors"])
+
+    def _err(self, msg: str) -> None:
+        """Append an error into total['errors']; truncate long text; cap the list with a marker."""
+        errors: list[str] = self.total["errors"]
+        if errors and errors[-1] == ERROR_CONFIG.MARKER:
+            return
+        if len(errors) >= ERROR_CONFIG.MAX_ERRORS - 1:
+            errors.append(ERROR_CONFIG.MARKER)
+            return
+        if len(msg) > ERROR_CONFIG.MAX_ERROR_LEN:
+            msg = f"{msg[: ERROR_CONFIG.MAX_ERROR_LEN - 1]}…"
+        errors.append(msg)
+
+
+class HopeProcessor(ProcessorBase):
     """Push pipeline: validate, prepare, send and track results via Hope API."""
 
     def __init__(self, config: PushWorkflowConfig) -> None:
+        super().__init__()
         self.api = HopeApi(co_slug=config["co_slug"], err=self._err)
         self.batch_name: str = config["batch_name"]
         self.hope_rdi_id: str | None = None
@@ -29,7 +60,6 @@ def __init__(self, config: PushWorkflowConfig) -> None:
         self.program_hope_id: str = config["program_hope_id"]
         self.queryset: QuerySet | None = None
         self.rdp_id: int | None = config.get("rdp_id")
-        self.total: dict[str, Any] = {"errors": []}
 
     @cached_property
     def serializer(self) -> Serializer:
@@ -85,18 +115,6 @@ def run_with(self, qs: QuerySet, step: Callable) -> None:
         with self._using_qs(qs):
             step()
 
-    def _err(self, msg: str) -> None:
-        """Append an error into total['errors']; truncate long text; cap the list with a marker."""
-        errors: list[str] = self.total["errors"]
-        if errors and errors[-1] == ERROR_CONFIG.MARKER:
-            return
-        if len(errors) >= ERROR_CONFIG.MAX_ERRORS - 1:
-            errors.append(ERROR_CONFIG.MARKER)
-            return
-        if len(msg) > ERROR_CONFIG.MAX_ERROR_LEN:
-            msg = f"{msg[: ERROR_CONFIG.MAX_ERROR_LEN - 1]}…"
-        errors.append(msg)
-
     def _prepare_households_batch(self, batch: Iterable[CountryHousehold]) -> tuple[list[int], list[dict]]:
         """Return (ids, payload) for a households batch: roles mapped, members resolved."""
         ids, rows = [], []
@@ -207,3 +225,65 @@ def _using_qs(self, qs: QuerySet) -> Iterator[None]:
             yield
         finally:
             self.queryset = prev
+
+
+class DedupProcessor(ProcessorBase):
+    """Dedup pipeline: collect images, create set, upload and start processing."""
+
+    def __init__(self, *, rdp_id: int) -> None:
+        super().__init__()
+        self.rdp = rdp_for_dedup(pk=rdp_id)
+        self.program_code = self.rdp.program.code
+
+    def run(self) -> None:
+        """Execute dedup workflow; collect errors in total."""
+        if self.rdp.status != Rdp.PushStatus.PENDING:
+            self._err(f"RDP: can not run dedup in status={self.rdp.status}")
+        if self.rdp.dedup_run_state == Rdp.DedupRunState.APPROVED:
+            self._err("RDP: can not run dedup after approval")
+        if self.has_errors:
+            return
+
+        self.total |= {"rdp_id": self.rdp.pk, "program": self.program_code}
+
+        images = self._collect_images()
+        self.total["images_sent"] = len(images)
+
+        if not images:
+            self.total["deduplication_set_id"] = None
+            return
+
+        ds_id = self._deduplicate(images)
+        self.total["deduplication_set_id"] = str(ds_id) if ds_id else None
+
+    def _collect_images(self) -> list[dict[str, str]]:
+        """Collect DedupEngine images payload (reference_pk, filename) from RDP individuals."""
+        rows = individuals_for_rdp(rdp=self.rdp).values_list("pk", "flex_fields__photo")
+
+        images: list[dict[str, str]] = []
+        for pk, photo in rows.iterator(chunk_size=IMAGES_TO_DEDUPLICATE_BULK_BATCH_SIZE * 5):
+            if isinstance(photo, str) and (photo := photo.strip()):
+                images.append({"reference_pk": str(pk), "filename": photo})
+        return images
+
+    def _deduplicate(self, images: list[dict[str, str]]) -> UUID | None:
+        """Run remote DedupEngine steps; return deduplication_set_id UUID on success."""
+        with dedup_api(self.program_code, self._err) as api:
+            if (raw := api.create_deduplication_set(settings={})) is None:
+                return None
+
+            try:
+                ds_id = UUID(str(raw))
+            except (TypeError, ValueError) as e:
+                self._err(f"DedupEngine: create_deduplication_set returned invalid UUID {raw!r}: {e}")
+                return None
+
+            Rdp.objects.filter(pk=self.rdp.pk).update(
+                deduplication_set_id=ds_id,
+                dedup_run_state=Rdp.DedupRunState.IN_PROGRESS,
+            )
+
+            if not (api.create_images(images) and api.process()):
+                return None
+
+            return ds_id