Merge pull request #4507 from unicef/feature/RDI-Access-change-for-retrieving-data-from-Kobo-NEW

RDI Access change for retrieving data from Kobo Reinstated

Author: pavlo-mk

development_tools/.env.example

@@ -11,6 +11,7 @@ PGUSER=postgres
 POSTGRES_HOST_AUTH_METHOD=trust
 DATABASE_URL=postgis://postgres:postgres@db:5432/postgres
 REP_DATABASE_URL=postgis://postgres:postgres@db:5432/postgres
+POSTGRES_SSL_MODE=off
 EMAIL_HOST=TBD
 EMAIL_HOST_USER=TBD
 EMAIL_HOST_PASSWORD=TBD
@@ -39,6 +40,7 @@ USE_DUMMY_EXCHANGE_RATES=no
 OPENAPI_URL=127.0.0.1:8080/api/rest/
 
 KOBO_URL=https://kobo.humanitarianresponse.info
+KOBO_PROJECT_VIEWS_ID=
 #MAILJET_API_KEY=
 #MAILJET_SECRET_KEY=
 #CATCH_ALL_EMAIL=

src/hct_mis_api/apps/core/kobo/api.py

@@ -2,130 +2,100 @@
 import time
 import typing
 from io import BytesIO
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple
 from urllib.parse import urlparse
 
 from django.conf import settings
 
 import requests
+from requests import Response
 from requests.adapters import HTTPAdapter
 from requests.exceptions import RetryError
 from requests.packages.urllib3.util.retry import Retry
 
-from hct_mis_api.apps.core.kobo.common import filter_by_owner
-from hct_mis_api.apps.core.models import BusinessArea, XLSXKoboTemplate
+from hct_mis_api.apps.core.models import XLSXKoboTemplate
 from hct_mis_api.apps.utils.exceptions import log_and_raise
 
 logger = logging.getLogger(__name__)
 
 
-class TokenNotProvided(Exception):
-    pass
-
-
-class TokenInvalid(Exception):
+class CountryCodeNotProvided(Exception):
     pass
 
 
 class KoboRequestsSession(requests.Session):
-    AUTH_DOMAINS = [urlparse(settings.KOBO_URL).hostname]
+    AUTH_DOMAINS = [urlparse(settings.KOBO_URL).hostname, urlparse(settings.KOBO_URL).hostname]
 
     def should_strip_auth(self, old_url: str, new_url: str) -> bool:
         new_parsed = urlparse(new_url)
-        if new_parsed.hostname in KoboRequestsSession.AUTH_DOMAINS:
+        if new_parsed.hostname in KoboRequestsSession.AUTH_DOMAINS:  # pragma: no cover
             return False
-        return super().should_strip_auth(old_url, new_url)  # type: ignore # FIXME: Call to untyped function "should_strip_auth" in typed context
+        return super().should_strip_auth(
+            old_url, new_url
+        )  # type: ignore # FIXME: Call to untyped function "should_strip_auth" in typed context
 
 
 class KoboAPI:
-    def __init__(self, business_area_slug: Optional[str] = None):
-        if business_area_slug is not None:
-            self.business_area = BusinessArea.objects.get(slug=business_area_slug)
-            self.KPI_URL = self.business_area.kobo_url or settings.KOBO_URL
-        else:
-            self.business_area = None
-            self.KPI_URL = settings.KOBO_URL
+    LIMIT = 30_000
+    FORMAT = "json"
 
-        self._get_token()
+    def __init__(
+        self, kpi_url: Optional[str] = None, token: Optional[str] = None, project_views_id: Optional[str] = None
+    ) -> None:
+        self._kpi_url = kpi_url or settings.KOBO_URL
+        self._token = token or settings.KOBO_MASTER_API_TOKEN
+        self._project_views_id = project_views_id or settings.KOBO_PROJECT_VIEWS_ID
+
+        self._client = KoboRequestsSession()
+        self._set_token()
+
+    def _set_token(self) -> None:
+        retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504], allowed_methods=False)
+        self._client.mount(self._kpi_url, HTTPAdapter(max_retries=retries))
+        self._client.headers.update({"Authorization": f"token {self._token}"})
 
-    def _handle_paginated_results(self, url: str) -> List[Dict]:
+    def _get_paginated_request(self, url: str) -> List[Dict]:
         next_url = url
         results: List = []
 
-        # if there will be more than 30000 results,
-        # we need to make additional queries
         while next_url:
-            data = self._handle_request(next_url)
+            response = self._get_request(next_url)
+            data = response.json()
             next_url = data["next"]
             results.extend(data["results"])
         return results
 
-    def _get_url(
-        self,
-        endpoint: str,
-        append_api: bool = True,
-        add_limit: bool = True,
-        additional_query_params: Optional[Any] = None,
-    ) -> str:
-        endpoint.strip("/")
-        if endpoint != "token" and append_api is True:
-            endpoint = f"api/v2/{endpoint}"
-        # According to the Kobo API documentation,
-        # the maximum limit per page is 30000
-        query_params = f"format=json{'&limit=30000' if add_limit else ''}"
-        if additional_query_params is not None:
-            query_params += f"&{additional_query_params}"
-        return f"{self.KPI_URL}/{endpoint}?{query_params}"
-
-    def _get_token(self) -> None:
-        self._client = KoboRequestsSession()
-        retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504], allowed_methods=False)
-        self._client.mount(self.KPI_URL, HTTPAdapter(max_retries=retries))
-
-        if self.business_area is None:
-            token = settings.KOBO_MASTER_API_TOKEN
-        else:
-            token = self.business_area.kobo_token
-
-        if not token:
-            msg = f"KOBO Token is not set for business area {self.business_area}"
-            logger.warning(msg)
-            raise TokenNotProvided(msg)
-
-        self._client.headers.update({"Authorization": f"token {token}"})
-
-    def _handle_request(self, url: str) -> Dict:
+    def _get_request(self, url: str) -> Response:
         response = self._client.get(url=url)
         try:
             response.raise_for_status()
-        except requests.exceptions.HTTPError as e:
-            logger.warning(e)
+        except requests.exceptions.HTTPError as e:  # pragma: no cover
+            logger.exception(e)
             raise
-        return response.json()
+        return response
 
     def _post_request(
         self, url: str, data: Optional[Dict] = None, files: Optional[typing.IO] = None
-    ) -> requests.Response:
+    ) -> Response:  # pragma: no cover
         return self._client.post(url=url, data=data, files=files)
 
-    def _patch_request(
-        self, url: str, data: Optional[Dict] = None, files: Optional[typing.IO] = None
-    ) -> requests.Response:
-        return self._client.patch(url=url, data=data, files=files)
-
     def create_template_from_file(
-        self, bytes_io_file: Optional[typing.IO], xlsx_kobo_template_object: XLSXKoboTemplate, template_id: str = ""
-    ) -> Optional[Tuple[Dict, str]]:
-        data = {
-            "name": "Untitled",
-            "asset_type": "template",
-            "description": "",
-            "sector": "",
-            "country": "",
-            "share-metadata": False,
-        }
+        self, bytes_io_file: typing.IO, xlsx_kobo_template_object: XLSXKoboTemplate, template_id: str = ""
+    ) -> Optional[Tuple[Dict, str]]:  # pragma: no cover
+        # TODO: not sure if this actually works
         if not template_id:
-            asset_response = self._post_request(url=self._get_url("assets/", add_limit=False), data=data)
+            data = {
+                "name": "Untitled",
+                "asset_type": "template",
+                "description": "",
+                "sector": "",
+                "country": "",
+                "share-metadata": False,
+            }
+            endpoint = "api/v2/assets"
+            query_params = f"format={self.FORMAT}"
+            url = f"{self._kpi_url}/{endpoint}?{query_params}"
+            asset_response = self._post_request(url=url, data=data)
             try:
                 asset_response.raise_for_status()
             except requests.exceptions.HTTPError as e:
@@ -135,12 +105,13 @@ def create_template_from_file(
             asset_uid = asset_response_dict.get("uid")
         else:
             asset_uid = template_id
+
         file_import_data = {
             "assetUid": asset_uid,
-            "destination": self._get_url(f"assets/{asset_uid}/", append_api=False, add_limit=False),
+            "destination": f"{self._kpi_url}/assets/{asset_uid}?format={self.FORMAT}",
         }
         file_import_response = self._post_request(
-            url=self._get_url("imports/", append_api=False, add_limit=False),
+            url=f"{self._kpi_url}/imports?format={self.FORMAT}",
             data=file_import_data,
             files={"file": bytes_io_file},  # type: ignore # FIXME
         )
@@ -149,7 +120,8 @@ def create_template_from_file(
 
         attempts = 5
         while attempts >= 0:
-            response_dict = self._handle_request(url)
+            response = self._get_request(url)
+            response_dict = response.json()
             import_status = response_dict.get("status")
             if import_status == "processing":
                 xlsx_kobo_template_object.status = XLSXKoboTemplate.PROCESSING
@@ -162,36 +134,31 @@ def create_template_from_file(
         log_and_raise("Fetching import data took too long", error_type=RetryError)
         return None
 
-    def get_all_projects_data(self) -> List:
-        if not self.business_area:
-            logger.warning("Business area is not provided")
-            raise ValueError("Business area is not provided")
-        projects_url = self._get_url("assets/")
-
-        results = self._handle_paginated_results(projects_url)
-        return filter_by_owner(results, self.business_area)
+    def get_all_projects_data(self, country_code: str) -> List:
+        if not country_code:
+            raise CountryCodeNotProvided("No country code provided")
+        endpoint = f"api/v2/project-views/{self._project_views_id}/assets/"
+        query_params = f"format={self.FORMAT}&limit={self.LIMIT}"
+        query_params += f"&q=settings__country_codes__icontains:{country_code.upper()}"
+        url = f"{self._kpi_url}/{endpoint}?{query_params}"
+        return self._get_paginated_request(url)
 
     def get_single_project_data(self, uid: str) -> Dict:
-        projects_url = self._get_url(f"assets/{uid}")
-
-        return self._handle_request(projects_url)
+        endpoint = f"api/v2/assets/{uid}/"
+        query_params = f"format={self.FORMAT}&limit={self.LIMIT}"
+        url = f"{self._kpi_url}/{endpoint}?{query_params}"
+        response = self._get_request(url)
+        return response.json()
 
-    def get_project_submissions(self, uid: str, only_active_submissions: bool) -> List:
-        additional_query_params = None
+    def get_project_submissions(self, uid: str, only_active_submissions: bool) -> List[Dict]:
+        endpoint = f"api/v2/assets/{uid}/data/"
+        query_params = f"format={self.FORMAT}&limit={self.LIMIT}"
         if only_active_submissions:
             additional_query_params = 'query={"_validation_status.uid":"validation_status_approved"}'
-        submissions_url = self._get_url(
-            f"assets/{uid}/data/",
-            additional_query_params=additional_query_params,
-        )
-
-        return self._handle_paginated_results(submissions_url)
+            query_params += f"&{additional_query_params}"
+        url = f"{self._kpi_url}/{endpoint}?{query_params}"
+        return self._get_paginated_request(url)
 
-    def get_attached_file(self, url: str) -> BytesIO:
-        response = self._client.get(url=url)
-        try:
-            response.raise_for_status()
-        except requests.exceptions.HTTPError as e:
-            logger.warning(e)
-            raise
+    def get_attached_file(self, url: str) -> BytesIO:  # pragma: no cover
+        response = self._get_request(url)
         return BytesIO(response.content)

src/hct_mis_api/apps/core/kobo/common.py

@@ -111,13 +111,6 @@ def count_population(results: list, business_area: BusinessArea) -> tuple[int, i
     return total_households_count, total_individuals_count
 
 
-def filter_by_owner(data: List, business_area: BusinessArea) -> List:
-    kobo_username = business_area.kobo_username
-    if data:
-        return [element for element in data if element["owner__username"] == kobo_username]
-    return []
-
-
 def get_submission_metadata(household_data_dict: Dict) -> Dict:
     meta_fields_mapping = {
         "_uuid": "kobo_submission_uuid",

src/hct_mis_api/apps/core/models.py

@@ -98,7 +98,7 @@ class BusinessArea(NaturalKeyModel, TimeStampedUUIDModel):
     is_accountability_applicable = models.BooleanField(default=False)
     active = models.BooleanField(default=False)
     enable_email_notification = models.BooleanField(default=True, verbose_name="Automatic Email notifications enabled")
-
+    # TODO: deprecated to remove in the next release
     kobo_username = models.CharField(max_length=255, null=True, blank=True)
     kobo_token = models.CharField(max_length=255, null=True, blank=True)
     kobo_url = models.URLField(max_length=255, null=True, blank=True)

src/hct_mis_api/apps/core/schema.py

@@ -14,7 +14,7 @@
 
 from django.core.exceptions import ObjectDoesNotExist
 from django.db import models
-from django.db.models import Q
+from django.db.models import F, Q
 
 import graphene
 from constance import config
@@ -32,7 +32,7 @@
     TYPE_STRING,
     Scope,
 )
-from hct_mis_api.apps.core.kobo.api import KoboAPI
+from hct_mis_api.apps.core.kobo.api import CountryCodeNotProvided, KoboAPI
 from hct_mis_api.apps.core.kobo.common import reduce_asset, reduce_assets_list
 from hct_mis_api.apps.core.languages import Language, Languages
 from hct_mis_api.apps.core.models import (
@@ -313,11 +313,11 @@ def get_fields_attr_generators(
 
 def resolve_asset(business_area_slug: str, uid: str) -> Dict:
     try:
-        assets = KoboAPI(business_area_slug).get_single_project_data(uid)
+        assets = KoboAPI().get_single_project_data(uid)
     except ObjectDoesNotExist as e:
         logger.warning(f"Provided business area: {business_area_slug}, does not exist.")
         raise GraphQLError("Provided business area does not exist.") from e
-    except AttributeError as error:
+    except AttributeError as error:  # pragma: no cover
         logger.warning(error)
         raise GraphQLError(str(error)) from error
 
@@ -326,13 +326,18 @@ def resolve_asset(business_area_slug: str, uid: str) -> Dict:
 
 def resolve_assets_list(business_area_slug: str, only_deployed: bool = False) -> List:
     try:
-        assets = KoboAPI(business_area_slug).get_all_projects_data()
+        business_area = BusinessArea.objects.annotate(country_code=F("countries__iso_code3")).get(
+            slug=business_area_slug
+        )
+        assets = KoboAPI().get_all_projects_data(business_area.country_code)
     except ObjectDoesNotExist as e:
         logger.warning(f"Provided business area: {business_area_slug}, does not exist.")
         raise GraphQLError("Provided business area does not exist.") from e
-    except AttributeError as error:
+    except AttributeError as error:  # pragma: no cover
         logger.warning(error)
         raise GraphQLError(str(error)) from error
+    except CountryCodeNotProvided:
+        raise GraphQLError(f"Business area {business_area_slug} does not have a country code.")
 
     return reduce_assets_list(assets, only_deployed=only_deployed)
 

src/hct_mis_api/apps/registration_datahub/tasks/pull_kobo_submissions.py

@@ -22,11 +22,11 @@ class PullKoboSubmissions:
     def execute(self, kobo_import_data: KoboImportData, program: Program) -> Dict:
         kobo_import_data.status = KoboImportData.STATUS_RUNNING
         kobo_import_data.save()
-        kobo_api = KoboAPI(kobo_import_data.business_area_slug)
+        business_area = BusinessArea.objects.get(slug=kobo_import_data.business_area_slug)
+        kobo_api = KoboAPI()
         submissions = kobo_api.get_project_submissions(
             kobo_import_data.kobo_asset_id, kobo_import_data.only_active_submissions
         )
-        business_area = BusinessArea.objects.get(slug=kobo_import_data.business_area_slug)
         validator = KoboProjectImportDataInstanceValidator(program)
         skip_validate_pictures = kobo_import_data.pull_pictures is False
         validation_errors = validator.validate_everything(submissions, business_area, skip_validate_pictures)

src/hct_mis_api/apps/registration_datahub/tasks/rdi_kobo_create.py

@@ -98,7 +98,7 @@ def _handle_image_field(self, value: Any, is_flex_field: bool) -> Optional[Union
             return None
         current_download_url = attachment.get("download_url", "")
         download_url = current_download_url.replace("?format=json", "")
-        api = KoboAPI(self.business_area.slug)
+        api = KoboAPI()
         image_bytes = api.get_attached_file(download_url)
         file = File(image_bytes, name=value)
         if is_flex_field:

src/hct_mis_api/config/env.py

@@ -54,6 +54,7 @@
     "DEFAULT_EMAIL_DISPLAY": (str, ""),
     "KOBO_URL": (str, "https://kobo-hope-trn.unitst.org"),
     "KOBO_MASTER_API_TOKEN": (str, "KOBO_TOKEN"),
+    "KOBO_PROJECT_VIEWS_ID": (str, ""),
     "AZURE_CLIENT_ID": (str, ""),
     "AZURE_CLIENT_SECRET": (str, ""),
     "AZURE_TENANT_KEY": (str, ""),

src/hct_mis_api/config/fragments/kobo.py

@@ -2,3 +2,4 @@
 
 KOBO_URL = env("KOBO_URL")
 KOBO_MASTER_API_TOKEN = env("KOBO_MASTER_API_TOKEN")
+KOBO_PROJECT_VIEWS_ID = env("KOBO_PROJECT_VIEWS_ID")