gdacs: pass file as input in transformer

Author: Rup-Narayan-Rajbanshi

apps/etl/extraction/sources/gdacs/extract.py

@@ -147,7 +147,7 @@ def handle_extract(self):
     @app.task(
         bind=True,
         base=RetryableTask,
-        rate_limit="100/m",
+        rate_limit="50/m",
     )
     def task(celery_task, extraction_id) -> int:
         return GdacsExtraction(celery_task, extraction_id).handle()

apps/etl/transform/sources/gdacs.py

@@ -1,4 +1,3 @@
-import json
 import logging
 
 from pystac_monty.sources.gdacs import (
@@ -8,6 +7,7 @@
 )
 
 from apps.etl.transform.sources.handler import BaseTransformerHandler
+from apps.etl.utils import write_into_temp_file
 from main.celery import app
 
 logger = logging.getLogger(__name__)
@@ -21,23 +21,34 @@ class GDACSTransformHandler(BaseTransformerHandler[GDACSTransformer, GDACSDataSo
 
     @classmethod
     def get_schema_data(cls, extraction_object):
-        data = extraction_object.resp_data.read()
+        with extraction_object.resp_data.open("rb") as f:
+            file_content = f.read()
+        data_file = write_into_temp_file(file_content)
+
         episodes = []
         event_objects = extraction_object.child_extractions.all()
         for episode_obj in event_objects:
             episodes_data_dict = {}
-            event_episode_data = episode_obj.resp_data.read()
-            episodes_data_dict[GDACSDataSourceType.EVENT] = (episode_obj.url, json.loads(event_episode_data))
+
+            with episode_obj.resp_data.open("rb") as f:
+                file_content = f.read()
+            episode_data_temp_file = write_into_temp_file(file_content)
+
+            episodes_data_dict[GDACSDataSourceType.EVENT] = (episode_obj.url, episode_data_temp_file.name)
             geometry_objects = episode_obj.child_extractions.all()
 
             for geometry_detail in geometry_objects:
-                geometry_episode_data = geometry_detail.resp_data.read()
-                episodes_data_dict[GDACSDataSourceType.GEOMETRY] = (geometry_detail.url, json.loads(geometry_episode_data))
+                with geometry_detail.resp_data.open("rb") as f:
+                    file_content = f.read()
+                geometry_detail_temp_file = write_into_temp_file(file_content)
+
+                episodes_data_dict[GDACSDataSourceType.GEOMETRY] = (geometry_detail.url, geometry_detail_temp_file.name)
 
             episodes.append(episodes_data_dict)
-        return cls.transformer_schema(source_url=extraction_object.url, data=json.loads(data), episodes=episodes)
+
+        return cls.transformer_schema(source_url=extraction_object.url, data=data_file.name, episodes=episodes)
 
     @staticmethod
-    @app.task
+    @app.task(rate_limit="50/m")
     def task(extraction_id):
         GDACSTransformHandler().handle_transformation(extraction_id)

apps/etl/utils.py

@@ -1,6 +1,7 @@
 import datetime
 import logging
 import os
+import tempfile
 from pathlib import Path
 
 import pandas as pd
@@ -12,6 +13,13 @@
 logger = logging.getLogger(__name__)
 
 
+def write_into_temp_file(file_content):
+    tmp_file = tempfile.NamedTemporaryFile(suffix=".json", delete=False)
+    tmp_file.write(file_content)
+    tmp_file.close()
+    return tmp_file
+
+
 def get_cluster_codes():
     hazard_profiles_path = Path("./libs/pystac-monty/pystac_monty/HazardProfiles.csv")
     if os.path.exists(hazard_profiles_path):