chg ! sync models with HOPE core

Author: vitali-yanushchyk-valor

src/country_workspace/admin/office.py

@@ -24,7 +24,7 @@ def programmes(self, btn: LinkButton) -> None:
 
     @button()
     def sync(self, request: HttpRequest) -> None:
-        from country_workspace.contrib.hope.sync.office import sync_offices
+        from country_workspace.contrib.hope.sync.data import SyncStep, sync_hope_data
 
-        totals = sync_offices()
+        totals = sync_hope_data(step=SyncStep.OFFICES)["offices"]
         self.message_user(request, f"{totals['add']} created - {totals['upd']} updated")

src/country_workspace/admin/program.py

@@ -76,7 +76,7 @@ def _action(request: HttpRequest) -> HttpResponse:
 
     @button()
     def sync(self, request: HttpRequest) -> None:
-        from country_workspace.contrib.hope.sync.office import sync_programs
+        from country_workspace.contrib.hope.sync.data import SyncStep, sync_hope_data
 
-        totals = sync_programs()
-        self.message_user(request, f"{totals['add']} created - {totals['upd']} updated - {totals['skip']} skipped")
+        totals = sync_hope_data(step=SyncStep.PROGRAMS)["programs"]
+        self.message_user(request, f"{totals['add']} created - {totals['upd']} updated")

src/country_workspace/contrib/aurora/sync.py

@@ -5,7 +5,7 @@
 
 from country_workspace.contrib.aurora.client import AuroraClient
 from country_workspace.contrib.aurora.models import Project, Registration
-from country_workspace.contrib.hope.sync.office import sync_programs
+from country_workspace.contrib.hope.sync.data import SyncStep, sync_hope_data
 from country_workspace.models import AsyncJob, SyncLog
 
 
@@ -23,7 +23,7 @@ def sync_all(job: AsyncJob) -> dict[str, Any]:
     client = AuroraClient()
     with cache.lock("sync-aurora"):
         return {
-            "programs": sync_programs(),
+            "programs": sync_hope_data(step=SyncStep.PROGRAMS),
             "projects": sync_projects(client),
             "registrations": sync_registrations(client),
         }

src/country_workspace/contrib/hope/sync/data.py

@@ -0,0 +1,220 @@
+from io import TextIOBase
+from collections.abc import Generator, Callable
+from typing import Any, Final
+from dataclasses import dataclass, field
+from enum import Enum, auto
+
+from django.core.cache import cache
+from django.db import DatabaseError
+from django.db.models import Q
+from hope_flex_fields.models import DataChecker
+from country_workspace.exceptions import RemoteError
+
+from country_workspace.models import BeneficiaryGroup, Office, Program, SyncLog
+
+from .. import constants
+from ..client import HopeClient
+
+
+MESSAGES: Final[dict[str, str]] = {
+    "DEACTIVATION_FAILURE": "Failed during deactivation: {error}",
+    "RECORDS_DEACTIVATED": "Deactivated '{count}' records (obsolete or marked inactive in source).",
+    "RECORD_MISSING_ID": "Skipping record due to missing 'id': {record}",
+    "RECORD_SYNC_FAILURE": "Failed to sync DB record '{hope_id}': {error}",
+    "REMOTE_API_FAILURE": "API Error fetching '{path}': {error}",
+    "SYNC_COMPLETE": "Sync complete for {entity} with result {result}.",
+    "SYNC_START": "Start fetching '{entity}' data from HOPE...",
+}
+
+
+class LogLevel(Enum):
+    INFO = auto()
+    ERROR = auto()
+
+
+@dataclass
+class BaseSync:
+    client: HopeClient = field(default_factory=HopeClient)
+    stdout: TextIOBase | None = None
+    total: dict[str, dict[str, int] | list[str]] = field(default_factory=dict)
+
+    def safe_get(self, path: str) -> Generator[dict[str, Any], None, None]:
+        try:
+            yield from self.client.get(path)
+        except RemoteError as e:
+            self.emit_log("REMOTE_API_FAILURE", LogLevel.ERROR, path=path, error=e)
+
+    def emit_log(self, key: str, tag: LogLevel = LogLevel.INFO, **kwargs: Any) -> None:
+        """Emit a log message with the specified key and tag."""
+        template = MESSAGES.get(key)
+        if template is None:
+            raise KeyError(f"Log key '{key}' not found in MESSAGES configuration.")
+        try:
+            msg = template.format(**kwargs)
+        except KeyError as e:
+            raise ValueError(
+                f"Log format error for key '{key}': incorrect arguments provided. "
+                f"Missing placeholder: {e}. Provided args: {kwargs}"
+            )
+
+        if self.stdout:
+            self.stdout.write(f"[{tag.name}] {msg}\n")
+            self.stdout.flush()
+        if tag == LogLevel.ERROR:
+            self.total.setdefault("errors", []).append(msg)
+
+
+@dataclass
+class SyncAll(BaseSync):
+    programs_limit_to_office: Office | None = None
+    total: dict[str, dict[str, int] | list[str]] = field(
+        default_factory=lambda: {
+            **{_: {"add": 0, "upd": 0} for _ in ("offices", "programs", "beneficiary_groups")},
+            "errors": [],
+        }
+    )
+    default_checkers: dict[str, DataChecker] = field(
+        default_factory=lambda: {
+            "hh": DataChecker.objects.filter(name=constants.HOUSEHOLD_CHECKER_NAME).first(),
+            "ind": DataChecker.objects.filter(name=constants.INDIVIDUAL_CHECKER_NAME).first(),
+            "ppl": DataChecker.objects.filter(name=constants.PEOPLE_CHECKER_NAME).first(),
+        }
+    )
+
+    def sync_offices(self) -> None:
+        with cache.lock("sync-offices"):
+            self.emit_log("SYNC_START", entity="Office")
+            all_processed_hope_ids, inactive_hope_ids_from_source = set(), set()
+            for record in self.safe_get("business_areas"):
+                if not (hope_id := record.get("id")):
+                    self.emit_log("RECORD_MISSING_ID", LogLevel.ERROR, record=record)
+                    continue
+                all_processed_hope_ids.add(hope_id)
+                if record.get("active"):
+                    try:
+                        __, created = Office.objects.update_or_create(
+                            hope_id=hope_id,
+                            defaults={k: record.get(k) for k in ("name", "slug", "code", "long_name", "active")},
+                        )
+                        self.total["offices"]["add" if created else "upd"] += 1
+                    except DatabaseError as e:
+                        self.emit_log("RECORD_SYNC_FAILURE", LogLevel.ERROR, hope_id=hope_id, error=e)
+                else:
+                    inactive_hope_ids_from_source.add(hope_id)
+            # Deactivate records in the database that are not present in the source or are inactive.
+            try:
+                deactivated_count = Office.objects.filter(
+                    Q(active=True)
+                    & (~Q(hope_id__in=all_processed_hope_ids) | Q(hope_id__in=inactive_hope_ids_from_source))
+                ).update(active=False)
+                if deactivated_count > 0:
+                    self.total["offices"]["deactivated"] = deactivated_count
+                    self.emit_log("RECORDS_DEACTIVATED", count=deactivated_count)
+            except DatabaseError as e:
+                self.emit_log("DEACTIVATION_FAILURE", LogLevel.ERROR, error=e)
+
+            SyncLog.objects.register_sync(Office)
+            self.emit_log("SYNC_COMPLETE", entity="Office", result=self.total["offices"])
+
+    def sync_beneficiary_groups(self) -> None:
+        with cache.lock("sync-beneficiary_groups"):
+            self.emit_log("SYNC_START", entity="Beneficiary Group")
+            for record in self.safe_get("beneficiary-groups"):
+                if not (hope_id := record.get("id")):
+                    self.emit_log("RECORD_MISSING_ID", LogLevel.ERROR, record=record)
+                    continue
+                try:
+                    __, created = BeneficiaryGroup.objects.update_or_create(
+                        hope_id=hope_id,
+                        defaults={
+                            k: record.get(k)
+                            for k in (
+                                "name",
+                                "group_label",
+                                "group_label_plural",
+                                "member_label",
+                                "member_label_plural",
+                                "master_detail",
+                            )
+                        },
+                    )
+                    self.total["beneficiary_groups"]["add" if created else "upd"] += 1
+                except DatabaseError as e:
+                    self.emit_log("RECORD_SYNC_FAILURE", LogLevel.ERROR, hope_id=hope_id, error=e)
+
+            SyncLog.objects.register_sync(BeneficiaryGroup)
+            self.emit_log("SYNC_COMPLETE", entity="Beneficiary Group", result=self.total["beneficiary_groups"])
+
+    def sync_programs(self) -> None:
+        def is_valid() -> bool:
+            if not hope_id:
+                self.emit_log("RECORD_MISSING_ID", LogLevel.ERROR, record=record)
+                return False
+            return record["status"] in [Program.ACTIVE, Program.DRAFT]
+
+        self.sync_beneficiary_groups()
+        with cache.lock("sync-programs"):
+            self.emit_log("SYNC_START", entity="Program")
+            office = self.programs_limit_to_office if self.programs_limit_to_office else None
+            for record in self.safe_get("programs"):
+                try:
+                    if (hope_id := record.get("id")) and is_valid():
+                        office = Office.objects.get(code=record["business_area_code"])
+                        bg = BeneficiaryGroup.objects.get(hope_id=record["beneficiary_group"])
+                        p, created = Program.objects.update_or_create(
+                            hope_id=hope_id,
+                            defaults={
+                                "name": record["name"],
+                                "code": record["programme_code"],
+                                "status": record["status"],
+                                "sector": record["sector"],
+                                "country_office": office,
+                                "beneficiary_group": bg,
+                            },
+                        )
+                        if created:
+                            p.household_checker = self.default_checkers["hh"]
+                            p.individual_checker = (
+                                self.default_checkers["ind"] if bg.master_detail else self.default_checkers["ppl"]
+                            )
+                            p.save(update_fields=("household_checker", "individual_checker"))
+                        self.total["programs"]["add" if created else "upd"] += 1
+                except Office.DoesNotExist:
+                    continue
+                except BeneficiaryGroup.DoesNotExist as e:
+                    self.emit_log(
+                        "RECORD_SYNC_FAILURE", LogLevel.ERROR, hope_id=hope_id, error=f"BeneficiaryGroup not found: {e}"
+                    )
+                except DatabaseError as e:
+                    self.emit_log("RECORD_SYNC_FAILURE", LogLevel.ERROR, hope_id=hope_id, error=e)
+
+            SyncLog.objects.register_sync(Program)
+            self.emit_log("SYNC_COMPLETE", entity="Program", result=self.total["programs"])
+
+
+class SyncStep(Enum):
+    OFFICES = auto()
+    PROGRAMS = auto()
+
+    @property
+    def func(self) -> Callable[[SyncAll], None]:
+        func_map = {
+            SyncStep.OFFICES: SyncAll.sync_offices,
+            SyncStep.PROGRAMS: SyncAll.sync_programs,
+        }
+        return func_map[self]
+
+
+def sync_hope_data(
+    step: SyncStep | None = None, programs_limit_to_office: Office | None = None, stdout: TextIOBase | None = None
+) -> dict[str, Any]:
+    sync = SyncAll(stdout=stdout)
+    steps = tuple(SyncStep) if step is None else (step,)
+
+    for current_step in steps:
+        current_step.func(sync)
+        if sync.total["errors"]:
+            SyncLog.objects.refresh()
+            return sync.total
+    SyncLog.objects.refresh()
+    return sync.total