Feat/alert system #2599

sandeshit · 2025-11-28T10:58:11Z

Changes

Alert-system app
Polling, ETL

Checklist

Things that should succeed before merging.

Updated/ran unit tests
Updated CHANGELOG.md

Release

If there is a version update, make sure to tag the repository with the latest version.

main/sentry.py

deploy/helm/ifrcgo-helm/values.yaml

alert_system/admin.py

alert_system/models.py

alert_system/management/commands/poll_gdacs_flood.py

- Nested extraction with correlation id as fliter. - Self retry with exponential backoff. - Different models for Hazards and Impacts.

This commit splits the data extraction, transform and load part. The filtration part is also included in the load part.

susilnem

Minor changes...

alert_system/tasks.py

alert_system/etl/base/extraction.py

alert_system/models.py

susilnem

Small changes!

alert_system/models.py

alert_system/tasks.py

alert_system/etl/base/extraction.py

sudip-khanal

Minor changes

alert_system/management/commands/poll_gdacs_cyclone.py

alert_system/management/commands/poll_gdacs_flood.py

alert_system/management/commands/poll_usgs_earthquake.py

susilnem

Do we have any analytics on usage and memory consumption?
Has any profiling been done so far?

susilnem · 2025-12-30T05:07:11Z

alert_system/tasks.py

+        return
+
+    try:
+        eligible_items = LoadItem.objects.filter(


Suggested change

eligible_items = LoadItem.objects.filter(

eligible_items_qs = LoadItem.objects.filter(

susilnem · 2025-12-30T05:08:53Z

alert_system/tasks.py

+        count = eligible_items.count()
+
+        if count == 0:
+            logger.info(f"[Past Events] No eligible items in run {extraction_run_id}")
+            return
+
+        logger.info(f"[Past Events] Processing {count} items from run {extraction_run_id}")
+
+        first_item = eligible_items.first()
+        if not first_item:
+            logger.info("No Connector found for the Event")
+            return


These are almost same checks.

susilnem · 2025-12-30T05:10:42Z

alert_system/tasks.py

+                    processed += 1
+            except Exception as e:
+                failed += 1
+                logger.error(f"[Past Events] Failed for item {load_obj.id} in run {extraction_run_id}: {e}", exc_info=True)


Suggested change

logger.error(f"[Past Events] Failed for item {load_obj.id} in run {extraction_run_id}: {e}", exc_info=True)

logger.warning(f"[Past Events] Failed for item {load_obj.id} in run {extraction_run_id}")

susilnem · 2025-12-30T05:12:54Z

alert_system/tasks.py

+
+        logger.info(f"[Past Events] Completed run {extraction_run_id}: " f"{processed} processed, {failed} failed")
+
+        return {"extraction_run_id": extraction_run_id, "processed": processed, "failed": failed, "total": count}


Add a type dict for returning dict value

susilnem · 2025-12-30T05:16:49Z

alert_system/tasks.py

+        go_event_ids = list(
+            Event.objects.filter(field_reports__num_affected__gte=item.total_people_exposed, dtype=connector.dtype)
+            .values_list("id", flat=True)
+            .distinct()
+        )


thenav56 · 2025-12-30T09:18:35Z

alert_system/tasks.py

+            .distinct()
+        )
+
+        item.related_go_events = go_event_ids


Check if we have instead use M2M

Use bulk insert to add M2M (Use ignore existing)

thenav56 · 2025-12-30T09:27:29Z

alert_system/tasks.py

+        return extraction_run_id
+
+    except Exception as exc:
+        logger.exception(f"[ETL] Connector {connector.id} failed: {exc}")


Suggested change

logger.exception(f"[ETL] Connector {connector.id} failed: {exc}")

logger.warning(f"[ETL] Connector {connector.id} failed", exc_info=True)

thenav56 · 2025-12-30T09:28:25Z

alert_system/tasks.py

+        try:
+            raise self.retry(exc=exc)
+        except MaxRetriesExceededError:
+            logger.error(f"[ETL] Max retries exceeded for connector {connector.id}", exc_info=True)


Suggested change

logger.error(f"[ETL] Max retries exceeded for connector {connector.id}", exc_info=True)

logger.error(f"[ETL] Max retries exceeded for connector {connector.label}", exc_info=True)

thenav56 · 2025-12-30T09:33:25Z

alert_system/models.py

+        help_text=_("Unique identifier for the event item"),
+    )
+
+    resp_data = models.JSONField(


Add a

XXX: This might increase the database size

thenav56 · 2025-12-30T09:37:57Z

alert_system/management/commands/poll_usgs_earthquake.py

+    SOURCE_TYPE = Connector.ConnectorType.USGS_EARTHQUAKE
+
+    def handle(self, *args, **options):
+        if not self.SOURCE_TYPE:


Suggested change

if not self.SOURCE_TYPE:

thenav56 · 2025-12-30T09:40:41Z

alert_system/etl/base/loader.py

+        )
+
+        action = "Created" if created else "Updated"
+        logger.info(f"{action} Event for correlation_id={correlation_id}")


Suggested change

logger.info(f"{action} Event for correlation_id={correlation_id}")

logger.info(f"{action} Event for {correlation_id=}")

thenav56 · 2025-12-30T09:41:06Z

alert_system/etl/Gdacs_flood/extraction.py

alert_system/etl/Gdacs_flood/extraction.py
alert_system/etl/gdacs_flood/extraction.py

thenav56 · 2025-12-30T09:43:01Z

alert_system/etl/base/extraction.py

+            ISO 8601 datetime range string
+        """
+
+        now = datetime.now(timezone.utc)


Suggested change

now = datetime.now(timezone.utc)

from django.utils import timezone

...

now = timezeone.now()

thenav56 · 2025-12-30T09:49:34Z

alert_system/etl/Gdacs_flood/extraction.py

+    event_collection_type = gdacs_flood_config.event_collection_type
+    hazard_collection_type = getattr(gdacs_flood_config, "hazard_collection_type", None)
+    impact_collection_type = getattr(gdacs_flood_config, "impact_collection_type", None)
+    filter_event = getattr(gdacs_flood_config, "filter_event", None)
+    transformer_class = GdacsTransformer
+    loader_class = GdacsLoader


We can replace all this with a single typed dict...

sandeshit mentioned this pull request Nov 28, 2025

Alert-system polling #2584

Closed

2 tasks

sandeshit requested review from susilnem and thenav56 December 2, 2025 04:46

susilnem requested changes Dec 5, 2025

View reviewed changes

sandeshit added 11 commits December 7, 2025 23:29

chore(app): base setup for alert system app

ac93985

feat(polling): extraction logic for polling

7ecacb4

chore(sentry): add cron monitor for tracking

b2f11e8

feat(commands): management commands to run different polling tasks

eb322c5

refactor(extraction): refactor extraction logic for hazards and impacts.

d030147

- Nested extraction with correlation id as fliter. - Self retry with exponential backoff. - Different models for Hazards and Impacts.

feat(extraction): use different classes for different sources

a54e884

feat(filter): add filtration classes and cronjob

a434312

feat(etl): refactor existing extraction into ETL.

72d5c50

This commit splits the data extraction, transform and load part. The filtration part is also included in the load part.

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

2e3b8fa

feat(etl): add past events fetcher

04fcdea

chore(alert-system): clean up naming and cronjobs

9fd5372

sandeshit force-pushed the feat/alert-system branch from 3fa8ab2 to 9fd5372 Compare December 7, 2025 17:49

fixup! chore(alert-system): clean up naming and cronjobs

ef60c54

sandeshit changed the base branch from project/alert-system-workflow to develop December 7, 2025 18:22

sandeshit changed the base branch from develop to project/alert-system-workflow December 7, 2025 18:23

sandeshit marked this pull request as draft December 7, 2025 18:24

sandeshit requested a review from susilnem December 8, 2025 09:09

susilnem requested changes Dec 9, 2025

View reviewed changes

fixup! chore(alert-system): clean up naming and cronjobs

47188d6

susilnem requested changes Dec 11, 2025

View reviewed changes

sudip-khanal requested changes Dec 11, 2025

View reviewed changes

alert_system/management/commands/poll_gdacs_cyclone.py Outdated Show resolved Hide resolved

alert_system/management/commands/poll_gdacs_flood.py Outdated Show resolved Hide resolved

alert_system/management/commands/poll_usgs_earthquake.py Outdated Show resolved Hide resolved

fixup! chore(alert-system): clean up naming and cronjobs

e3d98f9

sandeshit requested a review from susilnem December 12, 2025 09:02

sandeshit added 3 commits December 16, 2025 15:37

fixup! chore(alert-system): clean up naming and cronjobs

9abc84a

fixup! chore(alert-system): clean up naming and cronjobs

1b1a3eb

feat(etl): add past events from GO event table

b161943

sandeshit added 3 commits December 19, 2025 10:54

feat(etl): add usgs source

2b38434

fixup! feat(etl): add usgs source

0ba1868

chore(etl): separate past event extraction into different class

3c874a1

susilnem requested changes Dec 30, 2025

View reviewed changes

thenav56 requested changes Dec 30, 2025

View reviewed changes

	eligible_items = LoadItem.objects.filter(
	eligible_items_qs = LoadItem.objects.filter(

	logger.error(f"[Past Events] Failed for item {load_obj.id} in run {extraction_run_id}: {e}", exc_info=True)
	logger.warning(f"[Past Events] Failed for item {load_obj.id} in run {extraction_run_id}")


		logger.info(f"[Past Events] Completed run {extraction_run_id}: " f"{processed} processed, {failed} failed")

		return {"extraction_run_id": extraction_run_id, "processed": processed, "failed": failed, "total": count}

	logger.exception(f"[ETL] Connector {connector.id} failed: {exc}")
	logger.warning(f"[ETL] Connector {connector.id} failed", exc_info=True)

	logger.error(f"[ETL] Max retries exceeded for connector {connector.id}", exc_info=True)
	logger.error(f"[ETL] Max retries exceeded for connector {connector.label}", exc_info=True)

	logger.info(f"{action} Event for correlation_id={correlation_id}")
	logger.info(f"{action} Event for {correlation_id=}")

-        now = datetime.now(timezone.utc)
+from django.utils import timezone
+...
+        now = timezeone.now()

Feat/alert system #2599

Are you sure you want to change the base?

Feat/alert system #2599

Uh oh!

Conversation

sandeshit commented Nov 28, 2025

Changes

Checklist

Release

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

susilnem left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

susilnem left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sudip-khanal left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

susilnem left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects