Fix bugs and code quality issues found during audit

pykobo/form.py

@@ -1,5 +1,3 @@
-from typing import Union
-
 import numpy as np
 import pandas as pd
 import requests
@@ -30,7 +28,7 @@ def __init__(self, uid: str) -> None:
     def __repr__(self):
         return f"KoboForm('{self.uid}')"
 
-    def fetch_data(self) -> Union[pd.DataFrame, dict]:
+    def fetch_data(self) -> None:
         """Fetch the form's data and store them as a Pandas DF in the attribute `data`.
         If the form has repeat groups, extract them as separate DFs"""
 
@@ -45,9 +43,10 @@ def fetch_data(self) -> Union[pd.DataFrame, dict]:
         # Fetch the data
         res = requests.get(url=self.url_data, headers=self.headers)
 
-        # If error while fetching the data, return an empty DF
+        # If error while fetching the data, store an empty DF and return
         if res.status_code != 200:
-            return pd.DataFrame()
+            self.data = pd.DataFrame()
+            return
 
         data = res.json()["results"]
 
@@ -168,9 +167,11 @@ def fetch_media(self):
         # Request media and extract dataframe
         res = requests.get(url=media_url, headers=self.headers)
         media = res.json()["results"]
-        if not media.empty:
-            media[["hash", "filename", "mimetype"]] = pd.json_normalize(media.metadata)
         self.media = pd.DataFrame(media)
+        if not self.media.empty:
+            self.media[["hash", "filename", "mimetype"]] = pd.json_normalize(
+                self.media["metadata"]
+            )
 
     def fetch_attachments(self, media_columns: list):
         """
@@ -187,7 +188,7 @@ def _get_survey(self) -> None:
         of the repeat groups if any) as a list of `Question` objects. Each `Question` object has a name
         and a label so it's possible to display the data using any of the two."""
 
-        if not self.__asset:
+        if self.__asset is None:
             self._fetch_asset()
 
         if "naming_conflicts" in self.__asset["summary"]:
@@ -344,8 +345,9 @@ def _change_choices(
 
     def _fetch_asset(self):
         res = requests.get(url=self.url_asset, headers=self.headers)
-        self.__asset = res.json()
-        self.__content = res.json()["content"]
+        asset = res.json()
+        self.__asset = asset
+        self.__content = asset["content"]
 
     def _extract_from_asset(self, asset: dict) -> None:
         self.metadata["uid"] = asset["uid"]
@@ -495,7 +497,7 @@ def _split_gps_coords(self) -> None:
 
                         for idx, c in enumerate(new_geo_names):
                             q = Question(c, "geo", new_geo_labels[idx])
-                            repeat["columns"].append(index_geo + idx + 1, q)
+                            repeat["columns"].insert(index_geo + idx + 1, q)
 
                         self.data[new_geo_names] = self.repeats[repeat_name][
                             g.name

pykobo/manager.py

@@ -27,7 +27,7 @@ def api_version(self, value):
             raise ValueError("The value of 'api_version' has to be: 2.")
         self._api_version = value
 
-    def _fetch_forms(self) -> None:
+    def _fetch_forms(self) -> list:
         """Fetch the list of forms the user has access to with its token."""
         url_assets = f"{self.url}/api/v{self.api_version}/assets.json"
 
@@ -57,7 +57,7 @@ def _create_koboform(self, form: dict) -> KoboForm:
         return kform
 
     def get_forms(self) -> list:
-        if not self._assets:
+        if self._assets is None:
             self._assets = self._fetch_forms()
 
         kforms = []
@@ -67,7 +67,7 @@ def get_forms(self) -> list:
         return kforms
 
     def get_form(self, uid: str) -> Union[KoboForm, None]:
-        if not self._assets:
+        if self._assets is None:
             self._assets = self._fetch_forms()
 
         # If no forms
@@ -91,8 +91,8 @@ def redeploy_form(self, uid: str) -> None:
     def upload_media_from_local(
         self, uid: str, folder_path: str, file_name: str, rewrite: bool = False
     ) -> None:
-        file_extension = os.path.splitext(file_name)[1]
-        valid_media = [".jpeg", ".jpg", ".png", ".csv", ".JPGE", ".JPG", ".PNG"]
+        file_extension = os.path.splitext(file_name)[1].lower()
+        valid_media = [".jpeg", ".jpg", ".png", ".csv"]
 
         if not folder_path.endswith(("/", "\\")):
             folder_path += "/"
@@ -108,7 +108,8 @@ def upload_media_from_local(
         if not os.path.exists(file_path):
             raise FileNotFoundError(f"File not found: {file_path}")
 
-        self._upload_media(uid, open(file_path, "rb"), file_name, rewrite)
+        with open(file_path, "rb") as fh:
+            self._upload_media(uid, fh, file_name, rewrite)
 
     def upload_media_from_server(
         self, uid: str, media_data: bytes, file_name: str, rewrite: bool = False
@@ -135,11 +136,12 @@ def _upload_media(
             if each["metadata"]["filename"] == file_name:
                 if rewrite:
                     del_id = each["uid"]
-                    res.status_code = 403
-                    while res.status_code != 204:
+                    while True:
                         res = requests.delete(
                             f"{url_media}/{del_id}", headers=self.headers
                         )
+                        if res.status_code == 204:
+                            break
                         time.sleep(1)
                     break
                 else:

pykobo/utility.py

@@ -83,8 +83,10 @@ def reconcile_columns(
 def reconcile_columns_append(
     df: pd.DataFrame, col1: str, col2: str, new_column: str, to_replace: str
 ) -> None:
-    """TODO
-    XXX: See if can merge this function and the one above into 1
+    """
+    Given a DF, replace occurrences of `to_replace` in `col1` with the
+    corresponding value from `col2`, then merge both into `new_column`
+    and drop the originals.
     """
 
     # If the column is not of type str we convert it

tests/test_form.py

@@ -248,7 +248,7 @@ def test_extract_repeats_strips_group_prefix():
 
 
 def test_fetch_data_api_error(monkeypatch):
-    """A non-200 response from the data endpoint should return an empty DataFrame."""
+    """A non-200 response from the data endpoint stores an empty DataFrame in self.data."""
     f = _make_form()
     responses = [MockResponse(_ASSET, 200), MockResponse({}, 500)]
     idx = [0]
@@ -259,10 +259,10 @@ def mock_get(*args, **kwargs):
         return r
 
     monkeypatch.setattr(requests, "get", mock_get)
-    result = f.fetch_data()
+    f.fetch_data()
 
-    assert isinstance(result, pd.DataFrame)
-    assert len(result) == 0
+    assert isinstance(f.data, pd.DataFrame)
+    assert len(f.data) == 0
 
 
 def test_fetch_data_columns_and_index(monkeypatch):