Add class based etl for Glide.

Author: Rup-Narayan-Rajbanshi

apps/etl/etl_tasks/glide.py

@@ -1,61 +1,78 @@
+from datetime import datetime
+
 from celery import chain, shared_task
+from django.conf import settings
+
+from apps.etl.extraction.sources.glide.extract import GlideExtraction
+from apps.etl.models import ExtractionData, HazardType
+from apps.etl.transform.sources.glide import GlideTransformHandler
 
-from apps.etl.extraction.sources.glide.extract import (
-    extract_glide_historical_data,
-    extract_glide_latest_data,
-)
-from apps.etl.models import HazardType
-from apps.etl.transform.sources.glide import transform_glide_event_data
+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),
+]
 
 
 @shared_task
-def ext_and_transform_glide_historical_data(hazard_type: str, hazard_type_str: str, **kwargs):
-    event_workflow = chain(
-        extract_glide_historical_data.s(
+def _ext_and_transform_glide_latest_data(hazard_type, hazard_type_str):
+    ext_object = (
+        ExtractionData.objects.filter(
+            source=ExtractionData.Source.GLIDE,
             hazard_type=hazard_type,
-            hazard_type_str=hazard_type_str,
-        ),
-        transform_glide_event_data.s(),
+            status=ExtractionData.Status.SUCCESS,
+            resp_data__isnull=False,
+        )
+        .order_by("-created_at")
+        .first()
     )
-    event_workflow.apply_async()
+    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"https://www.glidenumber.net/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()
 
 
 @shared_task
-def ext_and_transform_data(hazard_type, hazard_type_str):
-    event_workflow = chain(
-        extract_glide_latest_data.s(
-            hazard_type=hazard_type,
-            hazard_type_str=hazard_type_str,
-        ),
-        transform_glide_event_data.s(),
-    )
-    event_workflow.apply_async()
+def _ext_and_transform_glide_historical_data(hazard_type, hazard_type_str):
+    to_date = datetime.today().date()
+    url = f"https://www.glidenumber.net/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()
 
 
 @shared_task
 def ext_and_transform_glide_latest_data():
-    ext_and_transform_data.delay("EQ", HazardType.EARTHQUAKE)
-    ext_and_transform_data.delay("TC", HazardType.CYCLONE)
-    ext_and_transform_data.delay("FL", HazardType.FLOOD)
-    ext_and_transform_data.delay("DR", HazardType.DROUGHT)
-    ext_and_transform_data.delay("WF", HazardType.WILDFIRE)
-    ext_and_transform_data.delay("VO", HazardType.VOLCANO)
-    ext_and_transform_data.delay("TS", HazardType.TSUNAMI)
-    ext_and_transform_data.delay("CW", HazardType.COLDWAVE)
-    ext_and_transform_data.delay("EP", HazardType.EPIDEMIC)
-    ext_and_transform_data.delay("EC", HazardType.EXTRATROPICAL_CYCLONE)
-    ext_and_transform_data.delay("ET", HazardType.EXTREME_TEMPERATURE)
-    ext_and_transform_data.delay("FR", HazardType.FIRE)
-    ext_and_transform_data.delay("FF", HazardType.FLASH_FLOOD)
-    ext_and_transform_data.delay("HT", HazardType.HEAT_WAVE)
-    ext_and_transform_data.delay("IN", HazardType.INSECT_INFESTATION)
-    ext_and_transform_data.delay("LS", HazardType.LANDSLIDE)
-    ext_and_transform_data.delay("MS", HazardType.MUD_SLIDE)
-    ext_and_transform_data.delay("ST", HazardType.SEVERE_LOCAL_STROM)
-    ext_and_transform_data.delay("SL", HazardType.SLIDE)
-    ext_and_transform_data.delay("AV", HazardType.SNOW_AVALANCHE)
-    ext_and_transform_data.delay("SS", HazardType.STORM)
-    ext_and_transform_data.delay("AC", HazardType.TECH_DISASTER)
-    ext_and_transform_data.delay("TO", HazardType.TORNADO)
-    ext_and_transform_data.delay("VW", HazardType.VIOLENT_WIND)
-    ext_and_transform_data.delay("WV", HazardType.WAVE_SURGE)
+    for hazard_type, hazard_type_str in GLIDE_HAZARDS:
+        _ext_and_transform_glide_latest_data(hazard_type, hazard_type_str)
+
+
+@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)

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

@@ -1,90 +1,16 @@
-import logging
-from datetime import datetime
-
-import requests
-from celery import shared_task
-from django.conf import settings
-
-from apps.etl.extraction.sources.base.extract import Extraction
-from apps.etl.extraction.sources.base.utils import store_extraction_data
+from apps.etl.extraction.sources.base.handler import BaseExtraction
 from apps.etl.models import ExtractionData
+from main.celery import app
 
-logger = logging.getLogger(__name__)
-
-
-@shared_task
-def extract_glide_latest_data(hazard_type, hazard_type_str):
-    ext_object = (
-        ExtractionData.objects.filter(
-            source=ExtractionData.Source.GLIDE,
-            hazard_type=hazard_type,
-            status=ExtractionData.Status.SUCCESS,
-            resp_data__isnull=False,
-        )
-        .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"https://www.glidenumber.net/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
-    return import_glide_hazard_data(hazard_type, hazard_type_str, url)
-
+HEADERS = {"accept": "application/json"}
 
-@shared_task
-def extract_glide_historical_data(hazard_type, hazard_type_str):
-    to_date = datetime.today().date()
-    url = f"https://www.glidenumber.net/glide/jsonglideset.jsp?toyear={to_date.year}&frommonth={to_date.month}&to_date={to_date.day}&events={hazard_type}"  # noqa: E501
-    return import_glide_hazard_data(hazard_type, hazard_type_str, url)
 
-
-@shared_task(bind=True, max_retries=3, default_retry_delay=60)
-def import_glide_hazard_data(self, hazard_type: str, hazard_type_str: str, url: str, **kwargs):
+class GlideExtraction(BaseExtraction):
     """
-    Import hazard data from glide api
+    Handles data extraction from the GLIDE API.
     """
-    logger.info(f"Importing GDACS - {hazard_type} data")
-
-    # Create a Extraction object in the begining
-    instance_id = kwargs.get("instance_id", None)
-    retry_count = kwargs.get("retry_count", None)
-
-    glide_instance = (
-        ExtractionData.objects.get(id=instance_id)
-        if instance_id
-        else ExtractionData.objects.create(
-            source=ExtractionData.Source.GLIDE,
-            status=ExtractionData.Status.PENDING,
-            source_validation_status=ExtractionData.ValidationStatus.NO_VALIDATION,
-            hazard_type=hazard_type_str,
-            attempt_no=0,
-            resp_code=0,
-        )
-    )
-
-    # Extract the data from api.
-    glide_extraction = Extraction(url=url)
-    response = None
-    try:
-        response = glide_extraction.pull_data(
-            source=ExtractionData.Source.GLIDE,
-            ext_object_id=glide_instance.id,
-            retry_count=retry_count if retry_count else 1,
-        )
-    except requests.exceptions.RequestException as exc:
-        self.retry(exc=exc, kwargs={"instance_id": glide_instance.id, "retry_count": self.request.retries})
-
-    if response:
-        # Save the extracted data into the existing glide object
-        glide_instance = store_extraction_data(
-            response=response,
-            source=ExtractionData.Source.GLIDE,
-            validate_source_func=None,
-            instance_id=glide_instance.id,
-        )
 
-        logger.info(f"{hazard_type} data imported sucessfully")
-        return glide_instance.id
+    @staticmethod
+    @app.task
+    def task(DATA_URL):
+        return GlideExtraction().handle_extraction(DATA_URL, None, HEADERS, ExtractionData.Source.GLIDE)

apps/etl/management/commands/extract_glide_data.py

@@ -1,39 +1,10 @@
-import logging
-
 from django.core.management.base import BaseCommand
 
 from apps.etl.etl_tasks.glide import ext_and_transform_glide_historical_data
-from apps.etl.models import HazardType
-
-logger = logging.getLogger(__name__)
 
 
 class Command(BaseCommand):
     help = "Import data from glide api"
 
     def handle(self, *args, **options):
-        ext_and_transform_glide_historical_data.delay("EQ", HazardType.EARTHQUAKE)
-        ext_and_transform_glide_historical_data.delay("TC", HazardType.CYCLONE)
-        ext_and_transform_glide_historical_data.delay("FL", HazardType.FLOOD)
-        ext_and_transform_glide_historical_data.delay("DR", HazardType.DROUGHT)
-        ext_and_transform_glide_historical_data.delay("WF", HazardType.WILDFIRE)
-        ext_and_transform_glide_historical_data.delay("VO", HazardType.VOLCANO)
-        ext_and_transform_glide_historical_data.delay("TS", HazardType.TSUNAMI)
-        ext_and_transform_glide_historical_data.delay("CW", HazardType.COLDWAVE)
-        ext_and_transform_glide_historical_data.delay("EP", HazardType.EPIDEMIC)
-        ext_and_transform_glide_historical_data.delay("EC", HazardType.EXTRATROPICAL_CYCLONE)
-        ext_and_transform_glide_historical_data.delay("ET", HazardType.EXTREME_TEMPERATURE)
-        ext_and_transform_glide_historical_data.delay("FR", HazardType.FIRE)
-        ext_and_transform_glide_historical_data.delay("FF", HazardType.FLASH_FLOOD)
-        ext_and_transform_glide_historical_data.delay("HT", HazardType.HEAT_WAVE)
-        ext_and_transform_glide_historical_data.delay("IN", HazardType.INSECT_INFESTATION)
-        ext_and_transform_glide_historical_data.delay("LS", HazardType.LANDSLIDE)
-        ext_and_transform_glide_historical_data.delay("MS", HazardType.MUD_SLIDE)
-        ext_and_transform_glide_historical_data.delay("ST", HazardType.SEVERE_LOCAL_STROM)
-        ext_and_transform_glide_historical_data.delay("SL", HazardType.SLIDE)
-        ext_and_transform_glide_historical_data.delay("AV", HazardType.SNOW_AVALANCHE)
-        ext_and_transform_glide_historical_data.delay("SS", HazardType.STORM)
-        ext_and_transform_glide_historical_data.delay("AC", HazardType.TECH_DISASTER)
-        ext_and_transform_glide_historical_data.delay("TO", HazardType.TORNADO)
-        ext_and_transform_glide_historical_data.delay("VW", HazardType.VIOLENT_WIND)
-        ext_and_transform_glide_historical_data.delay("WV", HazardType.WAVE_SURGE)
+        ext_and_transform_glide_historical_data()

apps/etl/tasks.py

@@ -11,7 +11,7 @@
 )
 from apps.etl.etl_tasks.gfd import ext_and_transform_gfd_latest_data  # noqa: F401
 from apps.etl.etl_tasks.glide import (  # noqa: F401
-    ext_and_transform_data,
+    ext_and_transform_glide_historical_data,
     ext_and_transform_glide_latest_data,
 )
 from apps.etl.etl_tasks.idu import (  # noqa: F401
@@ -34,9 +34,7 @@
 )
 from apps.etl.extraction.sources.gfd.extract import GFDExtraction
 from apps.etl.extraction.sources.gidd.extract import GIDDExtraction
-from apps.etl.extraction.sources.glide.extract import (  # noqa: F401
-    import_glide_hazard_data,
-)
+from apps.etl.extraction.sources.glide.extract import GlideExtraction
 from apps.etl.extraction.sources.idu.extract import IDUExtraction
 from apps.etl.extraction.sources.ifrc_event.extract import IFRCEventExtraction
 from apps.etl.models import ExtractionData, HazardType  # noqa: F401
@@ -50,7 +48,7 @@
 )
 from apps.etl.transform.sources.gfd import GFDTransformHandler  # noqa: F401
 from apps.etl.transform.sources.gidd import GIDDTransformHandler  # noqa: F401
-from apps.etl.transform.sources.glide import transform_glide_event_data  # noqa: F401
+from apps.etl.transform.sources.glide import GlideTransformHandler  # noqa: F401
 from apps.etl.transform.sources.handler import BaseTransformerHandler
 from apps.etl.transform.sources.idu import IDUTransformHandler  # noqa: F401
 from apps.etl.transform.sources.ifrc_event import (  # noqa: F401
@@ -62,6 +60,7 @@
 GIDDExtraction.handle_extraction
 GFDExtraction.handle_extraction
 IFRCEventExtraction.handle_extraction
+GlideExtraction.handle_extraction
 
 BaseTransformerHandler.handle_transformation
 

apps/etl/transform/sources/glide.py

@@ -1,68 +1,22 @@
-import logging
-import uuid
-
-from celery import shared_task
 from pystac_monty.sources.glide import GlideDataSource, GlideTransformer
 
-from apps.etl.models import ExtractionData, PyStacLoadData, Transform
-from apps.etl.utils import read_file_data
-from main.managers import BulkCreateManager
-
-logger = logging.getLogger(__name__)
-
-glide_item_type_map = {
-    "glide-events": PyStacLoadData.ItemType.EVENT,
-    "glide-hazards": PyStacLoadData.ItemType.HAZARD,
-}
-
-
-@shared_task
-def transform_glide_event_data(extraction_id):
-    logger.info("Transformation started for glide data")
-    glide_instance = ExtractionData.objects.get(id=extraction_id)
-
-    if not glide_instance.resp_data:
-        logger.info("Transformation ended due to no data")
-        return
-
-    data = read_file_data(glide_instance.resp_data)
-
-    transform_obj = Transform.objects.create(
-        extraction=glide_instance,
-        status=Transform.Status.PENDING,
-    )
-
-    bulk_mgr = BulkCreateManager(chunk_size=1000)
-    try:
-        transformer = GlideTransformer(GlideDataSource(source_url=glide_instance.url, data=data))
-        transformed_event_items = transformer.make_items()
+from apps.etl.models import ExtractionData
+from apps.etl.transform.sources.handler import BaseTransformerHandler
+from main.celery import app
 
-        transform_obj.status = Transform.Status.SUCCESS
-        transform_obj.save(update_fields=["status"])
-    except Exception as e:
-        logger.error("Glide transformation failed", exc_info=True, extra={"extraction_id": glide_instance.id})
-        transform_obj.status = Transform.Status.FAILED
-        transform_obj.save(update_fields=["status"])
-        # FIXME: Check if this creates duplicate entry in Sentry. if yes, remove this.
-        raise e
 
-    for item in transformed_event_items:
-        item_type = glide_item_type_map[item.collection_id]
-        transformed_item_dict = item.to_dict()
-        transformed_item_dict["properties"]["monty:etl_id"] = str(uuid.uuid4())
-        bulk_mgr.add(
-            PyStacLoadData(
-                transform_id=transform_obj,
-                item=transformed_item_dict,
-                collection_id=item.collection_id,
-                item_type=item_type,
-                load_status=PyStacLoadData.LoadStatus.PENDING,
-            )
-        )
+class GlideTransformHandler(BaseTransformerHandler):
+    transformer = GlideTransformer
+    transformer_schema = GlideDataSource
 
-    bulk_mgr.done()
+    @classmethod
+    def get_schema_data(cls, extraction_obj: ExtractionData):
+        with extraction_obj.resp_data.open() as file_data:
+            data = file_data.read()
 
-    transform_obj.is_loaded = True
-    transform_obj.save(update_fields=["is_loaded"])
+        return cls.transformer_schema(source_url=extraction_obj.url, data=data)
 
-    logger.info("Transformation ended for glide data")
+    @staticmethod
+    @app.task
+    def task(extraction_id):
+        return GlideTransformHandler().handle_transformation(extraction_id)

apps/etl/transform/sources/handler.py

@@ -24,6 +24,8 @@
     "pdc-events": PyStacLoadData.ItemType.EVENT,
     "pdc-hazards": PyStacLoadData.ItemType.HAZARD,
     "pdc-impacts": PyStacLoadData.ItemType.IMPACT,
+    "glide-events": PyStacLoadData.ItemType.EVENT,
+    "glide-hazards": PyStacLoadData.ItemType.HAZARD,
 }