Feature: Import data from kobo

src/country_workspace/admin/__init__.py

@@ -16,6 +16,7 @@
 from .role import UserRoleAdmin  # noqa
 from .sync import SyncLog  # noqa
 from .user import UserAdmin  # noqa
+from .kobo import KoboAsset  # noqa
 
 site.register(ContentType, admin_class=ContentTypeAdmin)
 site.register(Permission, admin_class=PermissionAdmin)

src/country_workspace/admin/job.py

@@ -1,12 +1,11 @@
 from typing import Optional, Sequence
 
+from adminfilters.autocomplete import AutoCompleteFilter, LinkedAutoCompleteFilter
 from django.contrib import admin
 from django.http import HttpRequest
-
-from adminfilters.autocomplete import AutoCompleteFilter, LinkedAutoCompleteFilter
 from django_celery_boost.admin import CeleryTaskModelAdmin
 
-from ..models import AsyncJob
+from ..models import AsyncJob, KoboSyncJob
 from .base import BaseModelAdmin
 from .filters import FailedFilter
 
@@ -25,5 +24,9 @@ class AsyncJobAdmin(CeleryTaskModelAdmin, BaseModelAdmin):
 
     def get_readonly_fields(self, request: "HttpRequest", obj: "Optional[AsyncJob]" = None) -> Sequence[str]:
         if obj:
-            return ("program", "batch", "owner", "local_status", "type", "action", "sentry_id")
+            return "program", "batch", "owner", "local_status", "type", "action", "sentry_id"
         return super().get_readonly_fields(request, obj)
+
+@admin.register(KoboSyncJob)
+class KoboSyncJobAdmin(CeleryTaskModelAdmin, BaseModelAdmin):
+    pass

src/country_workspace/admin/kobo.py

@@ -0,0 +1,37 @@
+from typing import Any
+
+from django.contrib import admin
+
+from .base import BaseModelAdmin
+from ..models import KoboAsset
+from ..models.kobo import KoboQuestion, KoboSubmission
+
+
+class ReadOnlyInlineAdmin(admin.TabularInline):
+    can_create = False
+    can_change = False
+    can_delete = False
+    extra = 0
+
+class KoboQuestionAdmin(ReadOnlyInlineAdmin):
+    model = KoboQuestion
+
+
+class KoboSubmissionAdmin(ReadOnlyInlineAdmin):
+    model = KoboSubmission
+
+
+@admin.register(KoboAsset)
+class KoboAssetAdmin(BaseModelAdmin):
+    list_display = ("uid", "name")
+    exclude = ("programs",)
+    inlines = (KoboQuestionAdmin, KoboSubmissionAdmin)
+
+    def has_add_permission(self, *args: Any, **kwargs: Any) -> bool:
+        return False
+
+    def has_change_permission(self, *args: Any, **kwargs: Any) -> bool:
+        return False
+
+    def has_delete_permission(self, *args: Any, **kwargs: Any) -> bool:
+        return False

src/country_workspace/admin/program.py

@@ -9,26 +9,33 @@
 
 from ..cache.manager import cache_manager
 from ..compat.admin_extra_buttons import confirm_action
-from ..models import Program
+from ..models import Program, KoboAsset
 from .base import BaseModelAdmin
 
 if TYPE_CHECKING:
     from admin_extra_buttons.buttons import LinkButton
 
 
+class KoboAssetInline(admin.TabularInline):
+    model = KoboAsset.programs.through
+    extra = 1
+
+
 @admin.register(Program)
 class ProgramAdmin(BaseModelAdmin):
     list_display = ("name", "sector", "status", "active")
     search_fields = ("name",)
     list_filter = (("country_office", AutoCompleteFilter), "status", "active", "sector")
     ordering = ("name",)
     autocomplete_fields = ("country_office",)
+    inlines = (KoboAssetInline,)
 
     @button()
     def invalidate_cache(self, request: HttpRequest, pk: str) -> None:
         obj: [Program] = Program.objects.select_related("country_office").get(pk=pk)
         cache_manager.incr_cache_version(program=obj)
 
+
     @link(change_list=False)
     def view_in_workspace(self, btn: "LinkButton") -> None:
         obj = btn.context["original"]

src/country_workspace/config/__init__.py

@@ -214,6 +214,8 @@ class Group(Enum):
     "AZURE_CLIENT_KEY": (str, "", "", False, "Azure client key for SSO"),
     # "AZURE_CONNECTION_STRING": (str, ""),
     # "CV2DNN_PATH": (str, ""),
+    "KOBO_BASE_URL": (str, "", "", False, "Kobo API base URL"),
+    "KOBO_TOKEN": (str, "", "", False, "Kobo API token"),
 }
 
 env = SmartEnv(**CONFIG)

src/country_workspace/config/fragments/kobo.py

@@ -0,0 +1,4 @@
+from .. import env
+
+KOBO_BASE_URL = env("KOBO_BASE_URL")
+KOBO_TOKEN = env("KOBO_TOKEN")

src/country_workspace/config/settings.py

@@ -242,6 +242,7 @@
 from .fragments.rest_framework import *  # noqa
 from .fragments.root import *  # noqa
 from .fragments.sentry import *  # noqa
+from .fragments.kobo import *  # noqa
 
 # from .fragments.smart_admin import *  # noqa
 from .fragments.social_auth import *  # noqa

src/country_workspace/contrib/kobo/auth.py

@@ -0,0 +1,14 @@
+from requests.auth import AuthBase
+from requests.models import PreparedRequest
+
+
+TOKEN = "Token"
+AUTHORIZATION = 'Authorization'
+
+class Auth(AuthBase):
+    def __init__(self, api_key: str) -> None:
+        self._auth_header = f"{TOKEN} {api_key}"
+
+    def __call__(self, request: PreparedRequest) -> PreparedRequest:
+        request.headers[AUTHORIZATION] = self._auth_header
+        return request

src/country_workspace/contrib/kobo/client.py

@@ -0,0 +1,76 @@
+from collections.abc import Generator, Callable
+from typing import cast
+
+from black.linegen import partial
+from requests import Session, Response
+
+from country_workspace.contrib.kobo.auth import Auth
+from country_workspace.contrib.kobo.data import Submission, Asset, Question
+from country_workspace.contrib.kobo.raw.common import ListResponse
+from country_workspace.contrib.kobo.raw import asset as raw_asset, asset_list as raw_asset_list, common as raw_common
+from country_workspace.contrib.kobo.raw import submission_list as raw_submission_list
+
+
+DataGetter = Callable[[str], Response]
+
+
+def _handle_paginated_response[T, U](data_getter: DataGetter,
+                                     url: str,
+                                     collection_mapper: Callable[[ListResponse], list[T]],
+                                     item_mapper: Callable[[T], U]) -> Generator[U, None, None]:
+    while url:
+        response = data_getter(url)
+        response.raise_for_status()
+        data: ListResponse = response.json()
+        yield from map(item_mapper, collection_mapper(data))
+        url = data["next"]
+
+
+def _get_raw_asset_list(data: raw_common.ListResponse) -> list[raw_asset_list.Asset]:
+    return [
+        datum for datum in
+        cast(raw_asset_list.AssetList, data)["results"]
+        if datum["asset_type"] == "survey"
+    ]
+
+
+def _get_raw_submission_list(data: raw_common.ListResponse) -> list[raw_submission_list.Submission]:
+    return cast(raw_submission_list.SubmissionList, data)["results"]
+
+
+def _get_asset_list(data_getter: DataGetter, url: str) -> Generator[Asset, None, 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, questions: list[Question]) -> Generator[Submission, None, None]:
+    return _handle_paginated_response(
+        data_getter,
+        url,
+        _get_raw_submission_list,
+        partial(Submission, data_getter, questions)
+    )
+
+
+def _get_asset(data_getter: DataGetter, raw: raw_asset_list.Asset) -> Asset:
+    response = data_getter(raw["url"])
+    response.raise_for_status()
+    data: raw_asset.Asset = response.json()
+    return Asset(data, partial(_get_submission_list, data_getter, raw["data"]))
+
+
+def _get_submission() -> Submission:
+    pass
+
+
+class Client:
+    def __init__(self, *, base_url: str, token: str) -> None:
+        self.base_url = base_url
+        session = Session()
+        session.auth = Auth(token)
+        self.data_getter: DataGetter = session.get
+
+    @property
+    def assets(self) -> Generator[Asset, None, None]:
+        yield from _get_asset_list(self.data_getter, f"{self.base_url}/api/v2/assets.json")

src/country_workspace/contrib/kobo/data.py

@@ -0,0 +1,134 @@
+from base64 import b64encode
+from collections import UserDict
+from collections.abc import Callable, Generator
+from enum import StrEnum, auto
+from functools import cached_property, reduce
+from requests import Response
+from typing import Any
+from uuid import UUID
+
+from country_workspace.contrib.kobo.raw import asset as raw_asset
+from country_workspace.contrib.kobo.raw import submission_list as raw_submission_list
+
+
+class SurveyItemType(StrEnum):
+    START = auto()
+    END = auto()
+    BEGIN_GROUP = auto()
+    END_GROUP = auto()
+    BEGIN_REPEAT = auto()
+    END_REPEAT = auto()
+
+
+
+class Raw[T]:
+    def __init__(self, raw: T) -> None:
+        self._raw = raw
+
+
+class Question(Raw[raw_asset.SurveyItem]):
+    def __init__(self, raw: raw_asset.SurveyItem, in_group: bool, in_roster: bool) -> None:
+        super().__init__(raw)
+        assert not (in_group and in_roster), "Cannot be both in group and roster"
+        self._in_group = in_group
+        self._in_roster = in_roster
+
+    def extract_answer(self, in_: raw_submission_list.Submission, out: dict[str, Any]) -> None:
+        if self._in_roster:
+            roster_key, _ = self.key.split("/")
+            roster = out.get(roster_key, [])
+            if self.key in in_:
+                if roster:
+                    roster[0][self.key] = in_.get(self.key)
+                else:
+                    roster.append({self.key: in_.get(self.key)})
+            else:
+                if roster_key in in_:
+                    for i, item in enumerate(in_[roster_key]):
+                        if len(roster) < i + 1:
+                            roster.append({})
+                        roster[i][self.key] = item[self.key]
+            out[roster_key] = roster
+        else:
+            out[self.key] = in_.get(self.key)
+
+    @property
+    def key(self) -> str:
+        return self._raw["$xpath"]
+
+    @property
+    def labels(self) -> list[str]:
+        return self._raw["label"]
+
+
+InAndOut = tuple[raw_submission_list.Submission, dict[str, Any]]
+
+
+def _extract_answer(in_and_out: InAndOut, question: Question) -> InAndOut:
+    question.extract_answer(*in_and_out)
+    return in_and_out
+
+
+def _download_attachments(data_getter: Callable[[str], Response], raw: raw_submission_list.Submission) -> None:
+    for attachment in raw["_attachments"]:
+        content = b64encode(data_getter(attachment["download_url"]).content).decode()
+        value = f"data:{attachment['mimetype']};base64,{content}"
+        key = attachment["question_xpath"]
+        if key in raw:
+            raw[key] = value
+        else:
+            parent, key = key.split("/")
+            parent, index = parent.split("[")
+            index = int(index.rstrip("]")) - 1
+            raw[parent][index][f"{parent}/{key}"] = value
+
+
+class Submission(Raw[raw_submission_list.Submission], UserDict):
+    def __init__(self, data_getter: Callable[[str], Response], questions: list[Question], raw: raw_submission_list.Submission) -> None:
+        Raw.__init__(self, raw)
+        _download_attachments(data_getter, self._raw)
+        _, answers = reduce(_extract_answer, questions, (raw, {}))
+        UserDict.__init__(self, answers)
+
+    @property
+    def uuid(self) -> UUID:
+        return UUID(self._raw["_uuid"])
+
+
+class Asset(Raw[raw_asset.Asset]):
+    def __init__(self, raw: raw_asset.Asset, submissions: Callable[[list[Question]], Generator[Submission, None, None]]) -> None:
+        super().__init__(raw)
+        self._submissions = submissions
+
+    @property
+    def uid(self) -> str:
+        return self._raw["uid"]
+
+    @property
+    def name(self) -> str:
+        return self._raw["name"]
+
+    @cached_property
+    def questions(self) -> list[Question]:
+        in_group = False
+        in_roster = False
+        questions = []
+        for raw_question in self._raw["content"]["survey"]:
+            match raw_question["type"]:
+                case SurveyItemType.START | SurveyItemType.END:
+                    pass
+                case SurveyItemType.BEGIN_GROUP:
+                    in_group = True
+                case SurveyItemType.END_GROUP:
+                    in_group = False
+                case SurveyItemType.BEGIN_REPEAT:
+                    in_roster = True
+                case SurveyItemType.END_REPEAT:
+                    in_roster = False
+                case _:
+                    questions.append(Question(raw_question, in_group, in_roster))
+        return questions
+
+    @property
+    def submissions(self) -> Generator[Submission, None, None]:
+        yield from self._submissions(self.questions)

src/country_workspace/contrib/kobo/raw/asset.py

@@ -0,0 +1,14 @@
+from typing import TypedDict
+
+class SurveyItem(TypedDict("SurveyItem", {"$xpath": str})):
+    type: str
+
+
+class Content(TypedDict):
+    survey: list[SurveyItem]
+
+
+class Asset(TypedDict):
+    content: Content
+    name: str
+    uid: str

src/country_workspace/contrib/kobo/raw/asset_list.py

@@ -0,0 +1,13 @@
+from typing import TypedDict
+
+from country_workspace.contrib.kobo.raw.common import ListResponse
+
+
+class Asset(TypedDict):
+    data: str
+    url: str
+    asset_type: str
+
+
+class AssetList(ListResponse):
+    results: list[Asset]

src/country_workspace/contrib/kobo/raw/common.py

@@ -0,0 +1,7 @@
+from typing import TypedDict
+
+
+class ListResponse(TypedDict):
+    count: int
+    next: str | None
+    previous: str | None