Allow user to set alerts_provider; don't override existing provider (#133)

* Provide alerts provider

* Adapt tests

Author: rudokemper

f/connectors/alerts/alerts_gcs.py

@@ -33,6 +33,7 @@
 def main(
     gcp_service_acct: gcp_service_account,
     alerts_bucket: str,
+    alerts_provider: str,
     territory_id: int,
     db: postgresql,
     db_table_name: str,
@@ -49,6 +50,7 @@ def main(
     return _main(
         storage_client,
         alerts_bucket,
+        alerts_provider,
         territory_id,
         db,
         db_table_name,
@@ -59,6 +61,7 @@ def main(
 def _main(
     storage_client: gcs.Client,
     alerts_bucket: str,
+    alerts_provider: str,
     territory_id: int,
     db: postgresql,
     db_table_name: str,
@@ -71,6 +74,8 @@ def _main(
     storage_client : google.cloud.storage.Client
     alerts_bucket : str
         The name of the GCS bucket containing alerts.
+    alerts_provider : str
+        The name of the alerts provider.
     territory_id : int
         The ID of the territory for which alerts are being processed.
     db : postgresql
@@ -97,10 +102,12 @@ def _main(
     convert_tiffs_to_jpg(tiff_files)
 
     prepared_alerts_metadata, alerts_statistics = prepare_alerts_metadata(
-        alerts_metadata, territory_id
+        alerts_metadata, territory_id, alerts_provider
     )
 
-    prepared_alerts_data = prepare_alerts_data(destination_path, geojson_files)
+    prepared_alerts_data = prepare_alerts_data(
+        destination_path, geojson_files, alerts_provider
+    )
 
     logger.info(f"Writing alerts to the database table [{db_table_name}].")
     alerts_writer = StructuredDBWriter(
@@ -317,7 +324,7 @@ def convert_tiffs_to_jpg(tiff_files):
     logger.info("Successfully converted TIFF files to JPEG.")
 
 
-def prepare_alerts_metadata(alerts_metadata, territory_id):
+def prepare_alerts_metadata(alerts_metadata, territory_id, alerts_provider):
     """
     Prepare alerts metadata by filtering and processing CSV data.
 
@@ -335,6 +342,7 @@ def prepare_alerts_metadata(alerts_metadata, territory_id):
         CSV data as a string containing alerts metadata.
     territory_id : int
         The identifier for the territory used to filter the metadata.
+    alerts_provider : str
 
     Returns
     -------
@@ -377,9 +385,7 @@ def prepare_alerts_metadata(alerts_metadata, territory_id):
         index=False,
     )
 
-    # TODO: Currently, this script is only used for Terras alerts. Let's discuss a more sustainable approach with the alerts provider(s).
-    # Also, if this changes for future alerts, we will need to ensure that existing records are not overwritten.
-    filtered_df["data_source"] = "terras"
+    filtered_df["data_source"] = alerts_provider
 
     # Replace all NaN values with None
     filtered_df.replace({float("nan"): None}, inplace=True)
@@ -419,7 +425,7 @@ def prepare_alerts_metadata(alerts_metadata, territory_id):
     return prepared_alerts_metadata, alerts_statistics
 
 
-def prepare_alerts_data(local_directory, geojson_files):
+def prepare_alerts_data(local_directory, geojson_files, alerts_provider):
     """
     Prepare alerts data by reading GeoJSON files from a local directory.
 
@@ -432,7 +438,8 @@ def prepare_alerts_data(local_directory, geojson_files):
         The local directory where GeoJSON files are stored.
     geojson_files : list of str
         A list of GeoJSON file names to be processed.
-
+    alerts_provider : str
+        The name of the alerts provider.
     Returns
     -------
     list of dict
@@ -485,7 +492,7 @@ def prepare_alerts_data(local_directory, geojson_files):
                     "g__type": geom.get("type"),
                     "g__coordinates": json.dumps(geom.get("coordinates")),
                     # Metadata
-                    "data_source": "terras",
+                    "data_source": alerts_provider,
                     "source_file_name": file_path,
                 }
             )

f/connectors/alerts/alerts_gcs.script.yaml

@@ -9,6 +9,7 @@ schema:
   order:
     - gcp_service_acct
     - alerts_bucket
+    - alerts_provider
     - territory_id
     - db
     - db_table_name
@@ -19,6 +20,11 @@ schema:
       description: The name of the Google Cloud Storage bucket where the alerts are stored.
       default: null
       originalType: string
+    alerts_provider:
+      type: string
+      description: The name of the alerts provider.
+      default: null
+      originalType: string
     db:
       type: object
       description: The database connection parameters for storing tabular data.
@@ -56,6 +62,7 @@ schema:
   required:
     - gcp_service_acct
     - alerts_bucket
+    - alerts_provider
     - territory_id
     - db
     - db_table_name

f/connectors/alerts/tests/alerts_gcs_test.py

@@ -26,7 +26,7 @@ def test_prepare_alerts_metadata():
     alerts_metadata = pd.read_csv(alerts_history_csv).to_csv(index=False)
 
     prepared_alerts_metadata, alert_statistics = prepare_alerts_metadata(
-        alerts_metadata, 100
+        alerts_metadata, 100, "test_provider"
     )
 
     # Check that alerts statistics is the latest month and year in the CSV
@@ -39,8 +39,8 @@ def test_metadata_id_stability():
     alerts_history_csv = Path(assets_directory, "alerts_history.csv")
     alerts_metadata = pd.read_csv(alerts_history_csv).to_csv(index=False)
 
-    first, _ = prepare_alerts_metadata(alerts_metadata, 100)
-    second, _ = prepare_alerts_metadata(alerts_metadata, 100)
+    first, _ = prepare_alerts_metadata(alerts_metadata, 100, "test_provider")
+    second, _ = prepare_alerts_metadata(alerts_metadata, 100, "test_provider")
 
     # Order shouldn't matter but just to be safe, sort by _id
     first_ids = sorted(r["_id"] for r in first)
@@ -53,7 +53,7 @@ def test_metadata_id_stability():
 def test_alert_id_generation(tmp_path):
     file_path = Path(assets_directory, "alert_202309900112345671.geojson")
     geojson_files = [str(file_path)]
-    prepared = prepare_alerts_data(tmp_path, geojson_files)
+    prepared = prepare_alerts_data(tmp_path, geojson_files, "test_provider")
 
     assert len(prepared) > 0
     for row in prepared:
@@ -99,6 +99,7 @@ def test_script_e2e(pg_database, mock_alerts_storage_client, tmp_path):
     alerts_metadata = _main(
         mock_alerts_storage_client,
         MOCK_BUCKET_NAME,
+        "test_provider",
         100,
         pg_database,
         "fake_alerts",
@@ -248,6 +249,7 @@ def test_script_e2e(pg_database, mock_alerts_storage_client, tmp_path):
     alerts_metadata = _main(
         mock_alerts_storage_client,
         MOCK_BUCKET_NAME,
+        "test_provider",
         100,
         pg_database,
         "fake_alerts",
@@ -264,6 +266,7 @@ def test_file_update_logic(pg_database, mock_alerts_storage_client, tmp_path):
     _main(
         mock_alerts_storage_client,
         MOCK_BUCKET_NAME,
+        "test_provider",
         100,
         pg_database,
         "fake_alerts",
@@ -278,6 +281,7 @@ def test_file_update_logic(pg_database, mock_alerts_storage_client, tmp_path):
     _main(
         mock_alerts_storage_client,
         MOCK_BUCKET_NAME,
+        "test_provider",
         100,
         pg_database,
         "fake_alerts",
@@ -304,6 +308,7 @@ def test_file_update_logic(pg_database, mock_alerts_storage_client, tmp_path):
     _main(
         mock_alerts_storage_client,
         MOCK_BUCKET_NAME,
+        "test_provider",
         100,
         pg_database,
         "fake_alerts",

f/connectors/alerts_download_post_notify.flow/flow.yaml

@@ -11,6 +11,9 @@ value:
           alerts_bucket:
             type: javascript
             expr: flow_input.alerts_bucket
+          alerts_provider:
+            type: javascript
+            expr: flow_input.alerts_provider
           db:
             type: javascript
             expr: flow_input.db
@@ -87,6 +90,11 @@ schema:
       description: The name of the Google Cloud Storage bucket where the alerts are stored.
       default: ''
       originalType: string
+    alerts_provider:
+      type: string
+      description: The name of the alerts provider.
+      default: null
+      originalType: string
     comapeo:
       type: object
       description: >
@@ -151,6 +159,7 @@ schema:
   required:
     - gcp_service_acct
     - alerts_bucket
+    - alerts_provider
     - territory_id
     - db
     - db_table_name