Project/dashboard backend

.dockerignore

@@ -130,4 +130,6 @@ dmypy.json
 
 # editors
 .idea/
+
+# FIXME: This is temporary
 assets/

.github/workflows/ci.yml

@@ -23,7 +23,7 @@ env:
   COMPOSE_FILE: docker-compose.yml:gh-docker-compose.yml
   DJANGO_SECRET_KEY: "ci-test-insecure-django-secret-key"
   # ETL app configs
-  GEOCODER_URL: https://geocoder.dummy.com
+  GEOCODER_URL: http://geocoding-service:8001 
 
 jobs:
   pre_commit_checks:
@@ -56,7 +56,7 @@ jobs:
     steps:
       - uses: actions/checkout@main
         with:
-          submodules: true
+          submodules: recursive
 
       - name: Login to GitHub Container Registry
         uses: docker/login-action@v3
@@ -118,7 +118,10 @@ jobs:
         env:
           DOCKER_IMAGE_BACKEND: ${{ steps.prep.outputs.tagged_image }}
         run: |
-          touch .env && docker compose run --rm web bash -c 'wait-for-it db:5432 && ./manage.py check'
+          touch .env
+          docker compose run --rm web bash -c 'wait-for-it db:5432 && ./manage.py check'
+          # Another wait for geocoding-service (Move this to wait-for-it using custom python script)
+          sleep 60
 
       - name: 🕮 Validate if there are no pending django migrations.
         env:
@@ -129,10 +132,16 @@ jobs:
             exit 1;
           }
 
-      # - name: 🤞 Run Test 🧪 & Publish coverage to code climate
-      #   env:
-      #     DOCKER_IMAGE_BACKEND: ${{ steps.prep.outputs.tagged_image }}
-      #   run: docker compose run --rm web /code/scripts/run_tests.sh
+      - name: 🤞 Run Test 🧪 & Publish coverage to code climate
+        env:
+          DOCKER_IMAGE_BACKEND: ${{ steps.prep.outputs.tagged_image }}
+        run: |
+          docker compose up -d web geocoding-service 
+          # Wait for 60 seconds to ensure services are initialized
+          echo "Waiting for 30 seconds for services to initialize..."
+          sleep 30
+          # Run the tests
+          docker compose exec web pytest --cov-report term --cov=apps apps/etl/tests/sources/*.py
 
       - name: 🐳 Docker push
         if: ${{ inputs.push_docker_image }}

.gitignore

@@ -135,3 +135,6 @@ dmypy.json
 assets/
 media/
 *.gpkg
+
+apps/etl/dataset
+.DS_Store

Dockerfile

@@ -22,6 +22,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
     apt-get update -y \
     && apt-get install -y --no-install-recommends \
         # Build required packages
+        build-essential libgdal-dev \
         gcc libc-dev gdal-bin libproj-dev \
         # Helper packages
         procps \
@@ -30,6 +31,7 @@ RUN --mount=type=cache,target=/root/.cache/uv \
     && uv sync --frozen --no-install-project --all-groups \
     # Clean-up
     && apt-get remove -y gcc libc-dev libproj-dev \
+     build-essential libgdal-dev \
     && apt-get autoremove -y \
     && rm -rf /var/lib/apt/lists/*
 

apps/etl/dataloaders.py

@@ -0,0 +1,43 @@
+import typing
+
+from asgiref.sync import sync_to_async
+from django.db import models
+from django.utils.functional import cached_property
+from strawberry.dataloader import DataLoader
+
+from apps.etl.models import ExtractionData, PyStacLoadData
+
+DjangoModel = typing.TypeVar("DjangoModel", bound=models.Model)
+
+
+def load_model_objects(
+    Model: typing.Type[DjangoModel],
+    keys: list[int],
+) -> list[DjangoModel]:
+    qs = Model.objects.filter(id__in=keys)
+    _map = {obj.pk: obj for obj in qs}
+    return [_map[key] for key in keys]
+
+
+if typing.TYPE_CHECKING:
+    from apps.etl.types import ExtractionDataType, PyStacLoadDataType
+
+
+def load_extraction(keys: list[int]) -> list["ExtractionDataType"]:
+    return load_model_objects(ExtractionData, keys)  # type: ignore[reportReturnType]
+
+
+def load_pystac(keys: list[int]) -> list["PyStacLoadDataType"]:
+    return load_model_objects(PyStacLoadData, keys)  # type: ignore[reportReturnType]
+
+
+class ExtractionDataLoader:
+    @cached_property
+    def load_data(self):
+        return DataLoader(load_fn=sync_to_async(load_extraction))
+
+
+class PystacDataLoader:
+    @cached_property
+    def load_data(self):
+        return DataLoader(load_fn=sync_to_async(load_pystac))

apps/etl/enums.py

@@ -0,0 +1,24 @@
+import strawberry
+
+from apps.etl.models import ExtractionData, PyStacLoadData, Status
+from utils.strawberry.enums import get_enum_name_from_django_field
+
+DataStatusTypeEnum = strawberry.enum(Status, name="DataStatusTypeEnum")
+ExtractionValidationTypeEnum = strawberry.enum(
+    ExtractionData.ValidationStatus, name="ExtractionDataValidationStatusTypeEnum"
+)
+SourceTypeEnum = strawberry.enum(ExtractionData.Source, name="SourceTypeEnum")
+PyStacLoadDataStatusEnum = strawberry.enum(PyStacLoadData.Status, name="PyStacLoadDataStatusEnum")
+PyStacLoadDataItemTypeEnum = strawberry.enum(PyStacLoadData.ItemType, name="PyStacLoadDataItemTypeEnum")
+
+
+enum_map = {
+    get_enum_name_from_django_field(field): enum
+    for field, enum in (
+        (ExtractionData.source, SourceTypeEnum),
+        (ExtractionData.status, DataStatusTypeEnum),
+        (ExtractionData.source_validation_status, ExtractionValidationTypeEnum),
+        (PyStacLoadData.status, PyStacLoadDataStatusEnum),
+        (PyStacLoadData.item_type, PyStacLoadDataItemTypeEnum),
+    )
+}

apps/etl/etl_tasks/gidd.py

@@ -1,7 +1,6 @@
-from celery import chain, shared_task
+from celery import shared_task
 
 from apps.etl.extraction.sources.gidd.extract import GIDDExtraction, GIDDExtractionMetadata, GIDDExtractionMetadataType
-from apps.etl.transform.sources.gidd import GIDDTransformHandler
 from main.configs import etl_config
 
 
@@ -12,7 +11,4 @@ def ext_and_transform_gidd_latest_data():
     extraction_obj = GIDDExtraction.init_extraction(
         metadata=GIDDExtractionMetadata(url=url, type=GIDDExtractionMetadataType.QUERY), add_to_queue=False
     )
-    chain(
-        GIDDExtraction.task.s(extraction_obj.id),
-        GIDDTransformHandler.task.s(),
-    ).apply_async()
+    GIDDExtraction.task.delay(extraction_obj.id)

apps/etl/etl_tasks/glide.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timedelta
 
-from celery import chain, shared_task
+from celery import shared_task
 
 from apps.etl.extraction.sources.glide.extract import (
     GlideExtraction,
@@ -9,7 +9,6 @@
     GlideExtractionParamsMetadata,
 )
 from apps.etl.models import ExtractionData, HazardType
-from apps.etl.transform.sources.glide import GlideTransformHandler
 from main.configs import etl_config
 
 GLIDE_HAZARDS = [
@@ -66,11 +65,8 @@ def _ext_and_transform_glide_historical_data(hazard_type: HazardType):
             ),
             add_to_queue=False,
         )
-        chain(
-            GlideExtraction.task.s(extraction_object.id),
-            GlideTransformHandler.task.s(),
-        ).apply_async()
 
+        GlideExtraction.task.delay(extraction_object.id)
         start_date = end_date + timedelta(days=1)
 
 

apps/etl/etl_tasks/idu.py

@@ -1,9 +1,8 @@
 import logging
 
-from celery import chain, shared_task
+from celery import shared_task
 
 from apps.etl.extraction.sources.idu.extract import IDUExtraction, IDUExtractionMetadata, IDUExtractionMetadataType
-from apps.etl.transform.sources.idu import IDUTransformHandler
 from main.configs import etl_config
 
 logger = logging.getLogger(__name__)
@@ -16,7 +15,7 @@ def ext_and_transform_idu_historical_data():
     extraction_obj = IDUExtraction.init_extraction(
         metadata=IDUExtractionMetadata(url=url, type=IDUExtractionMetadataType.QUERY), add_to_queue=False
     )
-    chain(IDUExtraction.task.s(extraction_obj.id), IDUTransformHandler.task.s()).apply_async()
+    IDUExtraction.task.delay(extraction_obj.id)
 
 
 @shared_task
@@ -27,4 +26,5 @@ def ext_and_transform_idu_latest_data():
     extraction_obj = IDUExtraction.init_extraction(
         metadata=IDUExtractionMetadata(url=url, type=IDUExtractionMetadataType.QUERY), add_to_queue=False
     )
-    chain(IDUExtraction.task.s(extraction_obj.id), IDUTransformHandler.task.s()).apply_async()
+
+    IDUExtraction.task.delay(extraction_obj.id)

apps/etl/etl_tasks/pdc.py

@@ -16,7 +16,7 @@
 
 
 @shared_task
-def extract_and_transform_pdc_data():
+def extract_and_transform_pdc_latest_data():
     data_url = f"{etl_config.PDC_SENTRY_BASE_URL}/hp_srv/services/hazards/t/json/search_hazard"
     pdc_latest_extraction = ExtractionData.objects.filter(
         source=ExtractionData.Source.PDC,
@@ -28,21 +28,26 @@ def extract_and_transform_pdc_data():
     else:
         start_date = datetime.strptime(str(etl_config.PDC_START_DATE), "%Y-%m-%d")
 
-    data = PdcHazardInputMetadata(
-        pagination=Pagination(page=1, pagesize=100),
-        restrictions=[
-            [
-                Restriction(searchType="GREATER_THAN", createDate=str(int(start_date.timestamp() * 1000))),  # type: ignore
-            ]
-        ],
-    )
-    PDCExtractionV2.init_extraction(
-        metadata=PDCExtractionMetadata(
-            hazard=data,
-            url=data_url,
-            type=PDCExtractionMetaDataType.HAZARD,
-        ),
-    )
+    end_date = datetime.now()
+
+    for event in HAZARD_TYPE_MAP.keys():
+        data = PdcHazardInputMetadata(
+            pagination=Pagination(page=1, pagesize=100),
+            restrictions=[
+                [
+                    Restriction(searchType="GREATER_THAN", createDate=str(int(start_date.timestamp() * 1000))),  # type: ignore
+                    Restriction(searchType="EQUALS", typeId=event),  # type: ignore
+                    Restriction(searchType="LESS_THAN", createDate=str(int(end_date.timestamp() * 1000))),  # type: ignore
+                ]
+            ],
+        )
+        PDCExtractionV2.init_extraction(
+            metadata=PDCExtractionMetadata(
+                hazard=data,
+                url=data_url,
+                type=PDCExtractionMetaDataType.HAZARD,
+            ),
+        )
 
 
 def extract_and_transform_historical_pdc_data():

apps/etl/etl_tasks/usgs.py

@@ -1,5 +1,5 @@
 import logging
-from datetime import date, timedelta
+from datetime import date, datetime, timedelta
 
 from celery import shared_task
 
@@ -30,7 +30,13 @@ def ext_and_transform_usgs_latest_data():
     else:
         start_date = etl_config.USGS_START_DATE
 
-    url = f"{etl_config.USGS_DATA_URL}/fdsnws/event/1/query?format=geojson&starttime={start_date.strftime('%Y-%m-%d')}"
+    end_date = datetime.now().date()
+
+    url = (
+        f"{etl_config.USGS_DATA_URL}/fdsnws/event/1/query?format=geojson"
+        f"&starttime={start_date.strftime('%Y-%m-%d')}"
+        f"&endtime={end_date.strftime('%Y-%m-%d')}"
+    )
 
     USGSExtraction.init_extraction(
         metadata=USGSExtractionMetadata(