Merge branch 'main' into HEA-752/Dagster-GraphQL-API-is-intermittently-failing-with-a-ProtocolError-when-accessed-via-the-revproxy-Django-view

Author: girumb

pipelines/assets/baseline.py

@@ -78,17 +78,15 @@ def get_wealth_group_dataframe(
         # In the Summary columns in the Data, Data2, Data3 worksheets, the Wealth
         # Group Category is in Row 4 (District)rather than Row 3 (Wealth Group Category)
         # so do a second lookup to update the blank rows.
-        # If this doesn't find any new values, then it's because in a WB worksheet
-        # there are no extra Wealth Group Categories on Row 4
-        try:
+        # Note that in a WB worksheet there are no extra Wealth Group Categories on Row 4
+        if worksheet_name != "WB":
             wealth_group_df = wealthgroupcategorylookup.do_lookup(
                 wealth_group_df, "district", "wealth_group_category", update=True
             )
             # Remove the duplicate wealth_group_category_original column created by the second do_lookup(),
             # which otherwise causes problems when trying to merge dataframes, e.g. when building the wealth_group_df.
             wealth_group_df = wealth_group_df.loc[:, ~wealth_group_df.columns.duplicated()]
-        except ValueError:
-            pass
+
         # Check if there are unrecognized wealth group categories and report
         wealth_group_missing_category_df = wealth_group_df[
             wealth_group_df["wealth_group_category"].isnull()
@@ -266,7 +264,7 @@ def baseline_instances(
     }
 
     try:
-        preview = json.dumps(result, indent=4)
+        preview = json.dumps(result, indent=4, ensure_ascii=False)
     except TypeError as e:
         raise ValueError("Cannot serialize Community fixture to JSON. Failing dict is\n %s" % result) from e
 
@@ -359,7 +357,7 @@ def community_instances(context: AssetExecutionContext, config: BSSMetadataConfi
     result = {"Community": community_df.to_dict(orient="records")}
 
     try:
-        preview = json.dumps(result, indent=4)
+        preview = json.dumps(result, indent=4, ensure_ascii=False)
     except TypeError as e:
         raise ValueError("Cannot serialize Community fixture to JSON. Failing dict is\n %s" % result) from e
 

pipelines/assets/fixtures.py

@@ -220,7 +220,7 @@ def validate_instances(
 
     metadata = {f"num_{key.lower()}": len(value) for key, value in instances.items()}
     metadata["total_instances"] = sum(len(value) for value in instances.values())
-    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(instances, indent=4)}\n```")
+    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(instances, indent=4, ensure_ascii=False)}\n```")
     return instances, metadata
 
 
@@ -287,7 +287,7 @@ def get_fixture_from_instances(instance_dict: dict[str, list[dict]]) -> tuple[li
             metadata[f'num_{str(model._meta).split(".")[-1]}'] += 1
 
     metadata["total_instances"] = len(fixture)
-    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(fixture, indent=4)}\n```")
+    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(fixture, indent=4, ensure_ascii=False)}\n```")
     return fixture, metadata
 
 
@@ -300,7 +300,7 @@ def import_fixture(fixture: list[dict]) -> dict:
     # We need to use a .verbose_json file extension for Django to use the correct serializer.
     with tempfile.NamedTemporaryFile(mode="w+", suffix=".verbose_json") as f:
         # Write the fixture to a temporary file so that Django can access it
-        f.write(json.dumps(fixture))
+        f.write(json.dumps(fixture, indent=4, ensure_ascii=False))
         f.seek(0)
         call_command(verbose_load_data.Command(), f.name, verbosity=2, format="verbose_json", stdout=output_buffer)
 
@@ -309,7 +309,7 @@ def import_fixture(fixture: list[dict]) -> dict:
     for instance in fixture:
         metadata[f'num_{instance["model"].split(".")[-1]}'] += 1
     metadata["total_instances"] = len(fixture)
-    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(fixture, indent=4)}\n```")
+    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(fixture, indent=4, ensure_ascii=False)}\n```")
     metadata["output"] = MetadataValue.md(f"```\n{output_buffer.getvalue()}\n```")
     return metadata
 

pipelines/assets/livelihood_activity.py

@@ -1061,7 +1061,7 @@ def get_instances_from_dataframe(
             )
             * 100
         ),
-        "preview": MetadataValue.md(f"```json\n{json.dumps(result, indent=4)}\n```"),
+        "preview": MetadataValue.md(f"```json\n{json.dumps(result, indent=4, ensure_ascii=False)}\n```"),
     }
     if not unrecognized_labels.empty:
         metadata["unrecognized_labels"] = MetadataValue.md(unrecognized_labels.to_markdown(index=False))
@@ -1116,7 +1116,9 @@ def livelihood_activity_valid_instances(
     valid_instances, metadata = validate_instances(context, livelihood_activity_instances, partition_key)
     metadata = {f"num_{key.lower()}": len(value) for key, value in valid_instances.items()}
     metadata["total_instances"] = sum(len(value) for value in valid_instances.values())
-    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(valid_instances, indent=4)}\n```")
+    metadata["preview"] = MetadataValue.md(
+        f"```json\n{json.dumps(valid_instances, indent=4, ensure_ascii=False)}\n```"
+    )
     return Output(
         valid_instances,
         metadata=metadata,

pipelines/assets/other_cash_income.py

@@ -171,7 +171,9 @@ def other_cash_income_valid_instances(
     valid_instances, metadata = validate_instances(context, other_cash_income_instances, partition_key)
     metadata = {f"num_{key.lower()}": len(value) for key, value in valid_instances.items()}
     metadata["total_instances"] = sum(len(value) for value in valid_instances.values())
-    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(valid_instances, indent=4)}\n```")
+    metadata["preview"] = MetadataValue.md(
+        f"```json\n{json.dumps(valid_instances, indent=4, ensure_ascii=False)}\n```"
+    )
     return Output(
         valid_instances,
         metadata=metadata,

pipelines/assets/wealth_characteristic.py

@@ -450,7 +450,7 @@ def wealth_characteristic_instances(
             )
             * 100
         ),
-        "preview": MetadataValue.md(f"```json\n{json.dumps(result, indent=4)}\n```"),
+        "preview": MetadataValue.md(f"```json\n{json.dumps(result, indent=4, ensure_ascii=False)}\n```"),
     }
     if not unrecognized_labels.empty:
         metadata["unrecognized_labels"] = MetadataValue.md(unrecognized_labels.to_markdown(index=False))
@@ -473,7 +473,9 @@ def wealth_characteristic_valid_instances(
     valid_instances, metadata = validate_instances(context, wealth_characteristic_instances, partition_key)
     metadata = {f"num_{key.lower()}": len(value) for key, value in valid_instances.items()}
     metadata["total_instances"] = sum(len(value) for value in valid_instances.values())
-    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(valid_instances, indent=4)}\n```")
+    metadata["preview"] = MetadataValue.md(
+        f"```json\n{json.dumps(valid_instances, indent=4, ensure_ascii=False)}\n```"
+    )
     return Output(
         valid_instances,
         metadata=metadata,

pipelines/assets/wild_foods.py

@@ -181,7 +181,9 @@ def wild_foods_valid_instances(
     valid_instances, metadata = validate_instances(context, wild_foods_instances, partition_key)
     metadata = {f"num_{key.lower()}": len(value) for key, value in valid_instances.items()}
     metadata["total_instances"] = sum(len(value) for value in valid_instances.values())
-    metadata["preview"] = MetadataValue.md(f"```json\n{json.dumps(valid_instances, indent=4)}\n```")
+    metadata["preview"] = MetadataValue.md(
+        f"```json\n{json.dumps(valid_instances, indent=4, ensure_ascii=False)}\n```"
+    )
     return Output(
         valid_instances,
         metadata=metadata,

pipelines/resources.py

@@ -70,7 +70,7 @@ def dump_to_path(self, context: OutputContext, obj: Any, path: "UPath"):
             self.unlink(path)
 
         with path.open("w") as file:
-            file.write(json.dumps(obj, indent=4))
+            file.write(json.dumps(obj, indent=4, ensure_ascii=False))
 
     def load_from_path(self, context: InputContext, path: "UPath") -> Any:
         with path.open("r") as file:

pipelines/utils.py

@@ -88,18 +88,21 @@ def prepare_lookup(data: str | list[str] | pd.Series | pd.DataFrame) -> pd.Serie
     """
     Prepare a Series or DataFrame for lookup operations by converting to lowercase strings and stripping whitespace.
     """
-    if isinstance(data, str):
-        result = pd.DataFrame([data])
+    if isinstance(data, pd.DataFrame):
+        result = data
     elif isinstance(data, (list, pd.Series)):
         result = pd.DataFrame(data)
     else:
-        result = data
+        # Handle other types (like str, int, float)
+        result = pd.DataFrame([data])
+
     result = result.map(str).map(str.strip).map(str.lower).replace(r"\s+", " ", regex=True)
-    if isinstance(data, str):
-        result = result.iloc[0, 0]
+
+    if isinstance(data, pd.DataFrame):
+        return result
     elif isinstance(data, (list, pd.Series)):
-        result = result.iloc[:, 0]
-    return result
+        return result.iloc[:, 0]
+    return result.iloc[0, 0]
 
 
 def verbose_pivot(df: pd.DataFrame, values: str | list[str], index: str | list[str], columns: str | list[str]):

pipelines_tests/test_utils/__init__.py

@@ -0,0 +1,107 @@
+import pandas as pd
+from django.test import TestCase
+from pipelines.utils import prepare_lookup
+
+
+class PrepareLookupTestCase(TestCase):
+
+    def test_prepare_lookup_with_premitive_type_input(self):
+        # empty string
+        result = prepare_lookup("")
+        self.assertEqual(result, "")
+
+        # a simple string
+        result = prepare_lookup("nbr Mois")
+        self.assertEqual(result, "nbr mois")
+
+        # an int
+        result = prepare_lookup(0)
+        self.assertEqual(result, "0")
+
+        # a float
+        result = prepare_lookup(7.55)
+        self.assertEqual(result, "7.55")
+
+        # a simple string with spaces
+        result = prepare_lookup(" nbr Mois ")
+        self.assertEqual(result, "nbr mois")
+
+        # string with multiple internal spaces
+        result = prepare_lookup("Autre    revenu  (ex. crédit)")
+        self.assertEqual(result, "autre revenu (ex. crédit)")
+
+    def test_prepare_lookup_with_list_input(self):
+        # list with single element
+        result = prepare_lookup(["water"])
+        self.assertIsInstance(result, pd.Series)
+        self.assertEqual(result[0], "water")
+
+        # list with multiple elements
+        result = prepare_lookup(["Water", "inputs", "Social serv."])
+        self.assertIsInstance(result, pd.Series)
+        pd.testing.assert_series_equal(result, pd.Series(["water", "inputs", "social serv."], name=0))
+
+    def test_prepare_lookup_with_series_input(self):
+        # simple series
+        data = pd.Series(["Camel number owned", "Cattle number owned"])
+        result = prepare_lookup(data)
+        self.assertIsInstance(result, pd.Series)
+        pd.testing.assert_series_equal(result, pd.Series(["camel number owned", "cattle number owned"], name=0))
+
+        # test with irrigular spaces in elements
+        data = pd.Series(["Camel number   owned ", "  cattle number  Owned"])
+        result = prepare_lookup(data)
+        pd.testing.assert_series_equal(result, pd.Series(["camel number owned", "cattle number owned"], name=0))
+
+        # test with numeric elemnts
+        data = pd.Series([123, 456])
+        result = prepare_lookup(data)
+        pd.testing.assert_series_equal(result, pd.Series(["123", "456"], name=0))
+
+    def test_prepare_lookup_with_dataframe_input(self):
+        # single column dataframe
+        data = pd.DataFrame({"lables": ["Livestock products"]})
+        result = prepare_lookup(data)
+        self.assertIsInstance(result, pd.DataFrame)
+        pd.testing.assert_frame_equal(result, pd.DataFrame({"lables": ["livestock products"]}))
+
+        # multiple columns dataframe
+        data = pd.DataFrame({"lables": ["Livestock products"], "another": ["Payment  in kind  "]})
+        result = prepare_lookup(data)
+        assert isinstance(result, pd.DataFrame)
+        expected = pd.DataFrame({"lables": ["livestock products"], "another": ["payment in kind"]})
+        pd.testing.assert_frame_equal(result, expected)
+
+        # numeric values
+        data = pd.DataFrame({"column1": [123, 456], "column2": [78.9, 1011.12]})
+        result = prepare_lookup(data)
+        expected = pd.DataFrame({"column1": ["123", "456"], "column2": ["78.9", "1011.12"]})
+        pd.testing.assert_frame_equal(result, expected)
+
+        # empty df
+        data = pd.DataFrame()
+        result = prepare_lookup(data)
+        self.assertIsInstance(result, pd.DataFrame)
+        self.assertTrue(result.empty)
+
+        # test that datafarme preserves structure
+        data = pd.DataFrame(
+            {
+                "label": ["Cowpeas: kg produced", "Sorghum: kg produced"],
+                "product": ["Cowpeas", "Sorghum"],
+                "unit": ["kg", "kg"],
+            }
+        )
+        result = prepare_lookup(data)
+        self.assertEqual(result.shape, data.shape)
+        self.assertEqual(list(result.columns), list(data.columns))
+
+    def test_prepare_lookup_with_special_characters(self):
+        result = prepare_lookup("Autre nourriture: Poisson 2(sec)!@#$%")
+        self.assertEqual(result, "autre nourriture: poisson 2(sec)!@#$%")
+        # with tabs
+        result = prepare_lookup("Autre nourriture: \tPoisson")
+        self.assertEqual(result, "autre nourriture: poisson")
+        # some unicode characters
+        result = prepare_lookup("Revenu (Espèces)")
+        self.assertEqual(result, "revenu (espèces)")