fixup! feat(etl): refactor existing extraction into ETL.

Author: sandeshit

alert_system/etl/Gdacs_flood/config.py

@@ -6,4 +6,5 @@ def __init__(self):
         self.impact_endpoint = "/gdacs-impacts/items"
         self.people_exposed_threshold = 5
 
+
 gdacs_flood_config = GdacsFloodConfig()

alert_system/etl/Gdacs_flood/extraction.py

@@ -1,21 +1,23 @@
+import logging
+
 from alert_system.etl.base.extraction import BaseExtractionClass
-from alert_system.etl.base.transform import BaseTransformerClass
 from alert_system.etl.base.loader import BaseLoaderClass
-import logging
-from .transform import GdacsTransformer
-from .loader import GdacsLoader
+from alert_system.etl.base.transform import BaseTransformerClass
+
 from .config import gdacs_flood_config
+from .loader import GdacsLoader
+from .transform import GdacsTransformer
 
 logger = logging.getLogger(__name__)
 
+
 class GdacsFloodExtraction(BaseExtractionClass):
     event_endpoint = gdacs_flood_config.event_endpoint
     hazard_endpoint = getattr(gdacs_flood_config, "hazard_endpoint", None)
     impact_endpoint = getattr(gdacs_flood_config, "impact_endpoint", None)
-    
+
     def get_transformer_class(self) -> type[BaseTransformerClass]:
         return GdacsTransformer
-    
+
     def get_loader_class(self) -> type[BaseLoaderClass]:
         return GdacsLoader
-

alert_system/etl/Gdacs_flood/loader.py

@@ -1,6 +1,8 @@
 from alert_system.etl.base.loader import BaseLoaderClass
+
 from .config import gdacs_flood_config
 
+
 class GdacsLoader(BaseLoaderClass):
     people_exposed_threshold = gdacs_flood_config.people_exposed_threshold
 

alert_system/etl/Gdacs_flood/transform.py

@@ -1,6 +1,7 @@
-from typing import Dict, List, Tuple, Any
-from alert_system.etl.base.transform import BaseTransformerClass
 import logging
+from typing import Dict, Tuple
+
+from alert_system.etl.base.transform import BaseTransformerClass
 
 logger = logging.getLogger(__name__)
 
@@ -21,23 +22,29 @@ class GdacsTransformer(BaseTransformerClass):
         ("people", "highest_risk"): "people.highest_risk",
         ("buildings", "destroyed"): "buildings.destroyed",
     }
-    
+
     # NOTE: This logic might change in future
     def compute_people_exposed(self, impacts: dict) -> int:
-        value = next((impacts.get(key) for key in ["people.affected_total", "people.potentially_affected", "people.affected_direct"] if impacts.get(key)), 0)
+        value = next(
+            (
+                impacts.get(key)
+                for key in ["people.affected_total", "people.potentially_affected", "people.affected_direct"]
+                if impacts.get(key)
+            ),
+            0,
+        )
         if not isinstance(value, int):
             logger.warning(f"people_exposed value is not int: {value}")
             return 0
         return value
 
-    
     # NOTE: This logic might change in future
     def compute_buildings_exposed(self, impacts: dict) -> int:
         """
         Compute the 'buildings_exposed' field.
         """
         return impacts.get("buildings.destroyed") or 0
-    
+
     def process_impact(self, impact_items) -> BaseTransformerClass.ImpactType:
         raw_impacts, metadata = {}, {}
         for item in impact_items:
@@ -56,7 +63,7 @@ def process_impact(self, impact_items) -> BaseTransformerClass.ImpactType:
             "buildings_exposed": self.compute_buildings_exposed(raw_impacts),
             "impact_metadata": metadata,
         }
-    
+
     def process_hazard(self, hazard_item) -> BaseTransformerClass.HazardType:
         if not hazard_item:
             return {
@@ -73,9 +80,9 @@ def process_hazard(self, hazard_item) -> BaseTransformerClass.HazardType:
             "severity_label": detail.get("severity_label", ""),
             "severity_value": detail.get("severity_value", 0),
         }
-    
+
     def process_event(self, event_item) -> BaseTransformerClass.EventType:
-        properties = event_item.resp_data.get("properties",{})
+        properties = event_item.resp_data.get("properties", {})
         return {
             "title": properties.get("title", ""),
             "description": properties.get("description", ""),

alert_system/etl/base/extraction.py

@@ -1,13 +1,17 @@
-import logging
-from datetime import datetime, timedelta, timezone
-from typing import Dict, Generator, Optional, List
 import abc
+import logging
 from abc import ABC
+from datetime import datetime, timedelta, timezone
+from typing import Dict, Generator, List, Optional
+
 import httpx
 from django.db import transaction
+
 from alert_system.models import Connector, ExtractionItem
-from .transform import BaseTransformerClass
+
 from .loader import BaseLoaderClass
+from .transform import BaseTransformerClass
+
 logger = logging.getLogger(__name__)
 
 
@@ -49,7 +53,7 @@ def fetch_stac_data(self, url: str, filters: Optional[Dict] = None) -> Generator
     @abc.abstractmethod
     def get_transformer_class(self) -> type[BaseTransformerClass]:
         raise NotImplementedError()
-    
+
     @abc.abstractmethod
     def get_loader_class(self) -> type[BaseLoaderClass]:
         raise NotImplementedError()
@@ -134,7 +138,6 @@ def _extract_impact_items(self, stac_obj: ExtractionItem) -> List[ExtractionItem
             if impact_object:
                 impact_objects.append(impact_object)
         return impact_objects
-            
 
     def _extract_hazard_items(self, stac_obj: ExtractionItem) -> ExtractionItem | None:
         """Process hazard items related to a STAC event object."""
@@ -196,14 +199,14 @@ def process_event_items(self) -> None:
                     impact_obj = self._extract_impact_items(event_obj)
 
                     transformer = transformer_class(
-                        event_obj = event_obj,
-                        hazard_obj = hazard_obj,
-                        impact_obj = impact_obj,
+                        event_obj=event_obj,
+                        hazard_obj=hazard_obj,
+                        impact_obj=impact_obj,
                     )
                     transformed_data = transformer.transform_stac_item()
-                
+
                 loader.load(transformed_data, self.connector)
-                
+
                 logger.info(f"Successfully processed event {event_id}")
 
             except Exception as e:
@@ -219,4 +222,4 @@ def run(self) -> None:
             logger.info("Connector run completed successfully")
         except Exception as e:
             logger.error(f"Connector run failed: {e}", exc_info=True)
-            raise 
+            raise

alert_system/etl/base/loader.py

@@ -1,8 +1,8 @@
-from django.db import transaction
 import logging
+from abc import ABC, abstractmethod
 from typing import Dict
+
 from alert_system.models import Connector, LoadItems
-from abc import ABC, abstractmethod
 
 logger = logging.getLogger(__name__)
 
@@ -13,40 +13,39 @@ class BaseLoaderClass(ABC):
     @abstractmethod
     def filter_eligible_items(self, load_obj):
         raise NotImplementedError()
-    
+
     def load(self, transformed_data: Dict, connector: Connector) -> LoadItems:
         """
         Save aggregated event.
-        
+
         Args:
             transformed_data: Output from transformer.transform()
             connector: The connector this data came from
-            
+
         Returns:
             Created DisasterEvent object
         """
-        correlation_id = transformed_data['correlation_id']
+        correlation_id = transformed_data["correlation_id"]
         is_item_eligible = self.filter_eligible_items(transformed_data)
-        
+
         load_obj, created = LoadItems.objects.update_or_create(
             correlation_id=correlation_id,
             defaults={
-                'connector': connector,
-                #'event_type': transformed_data.get('event_type'),
-                'event_title': transformed_data.get('title'),
-                'event_description': transformed_data.get('description'),
-                'country': transformed_data.get('country'),
-                'severity_value': transformed_data.get('severity_value'),
-                'severity_label': transformed_data.get('severity_label'),
-                'severity_unit': transformed_data.get('severity_unit'),
-                'total_people_exposed': transformed_data.get('people_exposed'),
-                'total_buildings_exposed': transformed_data.get('buildings_exposed'),
-                'impact_metadata': transformed_data.get('impact_metadata'),
-                'item_eligible': is_item_eligible,
-            }
+                "connector": connector,
+                "event_title": transformed_data.get("title"),
+                "event_description": transformed_data.get("description"),
+                "country": transformed_data.get("country"),
+                "severity_value": transformed_data.get("severity_value"),
+                "severity_label": transformed_data.get("severity_label"),
+                "severity_unit": transformed_data.get("severity_unit"),
+                "total_people_exposed": transformed_data.get("people_exposed"),
+                "total_buildings_exposed": transformed_data.get("buildings_exposed"),
+                "impact_metadata": transformed_data.get("impact_metadata"),
+                "item_eligible": is_item_eligible,
+            },
         )
-        
+
         action = "Created" if created else "Updated"
         logger.info(f"{action} Event for correlation_id={correlation_id}")
-        
+
         return load_obj

alert_system/etl/base/transform.py

@@ -1,10 +1,12 @@
-from abc import abstractmethod, ABC
 import logging
-from typing import TypedDict, Optional, List
+from abc import ABC, abstractmethod
+from typing import List, Optional, TypedDict
+
 from alert_system.models import ExtractionItem
 
 logger = logging.getLogger(__name__)
 
+
 class BaseTransformerClass(ABC):
 
     class ImpactType(TypedDict):
@@ -22,7 +24,9 @@ class EventType(TypedDict):
         description: str
         country: str
 
-    def __init__(self, event_obj: ExtractionItem, hazard_obj: Optional[ExtractionItem] = None, impact_obj: List[ExtractionItem] = []):
+    def __init__(
+        self, event_obj: ExtractionItem, hazard_obj: Optional[ExtractionItem] = None, impact_obj: List[ExtractionItem] = []
+    ):
         self.event_obj = event_obj
         self.hazard_obj = hazard_obj
         self.impact_obj = impact_obj
@@ -44,7 +48,7 @@ def transform_stac_item(self):
         """
         Transform STAC items for a given extraction object.
 
-        Fetches event, hazard and impact items separately, processes them, 
+        Fetches event, hazard and impact items separately, processes them,
         and returns processed data if available.
         """
         logger.info(f"Starting transformer for correlation_id={self.correlation_id}")
@@ -55,12 +59,8 @@ def transform_stac_item(self):
         impact_result = self.process_impact(self.impact_obj)
 
         return {
-            'correlation_id': self.correlation_id,
+            "correlation_id": self.correlation_id,
             **event_result,
             **hazard_result,
             **impact_result,
         }
-
-    
-
-