Merge pull request #847 from DalgoT4D/notification_dev

[442] - Scripts for creating and deleting notification

ddpui/celeryworkers/tasks.py

@@ -13,7 +13,7 @@
 from ddpui.models.userpreferences import UserPreferences
 from ddpui.utils.discord import send_discord_notification
 from ddpui.utils.sendgrid import send_email_notification, send_schema_changes_email
-from ddpui.utils.timezone import UTC, as_utc
+from ddpui.utils import timezone
 from ddpui.utils.custom_logger import CustomLogger
 from ddpui.core.orgtaskfunctions import get_edr_send_report_task
 from ddpui.models.org import (
@@ -61,6 +61,7 @@
 from ddpui.core import llm_service
 
 logger = CustomLogger("ddpui")
+UTC = timezone.UTC
 
 
 @app.task(bind=True)
@@ -883,7 +884,7 @@ def schedule_notification_task(self, notification_id, recipient_id):
     recipient = OrgUser.objects.get(user_id=recipient_id)
     user_preference, created = UserPreferences.objects.get_or_create(orguser=recipient)
 
-    notification.sent_time = as_utc(datetime.utcnow())
+    notification.sent_time = timezone.as_utc(datetime.now())
     notification.save()
 
     if user_preference.enable_email_notifications:

ddpui/management/commands/create_notification.py

@@ -0,0 +1,103 @@
+import sys
+from django.core.management.base import BaseCommand
+from django.utils.dateparse import parse_datetime
+from ddpui.core import notifications_service
+from ddpui.schemas.notifications_api_schemas import (
+    CreateNotificationPayloadSchema,
+    SentToEnum,
+)
+
+
+class Command(BaseCommand):
+    """Creates a notification"""
+
+    help = "Send notification to a user or a group of users"
+
+    def add_arguments(self, parser):
+        """adds command line arguments"""
+        parser.add_argument("author", type=str, help="Author of the notification")
+        parser.add_argument("message", type=str, help="Message of the notification")
+        parser.add_argument(
+            "sent_to",
+            type=str,
+            choices=[sent_to.value for sent_to in SentToEnum],
+            help="Target audience of the notification (e.g., 'all_users', 'single_user')",
+        )
+        parser.add_argument(
+            "--user_email",
+            type=str,
+            help="Email address of the single user if sent_to is 'single_user'",
+        )
+        parser.add_argument(
+            "--org_slug", type=str, help="Org slug if sent_to is 'all_org_users'"
+        )
+        parser.add_argument(
+            "--manager_or_above",
+            action="store_true",
+            help="Only applicable if sent_to is 'all_org_users'; flag to include managers or above",
+        )
+        parser.add_argument(
+            "--urgent",
+            action="store_true",
+            help="Whether the notification is urgent or not",
+        )
+        parser.add_argument(
+            "--scheduled_time",
+            type=str,
+            help="The scheduled time of the notification in ISO 8601 format (e.g., 2024-08-30T19:51:51Z)",
+        )
+
+    def handle(self, *args, **options):
+        # Parse scheduled_time
+        scheduled_time_str = options.get("scheduled_time")
+        scheduled_time = None
+        if scheduled_time_str:
+            try:
+                scheduled_time = parse_datetime(scheduled_time_str)
+                if not scheduled_time:
+                    raise ValueError("Invalid date format.")
+            except ValueError as e:
+                print(f"Error parsing scheduled_time: {e}")
+                sys.exit(1)
+
+        # Prepare payload
+        payload = CreateNotificationPayloadSchema(
+            author=options["author"],
+            message=options["message"],
+            sent_to=SentToEnum(options["sent_to"]),
+            urgent=options.get("urgent", False),
+            scheduled_time=scheduled_time,
+            user_email=options.get("user_email"),
+            org_slug=options.get("org_slug"),
+            manager_or_above=options.get("manager_or_above", False),
+        )
+
+        # Get recipients based on sent_to field
+        error, recipients = notifications_service.get_recipients(
+            payload.sent_to,
+            payload.org_slug,
+            payload.user_email,
+            payload.manager_or_above,
+        )
+
+        if error:
+            print(f"Error in getting recipients: {error}")
+            sys.exit(1)
+
+        # Create notification data
+        notification_data = {
+            "author": payload.author,
+            "message": payload.message,
+            "urgent": payload.urgent,
+            "scheduled_time": payload.scheduled_time,
+            "recipients": recipients,
+        }
+
+        # Call the create notification service
+        error, result = notifications_service.create_notification(notification_data)
+
+        if error:
+            print(f"Error in creating notification: {error}")
+            sys.exit(1)
+
+        print(f"Notification created successfully: {result}")

ddpui/management/commands/delete_notification.py

@@ -0,0 +1,31 @@
+import sys
+from django.core.management.base import BaseCommand
+from ddpui.core import notifications_service
+
+
+class Command(BaseCommand):
+    """Deletes a notification by its ID"""
+
+    help = "Deletes a notification by its ID"
+
+    def add_arguments(self, parser):
+        """Adds command line arguments"""
+        parser.add_argument(
+            "notification_id", type=int, help="ID of the notification to delete"
+        )
+
+    def handle(self, *args, **options):
+        notification_id = options["notification_id"]
+
+        # Call the notification service to delete the notification
+        error, result = notifications_service.delete_scheduled_notification(
+            notification_id
+        )
+
+        if error is not None:
+            self.stderr.write(f"Error: {error}")
+            sys.exit(1)
+
+        self.stdout.write(
+            f"Notification with ID {notification_id} deleted successfully."
+        )