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

Author: sandeshit

alert_system/management/commands/poll_base.py

@@ -0,0 +1,29 @@
+import logging
+
+from django.core.management.base import BaseCommand
+
+from alert_system.models import Connector
+from alert_system.tasks import process_connector_task
+
+logger = logging.getLogger(__name__)
+
+
+class BasePollingCommand(BaseCommand):
+    help = "Command to extract data from eoapi"
+
+    SOURCE_TYPE = None
+
+    def handle(self, *args, **options):
+        if not self.SOURCE_TYPE:
+            raise ValueError("SOURCE_TYPE must be defined in subclass.")
+        self.stdout.write("Starting extraction task...")
+        connector = Connector.objects.filter(type=self.SOURCE_TYPE).first()
+        if not connector:
+            logger.warning("No connectors found.")
+            return
+
+        process_connector_task.delay(connector.id)
+
+        logger.info("Connector task dispatched.")
+
+        self.stdout.write("Extraction task finished.")

alert_system/management/commands/poll_gdacs_cy.py

@@ -0,0 +1,13 @@
+from poll_base import BasePollingCommand
+from sentry_sdk.crons import monitor
+
+from main.sentry import SentryMonitor
+
+
+class Command(BasePollingCommand):
+    help = "Poll data for gdacs cyclone"
+    SOURCE_TYPE = "GDACS_CYCLONE"
+
+    @monitor(monitor_slug=SentryMonitor.POLL_GDACS_CY)
+    def handle(self, *args, **options):
+        super().handle(*args, **options)

alert_system/management/commands/poll_gdacs_fl.py

@@ -0,0 +1,14 @@
+from sentry_sdk.crons import monitor
+
+from main.sentry import SentryMonitor
+
+from .poll_base import BasePollingCommand
+
+
+class Command(BasePollingCommand):
+    help = "Poll data for gdacs flood"
+    SOURCE_TYPE = "GDACS_FLOOD"
+
+    @monitor(monitor_slug=SentryMonitor.POLL_GDACS_FL)
+    def handle(self, *args, **options):
+        super().handle(*args, **options)

alert_system/management/commands/poll_usgs_eq.py

@@ -0,0 +1,14 @@
+from sentry_sdk.crons import monitor
+
+from main.sentry import SentryMonitor
+
+from .poll_base import BasePollingCommand
+
+
+class Command(BasePollingCommand):
+    help = "Poll data for usgs eartquake"
+    SOURCE_TYPE = "USGS_EARTHQUAKE"
+
+    @monitor(monitor_slug=SentryMonitor.POLL_USGS_EQ)
+    def handle(self, *args, **options):
+        super().handle(*args, **options)