Fix state issue regarding Emdat extraaction

- Re-format graphql query
- Define query variables for emdat and glide
- Simplify logic for glide task

Author: tnagorra

apps/etl/etl_tasks/emdat.py

@@ -3,92 +3,102 @@
 from celery import chain, shared_task
 from django.conf import settings
 
-from apps.etl.extraction.sources.emdat.extract import EMDATExtraction
+from apps.etl.extraction.sources.emdat.extract import EMDATExtraction, EMDATQueryVars
 from apps.etl.transform.sources.emdat import EMDATTransformHandler
 
 QUERY = """
-    query monty ($limit: Int, $offset: Int, $include_hist: Boolean, $from: Int, $to: Int) {
-      api_version
-      public_emdat(
-        cursor: {
-            offset: $offset,
-            limit: $limit
-        }
-        filters: {
-           include_hist: $include_hist
-           from: $from
-           to: $to
-     }
-      ) {
+query monty(
+    $limit: Int
+    $offset: Int
+    $include_hist: Boolean
+    $from: Int
+    $to: Int
+) {
+    api_version
+    public_emdat(
+        cursor: { offset: $offset, limit: $limit }
+        filters: { include_hist: $include_hist, from: $from, to: $to }
+    ) {
         total_available
         info {
-          timestamp
-          filters
-          cursor
-          version
+            timestamp
+            filters
+            cursor
+            version
         }
         data {
-          disno
-          classif_key
-          group
-          subgroup
-          type
-          subtype
-          external_ids
-          name
-          iso
-          country
-          subregion
-          region
-          location
-          origin
-          associated_types
-          ofda_response
-          appeal
-          declaration
-          aid_contribution
-          magnitude
-          magnitude_scale
-          latitude
-          longitude
-          river_basin
-          start_year
-          start_month
-          start_day
-          end_year
-          end_month
-          end_day
-          total_deaths
-          no_injured
-          no_affected
-          no_homeless
-          total_affected
-          reconstr_dam
-          reconstr_dam_adj
-          insur_dam
-          insur_dam_adj
-          total_dam
-          total_dam_adj
-          cpi
-          admin_units
-          entry_date
-          last_update
+            disno
+            classif_key
+            group
+            subgroup
+            type
+            subtype
+            external_ids
+            name
+            iso
+            country
+            subregion
+            region
+            location
+            origin
+            associated_types
+            ofda_response
+            appeal
+            declaration
+            aid_contribution
+            magnitude
+            magnitude_scale
+            latitude
+            longitude
+            river_basin
+            start_year
+            start_month
+            start_day
+            end_year
+            end_month
+            end_day
+            total_deaths
+            no_injured
+            no_affected
+            no_homeless
+            total_affected
+            reconstr_dam
+            reconstr_dam_adj
+            insur_dam
+            insur_dam_adj
+            total_dam
+            total_dam_adj
+            cpi
+            admin_units
+            entry_date
+            last_update
         }
-      }
     }
-    """
+}
+"""
 
 
 @shared_task
 def ext_and_transform_emdat_latest_data(**kwargs):
-    to_year = datetime.now().year
-    from_year = int(settings.EMDAT_START_YEAR)
-    variables = {"limit": -1, "from": from_year, "to": to_year}
+    # FIXME: Why are we getting data from settings.EMDAT_START_YEAR to get the latest data?
+    # Also, the filtering only filters using year so we might have lot of duplicate data
+    variables: EMDATQueryVars = {
+        "limit": -1,
+        "from": int(settings.EMDAT_START_YEAR),
+        "to": datetime.now().year,
+        "include_hist": None,
+    }
 
     chain(EMDATExtraction.task.s(QUERY, variables), EMDATTransformHandler.task.s()).apply_async()
 
 
 @shared_task
 def ext_and_transform_emdat_historical_data(**kwargs):
-    variables = {"limit": -1, "include_hist": True}
+    variables: EMDATQueryVars = {
+        "limit": -1,
+        "from": None,
+        "to": None,
+        "include_hist": True,
+    }
+
     chain(EMDATExtraction.task.s(QUERY, variables), EMDATTransformHandler.task.s()).apply_async()

apps/etl/etl_tasks/glide.py

@@ -3,41 +3,41 @@
 from celery import chain, shared_task
 from django.conf import settings
 
-from apps.etl.extraction.sources.glide.extract import GlideExtraction
+from apps.etl.extraction.sources.glide.extract import GlideExtraction, GlideQueryVars
 from apps.etl.models import ExtractionData, HazardType
 from apps.etl.transform.sources.glide import GlideTransformHandler
 
 GLIDE_HAZARDS = [
-    ("EQ", HazardType.EARTHQUAKE),
-    ("TC", HazardType.CYCLONE),
-    ("FL", HazardType.FLOOD),
-    ("DR", HazardType.DROUGHT),
-    ("WF", HazardType.WILDFIRE),
-    ("VO", HazardType.VOLCANO),
-    ("TS", HazardType.TSUNAMI),
-    ("CW", HazardType.COLDWAVE),
-    ("EP", HazardType.EPIDEMIC),
-    ("EC", HazardType.EXTRATROPICAL_CYCLONE),
-    ("ET", HazardType.EXTREME_TEMPERATURE),
-    ("FR", HazardType.FIRE),
-    ("FF", HazardType.FLASH_FLOOD),
-    ("HT", HazardType.HEAT_WAVE),
-    ("IN", HazardType.INSECT_INFESTATION),
-    ("LS", HazardType.LANDSLIDE),
-    ("MS", HazardType.MUD_SLIDE),
-    ("ST", HazardType.SEVERE_LOCAL_STROM),
-    ("SL", HazardType.SLIDE),
-    ("AV", HazardType.SNOW_AVALANCHE),
-    ("SS", HazardType.STORM),
-    ("AC", HazardType.TECH_DISASTER),
-    ("TO", HazardType.TORNADO),
-    ("VW", HazardType.VIOLENT_WIND),
-    ("WV", HazardType.WAVE_SURGE),
+    HazardType.EARTHQUAKE,
+    HazardType.FLOOD,
+    HazardType.CYCLONE,
+    HazardType.EPIDEMIC,
+    HazardType.STORM,
+    HazardType.DROUGHT,
+    HazardType.TSUNAMI,
+    HazardType.WILDFIRE,
+    HazardType.VOLCANO,
+    HazardType.COLDWAVE,
+    HazardType.EXTRATROPICAL_CYCLONE,
+    HazardType.EXTREME_TEMPERATURE,
+    HazardType.FIRE,
+    HazardType.FLASH_FLOOD,
+    HazardType.HEAT_WAVE,
+    HazardType.INSECT_INFESTATION,
+    HazardType.LANDSLIDE,
+    HazardType.MUD_SLIDE,
+    HazardType.SEVERE_LOCAL_STROM,
+    HazardType.SLIDE,
+    HazardType.SNOW_AVALANCHE,
+    HazardType.TECH_DISASTER,
+    HazardType.TORNADO,
+    HazardType.VIOLENT_WIND,
+    HazardType.WAVE_SURGE,
 ]
 
 
 @shared_task
-def _ext_and_transform_glide_latest_data(hazard_type, hazard_type_str):
+def _ext_and_transform_glide_latest_data(hazard_type: HazardType):
     ext_object = (
         ExtractionData.objects.filter(
             source=ExtractionData.Source.GLIDE,
@@ -48,31 +48,55 @@ def _ext_and_transform_glide_latest_data(hazard_type, hazard_type_str):
         .order_by("-created_at")
         .first()
     )
+
     if ext_object:
         from_date = ext_object.created_at.date()
     else:
         from_date = datetime.strptime(settings.GLIDE_START_DATE, "%Y-%m-%d").date()
 
     to_date = datetime.today().date()
-    url = f"{settings.GLIDE_URL}/glide/jsonglideset.jsp?fromyear={from_date.year}&frommonth={from_date.month}&fromday={from_date.day}&toyear={to_date.year}&tomonth={to_date.month}&today={to_date.day}&events={hazard_type}"  # noqa: E501
 
-    chain(GlideExtraction.task.s(url), GlideTransformHandler.task.s()).apply_async()
+    # FIXME: Check if the date filters are inclusive
+    url = f"{settings.GLIDE_URL}/glide/jsonglideset.jsp"
+    variables: GlideQueryVars = {
+        "fromyear": from_date.year,
+        "frommonth": from_date.month,
+        "fromday": from_date.day,
+        "toyear": to_date.year,
+        "tomonth": to_date.month,
+        "today": to_date.day,
+        "events": hazard_type.value,
+    }
+
+    chain(GlideExtraction.task.s(url, variables), GlideTransformHandler.task.s()).apply_async()
 
 
 @shared_task
-def _ext_and_transform_glide_historical_data(hazard_type, hazard_type_str):
+def _ext_and_transform_glide_historical_data(hazard_type: HazardType):
     to_date = datetime.today().date()
-    url = f"{settings.GLIDE_URL}/glide/jsonglideset.jsp?toyear={to_date.year}&tomonth={to_date.month}&to_date={to_date.day}&events={hazard_type}"  # noqa: E501
-    chain(GlideExtraction.task.s(url), GlideTransformHandler.task.s()).apply_async()
+
+    # FIXME: Check if the date filters are inclusive
+    url = f"{settings.GLIDE_URL}/glide/jsonglideset.jsp"
+    variables: GlideQueryVars = {
+        "fromyear": None,
+        "frommonth": None,
+        "fromday": None,
+        "toyear": to_date.year,
+        "tomonth": to_date.month,
+        "today": to_date.day,
+        "events": hazard_type.value,
+    }
+
+    chain(GlideExtraction.task.s(url, variables), GlideTransformHandler.task.s()).apply_async()
 
 
 @shared_task
 def ext_and_transform_glide_latest_data():
-    for hazard_type, hazard_type_str in GLIDE_HAZARDS:
-        _ext_and_transform_glide_latest_data(hazard_type, hazard_type_str)
+    for hazard_type in GLIDE_HAZARDS:
+        _ext_and_transform_glide_latest_data(hazard_type)
 
 
 @shared_task
 def ext_and_transform_glide_historical_data():
-    for hazard_type, hazard_type_str in GLIDE_HAZARDS:
-        _ext_and_transform_glide_historical_data(hazard_type, hazard_type_str)
+    for hazard_type in GLIDE_HAZARDS:
+        _ext_and_transform_glide_historical_data(hazard_type)

apps/etl/extraction/sources/base/handler.py

@@ -27,7 +27,7 @@ def store_extraction_data(
         validate_source_func: Callable[[Any], None],
         source: int,
         response: dict,
-        instance_id: int = None,
+        instance_id: int | None = None,
     ):
         """
         Save extracted data into data base. Checks for duplicate conent using hashing.
@@ -72,14 +72,15 @@ def _create_extraction_instance(cls, url: str, source: int) -> ExtractionData:
             status=ExtractionData.Status.PENDING,
             source_validation_status=ExtractionData.ValidationStatus.NO_VALIDATION,
             trace_id=str(uuid.uuid4()),
+            # FIXME Pass hazard type
             hazard_type=None,
             attempt_no=0,
             resp_code=0,
         )
 
     @classmethod
     def _update_instance_status(
-        cls, instance: ExtractionData, status: int, validation_status: int = None, update_validation: bool = False
+        cls, instance: ExtractionData, status: int, validation_status: int | None = None, update_validation: bool = False
     ) -> None:
         """
         Update the status of the extraction instance.
@@ -116,7 +117,7 @@ def _save_response_data(cls, instance: ExtractionData, response: requests.Respon
         return json.loads(response.content)
 
     @classmethod
-    def handle_extraction(cls, url: str, params: dict, headers: dict, source: int) -> dict:
+    def handle_extraction(cls, url: str, params: dict | None, headers: dict, source: int) -> dict:
         """
         Process data extraction.
         Returns:

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

@@ -26,7 +26,7 @@ def store_extraction_data(
         instance_id: int = None,
     ):
         """
-        Save extracted data into database. Checks for duplicate content using hashing.
+        Save extracted data into database.
         """
         file_name = f"{source}.zip"
         resp_data = response
@@ -41,6 +41,7 @@ def store_extraction_data(
             # manage duplicate file content.
             manage_duplicate_file_content(
                 source=extraction_instance.source,
+                # FIXME: We need to calculate has for zip file
                 hash_content=None,
                 instance=extraction_instance,
                 response_data=resp_data.content,

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

@@ -71,4 +71,4 @@ def handle_extraction(cls, query: str, variables: EMDATQueryVars, source: int) -
     @staticmethod
     @app.task
     def task(query: str, variables: EMDATQueryVars):  # type: ignore[reportIncompatibleMethodOverride]
-        return EMDATExtraction().handle_extraction(query, variables, ExtractionData.Source.EMDAT)
+        return EMDATExtraction().handle_extraction(query, variables, ExtractionData.Source.EMDAT)

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

@@ -28,4 +28,4 @@ class GlideExtraction(BaseExtraction):
     @staticmethod
     @app.task
     def task(url: str, variables: GlideQueryVars):  # type: ignore[reportIncompatibleMethodOverride]
-        return GlideExtraction().handle_extraction(url, variables, HEADERS, ExtractionData.Source.GLIDE)
+        return GlideExtraction().handle_extraction(url, variables, HEADERS, ExtractionData.Source.GLIDE)

apps/etl/tasks.py

@@ -19,4 +19,4 @@ def extract_usgs_data():
 
 @shared_task
 def load_data():
-    call_command("load_data_to_stac")
+    call_command("load_data_to_stac")