Merge pull request #179 from FEWS-NET/HEA-572/relax_data_completeness_constraint_4

Hea 572/relax data completeness constraint

Author: rhunwicks

pipelines/assets/fixtures.py

@@ -387,6 +387,7 @@ def get_all_label_attributes(labels: pd.Series, activity_type: str, country_code
 
 def get_instances_from_dataframe(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     df: pd.DataFrame,
     livelihood_zone_baseline: LivelihoodZoneBaseline,
     activity_type: str,
@@ -727,8 +728,9 @@ def get_instances_from_dataframe(
                     for i, livelihood_activity in enumerate(livelihood_activities_for_strategy):
                         livelihood_activity["livelihood_strategy"] = livelihood_zone_baseline_key + [
                             livelihood_strategy["strategy_type"],
-                            livelihood_strategy["season"] if livelihood_strategy["season"] else "",
-                            livelihood_strategy["product_id"] if livelihood_strategy["product_id"] else "",
+                            livelihood_strategy["season"] or "",  # Natural key components must be "" rather than None
+                            livelihood_strategy["product_id"]
+                            or "",  # Natural key components must be "" rather than None
                             livelihood_strategy["additional_identifier"],
                         ]
 
@@ -1149,13 +1151,6 @@ def get_instances_from_dataframe(
                     % (partition_key, worksheet_name, row, label)
                 ) from e
 
-    raise_errors = True
-    if errors and raise_errors:
-        errors = "\n".join(errors)
-        raise RuntimeError(
-            "Missing or inconsistent metadata in BSS %s worksheet '%s':\n%s" % (partition_key, worksheet_name, errors)
-        )
-
     result = {
         "LivelihoodStrategy": livelihood_strategies,
         "LivelihoodActivity": livelihood_activities,
@@ -1177,6 +1172,19 @@ def get_instances_from_dataframe(
     if not unrecognized_labels.empty:
         metadata["unrecognized_labels"] = MetadataValue.md(unrecognized_labels.to_markdown(index=False))
 
+    if errors:
+        if config.strict:
+            raise RuntimeError(
+                "Missing or inconsistent metadata in BSS %s worksheet '%s':\n%s"
+                % (partition_key, worksheet_name, "\n".join(errors))
+            )
+        else:
+            context.log.error(
+                "Missing or inconsistent metadata in BSS %s worksheet '%s':\n%s"
+                % (partition_key, worksheet_name, "\n".join(errors))
+            )
+            metadata["errors"] = MetadataValue.md(f'```text\n{"\n".join(errors)}\n```')
+
     return Output(
         result,
         metadata=metadata,
@@ -1185,6 +1193,7 @@ def get_instances_from_dataframe(
 
 def get_annotated_instances_from_dataframe(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     livelihood_activity_dataframe: pd.DataFrame,
     livelihood_summary_dataframe: pd.DataFrame,
     activity_type: str,
@@ -1203,6 +1212,7 @@ def get_annotated_instances_from_dataframe(
     # Get the detail LivelihoodStrategy and LivelihoodActivity instances
     output = get_instances_from_dataframe(
         context,
+        config,
         livelihood_activity_dataframe,
         livelihood_zone_baseline,
         activity_type,
@@ -1214,6 +1224,7 @@ def get_annotated_instances_from_dataframe(
         # Get the summary instances
         reported_summary_output = get_instances_from_dataframe(
             context,
+            config,
             livelihood_summary_dataframe,
             livelihood_zone_baseline,
             ActivityLabel.LivelihoodActivityType.LIVELIHOOD_SUMMARY,
@@ -1325,7 +1336,9 @@ def get_annotated_instances_from_dataframe(
             summary_df.replace(pd.NA, None).to_markdown(floatfmt=",.0f")
         )
 
-        # Move the preview and metadata item to the end of the dict
+        # Move the preview and errors metadata item to the end of the dict
+        if "errors" in output.metadata:
+            output.metadata["errors"] = output.metadata.pop("errors")
         output.metadata["preview"] = output.metadata.pop("preview")
 
     return output
@@ -1334,26 +1347,27 @@ def get_annotated_instances_from_dataframe(
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
 def livelihood_activity_instances(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     livelihood_activity_dataframe: pd.DataFrame,
     livelihood_summary_dataframe: pd.DataFrame,
 ) -> Output[dict]:
     """
     LivelhoodStrategy and LivelihoodActivity instances extracted from the BSS.
     """
-    output = get_annotated_instances_from_dataframe(
+    return get_annotated_instances_from_dataframe(
         context,
+        config,
         livelihood_activity_dataframe,
         livelihood_summary_dataframe,
-        ActivityLabel.LivelihoodActivityType.LIVELIHOOD_SUMMARY,
+        ActivityLabel.LivelihoodActivityType.LIVELIHOOD_ACTIVITY,
         len(HEADER_ROWS),
     )
 
-    return output
-
 
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
 def livelihood_activity_valid_instances(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     livelihood_activity_instances: dict,
     wealth_characteristic_instances: dict,
 ) -> Output[dict]:
@@ -1369,16 +1383,7 @@ def livelihood_activity_valid_instances(
             **{"WealthGroup": wealth_characteristic_instances["WealthGroup"]},
             **livelihood_activity_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, ensure_ascii=False)}\n```"
-    )
-    return Output(
-        valid_instances,
-        metadata=metadata,
-    )
+    return validate_instances(context, config, livelihood_activity_instances, partition_key)
 
 
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
@@ -1390,11 +1395,7 @@ def livelihood_activity_fixture(
     """
     Django fixture for the Livelihood Activities from a BSS.
     """
-    fixture, metadata = get_fixture_from_instances(livelihood_activity_valid_instances)
-    return Output(
-        fixture,
-        metadata=metadata,
-    )
+    return get_fixture_from_instances(livelihood_activity_valid_instances)
 
 
 @asset(partitions_def=bss_instances_partitions_def)
@@ -1405,8 +1406,4 @@ def imported_livelihood_activities(
     """
     Imported Django fixtures for a BSS, added to the Django database.
     """
-    metadata = import_fixture(livelihood_activity_fixture)
-    return Output(
-        None,
-        metadata=metadata,
-    )
+    return import_fixture(livelihood_activity_fixture)

pipelines/assets/livelihood_activity.py

@@ -38,12 +38,11 @@
 | 32 | income                                                 |        |        |        |           |          |          |           |          |          |
 """  # NOQA: E501
 
-import json
 import os
 
 import django
 import pandas as pd
-from dagster import AssetExecutionContext, MetadataValue, Output, asset
+from dagster import AssetExecutionContext, Output, asset
 
 from ..configs import BSSMetadataConfig
 from ..partitions import bss_instances_partitions_def
@@ -127,29 +126,30 @@ def summary_other_cash_income_labels_dataframe(
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
 def other_cash_income_instances(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     other_cash_income_dataframe: pd.DataFrame,
     livelihood_summary_dataframe: pd.DataFrame,
 ) -> Output[dict]:
     """
     LivelhoodStrategy and LivelihoodActivity instances extracted from the BSS.
     """
     if other_cash_income_dataframe.empty:
-        output = {}
+        return Output({}, metadata={"message": "No Data2 worksheet found in this BSS"})
 
-    output = get_annotated_instances_from_dataframe(
+    return get_annotated_instances_from_dataframe(
         context,
+        config,
         other_cash_income_dataframe,
         livelihood_summary_dataframe,
         ActivityLabel.LivelihoodActivityType.OTHER_CASH_INCOME,
         len(HEADER_ROWS),
     )
 
-    return output
-
 
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
 def other_cash_income_valid_instances(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     other_cash_income_instances: dict,
     wealth_characteristic_instances: dict,
 ) -> Output[dict]:
@@ -165,16 +165,7 @@ def other_cash_income_valid_instances(
             **{"WealthGroup": wealth_characteristic_instances["WealthGroup"]},
             **other_cash_income_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, ensure_ascii=False)}\n```"
-    )
-    return Output(
-        valid_instances,
-        metadata=metadata,
-    )
+    return validate_instances(context, config, other_cash_income_instances, partition_key)
 
 
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
@@ -186,11 +177,7 @@ def other_cash_income_fixture(
     """
     Django fixture for the Livelihood Activities from a BSS.
     """
-    fixture, metadata = get_fixture_from_instances(other_cash_income_valid_instances)
-    return Output(
-        fixture,
-        metadata=metadata,
-    )
+    return get_fixture_from_instances(other_cash_income_valid_instances)
 
 
 @asset(partitions_def=bss_instances_partitions_def)
@@ -201,8 +188,4 @@ def imported_other_cash_income_activities(
     """
     Imported Django fixtures for a BSS, added to the Django database.
     """
-    metadata = import_fixture(other_cash_income_fixture)
-    return Output(
-        None,
-        metadata=metadata,
-    )
+    return import_fixture(other_cash_income_fixture)

pipelines/assets/other_cash_income.py

@@ -524,13 +524,14 @@ def wealth_characteristic_instances(
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
 def wealth_characteristic_valid_instances(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     wealth_characteristic_instances,
 ) -> Output[dict]:
     """
     Valid  WealthGroup and WealthGroupCharacteristicValue instances from a BSS, ready to be loaded via a Django fixture.
     """
     partition_key = context.asset_partition_key_for_output()
-    valid_instances, metadata = validate_instances(context, wealth_characteristic_instances, partition_key)
+    valid_instances, metadata = validate_instances(context, config, 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(

pipelines/assets/wealth_characteristic.py

@@ -56,12 +56,11 @@
     | 85 | TOTAL FISHING KCALS (%)                 | 0.009088932377 | 0.005577299413 | 0      | 0.009639776763 | 0.01133165595 | 0        | 0         | 0.009708632311 | 0        |
 """  # NOQA: E501
 
-import json
 import os
 
 import django
 import pandas as pd
-from dagster import AssetExecutionContext, MetadataValue, Output, asset
+from dagster import AssetExecutionContext, Output, asset
 
 from ..configs import BSSMetadataConfig
 from ..partitions import bss_instances_partitions_def
@@ -137,29 +136,30 @@ def summary_wild_foods_labels_dataframe(
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
 def wild_foods_instances(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     wild_foods_dataframe: pd.DataFrame,
     livelihood_summary_dataframe: pd.DataFrame,
 ) -> Output[dict]:
     """
     LivelhoodStrategy and LivelihoodActivity instances extracted from the BSS.
     """
     if wild_foods_dataframe.empty:
-        output = {}
+        return Output({}, metadata={"message": "No Data3 worksheet found in this BSS"})
 
-    output = get_annotated_instances_from_dataframe(
+    return get_annotated_instances_from_dataframe(
         context,
+        config,
         wild_foods_dataframe,
         livelihood_summary_dataframe,
         ActivityLabel.LivelihoodActivityType.WILD_FOODS,
         len(HEADER_ROWS),
     )
 
-    return output
-
 
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
 def wild_foods_valid_instances(
     context: AssetExecutionContext,
+    config: BSSMetadataConfig,
     wild_foods_instances: dict,
     wealth_characteristic_instances: dict,
 ) -> Output[dict]:
@@ -175,16 +175,7 @@ def wild_foods_valid_instances(
             **{"WealthGroup": wealth_characteristic_instances["WealthGroup"]},
             **wild_foods_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, ensure_ascii=False)}\n```"
-    )
-    return Output(
-        valid_instances,
-        metadata=metadata,
-    )
+    return validate_instances(context, config, wild_foods_instances, partition_key)
 
 
 @asset(partitions_def=bss_instances_partitions_def, io_manager_key="json_io_manager")
@@ -196,11 +187,7 @@ def wild_foods_fixture(
     """
     Django fixture for the Livelihood Activities from a BSS.
     """
-    fixture, metadata = get_fixture_from_instances(wild_foods_valid_instances)
-    return Output(
-        fixture,
-        metadata=metadata,
-    )
+    return get_fixture_from_instances(wild_foods_valid_instances)
 
 
 @asset(partitions_def=bss_instances_partitions_def)
@@ -211,8 +198,4 @@ def imported_wild_foods_activities(
     """
     Imported Django fixtures for a BSS, added to the Django database.
     """
-    metadata = import_fixture(wild_foods_fixture)
-    return Output(
-        None,
-        metadata=metadata,
-    )
+    return import_fixture(wild_foods_fixture)

pipelines/assets/wild_foods.py

@@ -1,6 +1,13 @@
+[project]
+name = "hea-database-development"
+version = "0.1.0"
+description = "The HEA Database manages HEA Baseline data."
+readme = "README.md"
+requires-python = ">=3.12"
+
 [tool.ruff]
 line-length = 119
-target-version = 'py310'
+target-version = 'py312'
 exclude = [
   '.eggs',        # exclude a few common directories in the
   '.git',         # root of the project
@@ -27,7 +34,7 @@ docstring-quotes = "double"
 
 [tool.black]
 line-length = 119
-target-version = ['py310']
+target-version = ['py312']
 include = '\.pyi?$'
 exclude = '''