update admin_notifications setup

Author: codinger41

.circleci/config.yml

@@ -191,7 +191,7 @@ jobs:
             ./cc-test-reporter before-build
             . venv/bin/activate
             coverage combine parallel-coverage/ 
-            coverage xml
+            coverage xml -i
             coverage report
             ./cc-test-reporter format-coverage -o ./.coverage -t coverage.py
             ./cc-test-reporter upload-coverage -i .coverage

.codeclimate.yml

@@ -2,4 +2,7 @@ version: "2"
 exclude_patterns:
   - "helpers/auth/authentication.py"
   - "helpers/calendar/events.py"
-  - "alembic/"
+  - "alembic/"
+  - "admin_notifications/__init__.py"
+  - "admin_notifications/helpers/__init__.py"
+  - "admin_notifications/helpers/notification_templates.py"

admin_notifications/helpers/create_notification.py

@@ -1,24 +1,34 @@
-from flask_socketio import send
 from admin_notifications.models import AdminNotification
-from manage import socketio
-
-def create_notification(title, message, location_id):
-  notification = AdminNotification(
-    title=title,
-    message=message,
-    location_id=location_id,
-    status="unread"
-  )
-  notification.save()
-  
-  @socketio.on('notification')
-  def send_notification():
-    notification = { "title": title, "message": message }
-    return send(notification, broadcast=True)
+from api.location.models import Location
+from datetime import datetime
+import celery
 
 
+def update_notification(notification_id):
+    notification = AdminNotification.query.filter_by(id=notification_id).first()
+    notification.date_received = datetime.now()
+    notification.save()
 
 
-
-def create_notif_lol():
-  return create_notification(title="twre3r", message="rerf3ef3rf3rf", location_id=1)
+@celery.task
+def create_notification(title, message, location_id):
+    """
+    Create notifications in the database and emit them to the client
+    """
+    from manage import socketio
+    location = Location.query.filter_by(id=location_id).first()
+    location_name = location.name
+    notification = AdminNotification(
+        title=title,
+        message=message,
+        location_id=location_id,
+        status="unread"
+    )
+    notification.save()
+    new_notification = {"title": title, "message": message}
+    return socketio.emit(
+            f"notifications-{location_name}",
+            {'notification': new_notification},
+            broadcast=True,
+            callback=update_notification(notification.id)
+        )

admin_notifications/helpers/device_last_seen.py

@@ -0,0 +1,6 @@
+def event_auto_cancelled_notification(event_name, room_name):
+    return {
+        "title": "Event Auto cancelled.",
+        "message": f"An event {event_name} in {room_name} \
+has been auto cancelled."
+    }

admin_notifications/helpers/notification_templates.py

@@ -15,4 +15,5 @@ class AdminNotification(Base, Utility):
     location_id = Column(
         Integer,
         ForeignKey('locations.id', ondelete="CASCADE"),
-        nullable=True)
+        nullable=True
+    )

admin_notifications/helpers/queue_manager.py

@@ -0,0 +1,67 @@
+import graphene
+from graphene_sqlalchemy import SQLAlchemyObjectType
+from graphql import GraphQLError
+from helpers.auth.authentication import Auth
+from datetime import datetime
+from admin_notifications.models import AdminNotification as \
+    AdminNotificationModel
+
+
+class AdminNotifications(SQLAlchemyObjectType):
+    """
+        Returns the admin_notificationspayload
+    """
+    class Meta:
+        model = AdminNotificationModel
+
+
+class NotificationsList(graphene.ObjectType):
+    """
+        Class to return the types for Admin notifications
+        \n- rooms: The rooms data
+    """
+    notifications = graphene.List(AdminNotifications)
+
+
+class Query(graphene.ObjectType):
+    all_unread_notifications = graphene.Field(
+        NotificationsList,
+        description="Returns a list of admin notifications"
+    )
+
+    @Auth.user_roles('Admin')
+    def resolve_all_unread_notifications(self, info):
+        query = AdminNotifications.get_query(info)
+        notifications = query.filter(
+            AdminNotificationModel.status == "unread").all()
+        return NotificationsList(notifications=notifications)
+
+
+class UpdateNotificationStatus(graphene.Mutation):
+    """
+        Class to update the status of a notification
+    """
+
+    class Arguments:
+        notification_id = graphene.Int(required=True)
+    notification = graphene.Field(AdminNotifications)
+
+    @Auth.user_roles('Admin')
+    def mutate(self, info, notification_id):
+        query = AdminNotifications.get_query(info)
+        unread_notifications = query.filter(
+            AdminNotificationModel.status == "unread")
+        notification = unread_notifications.filter(
+            AdminNotificationModel.id == notification_id).first()
+        if not notification:
+            raise GraphQLError("Notification is already read or not found.")
+        notification.status = "read"
+        notification.date_read = datetime.now()
+        notification.save()
+        return UpdateNotificationStatus(notification=notification)
+
+
+class Mutation(graphene.ObjectType):
+    update_notification_status = UpdateNotificationStatus.Field(
+        description="Updates the status of a notification and takes the argument\
+            \n- notification_id: The name of the room[required]")