added notification storage and endpoints to get all or delete one
This commit is contained in:
@@ -6,13 +6,24 @@ logging.basicConfig(
|
||||
|
||||
from flask import Flask, render_template, jsonify, send_from_directory
|
||||
from base.modulemanager import ModuleManager
|
||||
import base.notificationmanager
|
||||
import base.notificationmanager as ntfy
|
||||
|
||||
app = Flask(__name__, static_folder=os.path.join(".", "base", "static"))
|
||||
|
||||
mod_manager = ModuleManager()
|
||||
mod_manager.load(app, "modules")
|
||||
|
||||
@app.route("/notifications", methods=["GET"])
|
||||
def api_notifications():
|
||||
return jsonify(ntfy.NOTIFICATION_MANAGER.serialized())
|
||||
|
||||
@app.route("/notifications/delete/<nid>")
|
||||
def api_delete_notification(nid):
|
||||
is_deleted = ntfy.NOTIFICATION_MANAGER.delete(nid)
|
||||
if is_deleted:
|
||||
return jsonify({}), 200
|
||||
return jsonify({}), 400
|
||||
|
||||
@app.route("/modules")
|
||||
def api_modules():
|
||||
return jsonify(mod_manager.get_mod_list())
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import logging
|
||||
import logging, uuid
|
||||
from datetime import datetime
|
||||
from enum import StrEnum, auto
|
||||
from typing import Dict
|
||||
@@ -11,16 +11,19 @@ class Notification:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
module: str = "defaultmod",
|
||||
status: Status = Status.INFO,
|
||||
msg: str = "Default Notification",
|
||||
):
|
||||
self.status = status
|
||||
self.msg = msg
|
||||
self.module = module
|
||||
self.time = datetime.now()
|
||||
|
||||
def as_dict(self):
|
||||
return {
|
||||
"status": str(self.status),
|
||||
"module": self.module,
|
||||
"msg": self.msg,
|
||||
"time": self.time.isoformat()
|
||||
}
|
||||
@@ -29,11 +32,20 @@ class NotificationManager:
|
||||
def __init__(self):
|
||||
self.notifications = {}
|
||||
|
||||
def notify(self, module: str, notification: Notification):
|
||||
if module not in self.notifications:
|
||||
self.notifications[module] = []
|
||||
self.notifications[module].append(notification)
|
||||
# test notifications
|
||||
self.notify(notification=Notification("socialwatch", Notification.Status.INFO, "some info from sw"))
|
||||
self.notify(notification=Notification("testmodule",Notification.Status.ERROR, "some error from tm"))
|
||||
|
||||
def notify(self, notification: Notification):
|
||||
self.notifications[str(uuid.uuid4())] = notification
|
||||
|
||||
def delete(self, nid):
|
||||
is_removed: bool = self.notifications.pop(nid, None) is not None
|
||||
if is_removed: logging.getLogger(__name__).log(logging.INFO, f"deleted {nid}")
|
||||
return is_removed
|
||||
|
||||
def serialized(self):
|
||||
return {str(k): self.notifications[k].as_dict() for k in self.notifications.keys()}
|
||||
|
||||
print(f"IM NOTIFYING: {notification.status} {notification.msg} BY {module}")
|
||||
|
||||
NOTIFICATION_MANAGER = NotificationManager()
|
||||
|
||||
Reference in New Issue
Block a user