Feature/284491 kobo watermark (#265)

* Implement kobo watermark feature

* Add unit tests

* Use synclog instead of kobosubmission
Fix failing unit tests

* fix failing unit test
add field name to the synclog admin listing

* fix failing unit test
add field name to the synclog admin listing

* Remove unnecessary mocks from unit tests

* Improve error message on import failure

Author: datamik

src/country_workspace/admin/sync_log.py

@@ -8,7 +8,8 @@
 
 @admin.register(SyncLog)
 class SyncLogAdmin(BaseModelAdmin):
-    list_display = ("content_type", "content_object", "last_update_date", "last_id")
+    list_display = ("content_type", "name", "content_object", "last_update_date", "last_id")
+    search_fields = ("name", "content_type__model", "content_type__app_label")
 
     @button()
     def sync_flex_fields(self, request: HttpRequest) -> "HttpResponse":

src/country_workspace/config/settings.py

@@ -57,6 +57,7 @@
     # these should be optional in the future
     "country_workspace.contrib.hope.apps.Config",
     "country_workspace.contrib.aurora.apps.Config",
+    "country_workspace.contrib.kobo.apps.Config",
     *env("EXTRA_APPS"),
 )
 

src/country_workspace/contrib/kobo/api/client/helpers.py

@@ -74,13 +74,21 @@ def get_asset_list(data_getter: DataGetter, url: str) -> Generator[Asset, None,
     return handle_paginated_response(data_getter, url, get_raw_asset_list, partial(get_asset, data_getter))
 
 
-def get_submission_list(data_getter: DataGetter, url: str) -> Iterable[Submission]:
-    url_with_start_and_limit = change_url(
-        url, query={START_PARAMETER_NAME: START_PARAMETER_VALUE, LIMIT_PARAMETER_NAME: LIMIT_PARAMETER_VALUE}
-    )
+def get_submission_list(data_getter: DataGetter, url: str, min_id: int | None = None) -> Iterable[Submission]:
+    import json
+
+    query_params = {
+        START_PARAMETER_NAME: START_PARAMETER_VALUE,
+        LIMIT_PARAMETER_NAME: LIMIT_PARAMETER_VALUE,
+    }
+
+    if min_id is not None:
+        query_params["query"] = json.dumps({"_id": {"$gt": min_id}})
+
+    url_with_params = change_url(url, query=query_params)
     return map(
         partial(download_attachments, data_getter),
-        handle_paginated_response(data_getter, url_with_start_and_limit, get_raw_submission_list, Submission),
+        handle_paginated_response(data_getter, url_with_params, get_raw_submission_list, Submission),
     )
 
 

src/country_workspace/contrib/kobo/api/data/asset.py

@@ -6,7 +6,9 @@
 
 
 class Asset(Raw[raw_asset.Asset]):
-    def __init__(self, raw: raw_asset.Asset, submissions: Callable[[], Generator[Submission, None, None]]) -> None:
+    def __init__(
+        self, raw: raw_asset.Asset, submissions: Callable[[int | None], Generator[Submission, None, None]]
+    ) -> None:
         super().__init__(raw)
         self._submissions = submissions
 
@@ -18,9 +20,8 @@ def uid(self) -> str:
     def name(self) -> str:
         return self._raw["name"]
 
-    @property
-    def submissions(self) -> Generator[Submission, None, None]:
-        yield from self._submissions()
+    def submissions(self, min_id: int | None = None) -> Generator[Submission, None, None]:
+        yield from self._submissions(min_id)
 
     def __str__(self) -> str:
         return f"Asset: {self.name}"

src/country_workspace/contrib/kobo/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class Config(AppConfig):
+    name = __name__.rpartition(".")[0]
+    verbose_name = "Contributed Service | Kobo"

src/country_workspace/contrib/kobo/migrations/0001_initial.py

@@ -0,0 +1,20 @@
+# Generated by Django 5.2.8 on 2025-11-11 00:47
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="KoboSubmission",
+            fields=[
+                ("id", models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("asset_uid", models.CharField(editable=False, max_length=32, unique=True)),
+                ("last_submission_id", models.IntegerField(editable=False)),
+            ],
+        ),
+    ]

src/country_workspace/contrib/kobo/migrations/0002_delete_kobosubmission.py

@@ -0,0 +1,15 @@
+# Generated by Django 5.2.8 on 2025-11-11 13:42
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("kobo", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.DeleteModel(
+            name="KoboSubmission",
+        ),
+    ]

src/country_workspace/contrib/kobo/migrations/__init__.py

@@ -3,19 +3,22 @@
 from functools import partial
 from typing import Any, Final, TypedDict, cast
 from constance import config as constance_config
+from django.utils import timezone
 from requests import Session
 from requests.adapters import HTTPAdapter
 
+from django.contrib.contenttypes.models import ContentType
+
 from country_workspace.contrib.kobo.api.client.auth import Auth
 from country_workspace.contrib.kobo.api.client.main import Client
 from country_workspace.contrib.kobo.api.common import DataGetter
 from country_workspace.contrib.kobo.api.data.asset import Asset
 from country_workspace.contrib.kobo.api.data.submission import Submission
-from country_workspace.contrib.kobo.models import KoboSubmission
-from country_workspace.models import AsyncJob, Batch, Household, Individual
+from country_workspace.models import AsyncJob, Batch, Household, Individual, Program, SyncLog
 from country_workspace.utils.config import BatchNameConfig, ValidateModeConfig
 from country_workspace.utils.fields import clean_field_names, TO_UPPERCASE_FIELDS
 from country_workspace.utils.functional import compose
+from country_workspace.utils.sync_log import get_kobo_sync_log_name
 
 
 class Config(BatchNameConfig, ValidateModeConfig):
@@ -149,18 +152,51 @@ def set_roles_and_relationships(household: Household, individuals: list[Individu
 
 
 def import_asset(batch: Batch, asset: Asset, config: Config, id_generator: Callable[[], int]) -> ImportResult:
+    from django.db import transaction
+
     household_counter = 0
     individual_counter = 0
+    sync_log_name = get_kobo_sync_log_name(asset.uid)
+
+    program_ct = ContentType.objects.get_for_model(Program)
+    sync_log = SyncLog.objects.filter(name=sync_log_name, content_type=program_ct, object_id=batch.program.id).first()
+    last_id = int(sync_log.last_id) if sync_log and sync_log.last_id else 0
+
+    last_successful_id = last_id
+    current_submission = None
+
+    try:
+        for submission in asset.submissions(min_id=last_id):
+            current_submission = submission
 
-    submission_ids = set(KoboSubmission.objects.filter(asset_uid=asset.uid).values_list("submission_id", flat=True))
-    for submission in asset.submissions:
-        if submission.id in submission_ids:
-            continue
-        household = create_household(batch, submission, config, id_generator)
-        household_counter += 1
-        individuals = create_individuals(batch, household, submission, config)
-        individual_counter += len(individuals)
-        set_roles_and_relationships(household, individuals)
+            with transaction.atomic():
+                household = create_household(batch, submission, config, id_generator)
+                individuals = create_individuals(batch, household, submission, config)
+                set_roles_and_relationships(household, individuals)
+
+                household_counter += 1
+                individual_counter += len(individuals)
+
+            last_successful_id = submission.id
+
+    except Exception as e:
+        failed_id = current_submission.id if current_submission else "unknown (before first submission)"
+
+        error_msg = (
+            f"Successfully imported {household_counter} households, before stopping at submission {failed_id} due to:"
+            f"Error: {e}"
+            f"Last successful submission ID: {last_successful_id}."
+        )
+        raise ImportError(error_msg) from e
+
+    finally:
+        if last_successful_id > last_id:
+            SyncLog.objects.update_or_create(
+                name=sync_log_name,
+                content_type=program_ct,
+                object_id=batch.program.id,
+                defaults={"last_id": str(last_successful_id), "last_update_date": timezone.now()},
+            )
 
     return ImportResult(households=household_counter, individuals=individual_counter)