add ! sync countries through admin panel (#87)

Author: vitali-yanushchyk-valor

src/country_workspace/admin/__init__.py

@@ -15,7 +15,7 @@
 from .office import OfficeAdmin
 from .program import ProgramAdmin
 from .role import UserRoleAdmin
-from .sync import SyncLog
+from .sync_log import SyncLog
 from .user import UserAdmin
 
 site.register(ContentType, admin_class=ContentTypeAdmin)

src/country_workspace/admin/base.py

@@ -1,31 +1,7 @@
-from typing import TypedDict
 from admin_extra_buttons.mixins import ExtraButtonsMixin
 from adminfilters.mixin import AdminAutoCompleteSearchMixin, AdminFiltersMixin
-from django.contrib import admin, messages
-from django.db.models import Model
-from django.http import HttpRequest
-from admin_extra_buttons.api import button
-
-from country_workspace.contrib.hope.sync.context_programs import SyncStep, sync_context_programs
+from django.contrib import admin
 
 
 class BaseModelAdmin(ExtraButtonsMixin, AdminAutoCompleteSearchMixin, AdminFiltersMixin, admin.ModelAdmin):
     pass
-
-
-class SyncConfig(TypedDict):
-    model: type[Model]
-    step: SyncStep
-
-
-class SyncAdminMixin:
-    sync_config: SyncConfig
-
-    @button()
-    def sync(self, request: HttpRequest) -> None:
-        totals = sync_context_programs(step=self.sync_config["step"])
-        if errors := totals.get("errors"):
-            self.message_user(request, "; ".join(errors), level=messages.ERROR)
-        else:
-            info = totals[self.sync_config["model"]._meta.model_name]
-            self.message_user(request, f"{info['add']} created - {info['upd']} updated", level=messages.SUCCESS)

src/country_workspace/admin/locations.py

@@ -9,7 +9,11 @@
 from django.db.models import Field
 from django.forms import FileField, FileInput, Form
 
-from country_workspace.models.locations import Area, AreaType, Country
+from ..models.locations import Area, AreaType, Country
+from ..contrib.hope.sync.context_geo import SyncStep
+from .base import BaseModelAdmin
+from .sync import SyncAdminMixin, SyncConfig, ContextGeoSyncHandler
+
 
 if TYPE_CHECKING:
     from django.http import HttpRequest
@@ -22,7 +26,7 @@ class ImportCSVForm(Form):
 
 
 @admin.register(Country)
-class CountryAdmin(AdminFiltersMixin, admin.ModelAdmin):
+class CountryAdmin(SyncAdminMixin, BaseModelAdmin):
     list_display = (
         "name",
         "iso_code2",
@@ -31,6 +35,7 @@ class CountryAdmin(AdminFiltersMixin, admin.ModelAdmin):
         "name",
         "iso_code2",
     )
+    sync_config = SyncConfig(model=Country, step=SyncStep.COUNTRIES, sync_handler=ContextGeoSyncHandler())
 
 
 @admin.register(AreaType)

src/country_workspace/admin/office.py

@@ -2,10 +2,11 @@
 from admin_extra_buttons.decorators import link
 from django.contrib import admin
 from django.urls import reverse
-from country_workspace.contrib.hope.sync.context_programs import SyncStep
 
+from ..contrib.hope.sync.context_programs import SyncStep
 from ..models import Office
-from .base import BaseModelAdmin, SyncAdminMixin, SyncConfig
+from .base import BaseModelAdmin
+from .sync import SyncAdminMixin, SyncConfig, ContextProgramsSyncHandler
 
 
 @admin.register(Office)
@@ -15,7 +16,7 @@ class OfficeAdmin(SyncAdminMixin, BaseModelAdmin):
     list_filter = ("active",)
     readonly_fields = ("hope_id", "slug")
     ordering = ("name",)
-    sync_config = SyncConfig(model=Office, step=SyncStep.OFFICES)
+    sync_config = SyncConfig(model=Office, step=SyncStep.OFFICES, sync_handler=ContextProgramsSyncHandler())
 
     @link(change_list=False)
     def programmes(self, btn: LinkButton) -> None:

src/country_workspace/admin/program.py

@@ -9,8 +9,10 @@
 from ..cache.manager import cache_manager
 from ..compat.admin_extra_buttons import confirm_action
 from ..models import Program
-from .base import BaseModelAdmin, SyncAdminMixin, SyncConfig
-from country_workspace.contrib.hope.sync.context_programs import SyncStep
+from ..contrib.hope.sync.context_programs import SyncStep
+from .base import BaseModelAdmin
+from .sync import SyncAdminMixin, SyncConfig, ContextProgramsSyncHandler
+
 
 if TYPE_CHECKING:
     from admin_extra_buttons.buttons import LinkButton
@@ -39,7 +41,7 @@ class ProgramAdmin(SyncAdminMixin, BaseModelAdmin):
     )
     ordering = ("name",)
     autocomplete_fields = ("country_office",)
-    sync_config = SyncConfig(model=Program, step=SyncStep.PROGRAMS)
+    sync_config = SyncConfig(model=Program, step=SyncStep.PROGRAMS, sync_handler=ContextProgramsSyncHandler())
 
     @button()
     def invalidate_cache(self, request: HttpRequest, pk: str) -> None:

src/country_workspace/admin/sync.py

@@ -1,15 +1,46 @@
-from admin_extra_buttons.decorators import button
-from django.contrib import admin
-from django.http import HttpRequest, HttpResponse
+from typing import TypedDict, Protocol, TypeVar, Generic
+from django.contrib import messages
+from django.db.models import Model
+from django.http import HttpRequest
+from admin_extra_buttons.api import button
 
-from ..models import SyncLog
-from .base import BaseModelAdmin
+from ..contrib.hope.sync.context_programs import sync_context_programs, SyncStep as ContextProgramsSyncStep
+from ..contrib.hope.sync.context_geo import sync_context_geo, SyncStep as ContextGeoSyncStep
 
 
-@admin.register(SyncLog)
-class SyncLogAdmin(BaseModelAdmin):
-    list_display = ("content_type", "content_object", "last_update_date", "last_id")
+T_SyncStep = TypeVar("T_SyncStep", bound=ContextProgramsSyncStep | ContextGeoSyncStep)
+type SyncHandlerResp = dict[str, list[str] | dict[str, int]]
+
+
+class SyncHandler(Protocol, Generic[T_SyncStep]):
+    def sync(self, step: T_SyncStep) -> SyncHandlerResp:
+        pass
+
+
+class ContextProgramsSyncHandler:
+    def sync(self, step: ContextProgramsSyncStep) -> SyncHandlerResp:
+        return sync_context_programs(step)
+
+
+class ContextGeoSyncHandler:
+    def sync(self, step: ContextGeoSyncStep) -> SyncHandlerResp:
+        return sync_context_geo(step)
+
+
+class SyncConfig(TypedDict):
+    model: type[Model]
+    step: T_SyncStep
+    sync_handler: SyncHandler
+
+
+class SyncAdminMixin:
+    sync_config: SyncConfig
 
     @button()
-    def sync_all(self, request: HttpRequest) -> "HttpResponse":
-        SyncLog.objects.refresh()
+    def sync(self, request: HttpRequest) -> None:
+        totals = self.sync_config["sync_handler"].sync(step=self.sync_config["step"])
+        if errors := totals.get("errors"):
+            self.message_user(request, "; ".join(errors), level=messages.ERROR)
+        else:
+            info = totals[self.sync_config["model"]._meta.model_name]
+            self.message_user(request, f"{info['add']} created - {info['upd']} updated", level=messages.SUCCESS)

src/country_workspace/admin/sync_log.py

@@ -0,0 +1,15 @@
+from admin_extra_buttons.decorators import button
+from django.contrib import admin
+from django.http import HttpRequest, HttpResponse
+
+from ..models import SyncLog
+from .base import BaseModelAdmin
+
+
+@admin.register(SyncLog)
+class SyncLogAdmin(BaseModelAdmin):
+    list_display = ("content_type", "content_object", "last_update_date", "last_id")
+
+    @button()
+    def sync_all(self, request: HttpRequest) -> "HttpResponse":
+        SyncLog.objects.refresh()

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

@@ -0,0 +1,56 @@
+from io import TextIOBase
+from typing import Any, Final
+from dataclasses import dataclass
+from enum import auto
+from django.db.models import Model
+
+from ....models import Country
+from .base import BaseSync, SyncConfig, BaseSyncStep, sync_context
+
+
+MODELS: Final[tuple[type[Model], ...]] = (Country,)
+"""List of models to synchronize."""
+
+
+class SyncStep(BaseSyncStep):
+    """Synchronization steps for geo-related models."""
+
+    COUNTRIES = (auto(), lambda self: self.sync_countries)
+
+
+@dataclass
+class SyncContextGeo(BaseSync):
+    """Context for synchronizing geo-related models."""
+
+    SyncStep = SyncStep
+
+    def sync_countries(self) -> None:
+        """Fetch and process Country records from the remote API."""
+        self.sync_entity(
+            SyncConfig(
+                model=Country,
+                path="lookups/country",
+                prepare_defaults=lambda r: {f: r.get(f) for f in ("name", "iso_code2")},
+            ),
+        )
+
+
+def sync_context_geo(
+    step: SyncStep | None = None,
+    stdout: TextIOBase | None = None,
+) -> dict[str, Any]:
+    """Run synchronization for geo-related models.
+
+    Args:
+        step (SyncStep | None): Specific step to execute (e.g., SyncStep.COUNTRIES). If None, all steps are run.
+        stdout (TextIOBase | None): Optional output stream for logging.
+
+    Returns:
+        dict[str, Any]: Synchronization results, including counts and errors.
+
+    """
+    return sync_context(
+        SyncContextGeo,
+        step=step,
+        stdout=stdout,
+    )

src/country_workspace/migrations/0012_country_hope_id.py

@@ -0,0 +1,31 @@
+from django.db import migrations, models
+from django.db.migrations.state import StateApps
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+from uuid import uuid4
+
+
+def fill_hope_id(apps: StateApps, schema_editor: BaseDatabaseSchemaEditor) -> None:
+    Country = apps.get_model("country_workspace", "Country")
+    for country in Country.objects.all():
+        country.hope_id = f"TEMP_COUNTRY_{uuid4()}"
+        country.save()
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("country_workspace", "0011_remove_household_updates_update_and_more"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="country",
+            name="hope_id",
+            field=models.CharField(editable=False, max_length=200, null=True, unique=True),
+        ),
+        migrations.RunPython(fill_hope_id, reverse_code=migrations.RunPython.noop),
+        migrations.AlterField(
+            model_name="country",
+            name="hope_id",
+            field=models.CharField(editable=False, max_length=200, unique=True),
+        ),
+    ]

src/country_workspace/models/locations.py

@@ -10,6 +10,7 @@
 class Country(BaseModel):
     name = models.CharField(max_length=255, db_index=True)
     iso_code2 = models.CharField(max_length=2, unique=True)
+    hope_id = models.CharField(max_length=200, unique=True, editable=False)
 
     class Meta:
         verbose_name_plural = "Countries"