Commit 35e96c89 authored by Martin Matějek's avatar Martin Matějek

WIP; Better notification skeleton handling logic

parent bebb389e
......@@ -41,21 +41,22 @@ def create_argparser():
def print_plugins(plugins):
"""Pretty print plugin list"""
print("Available plugins:")
for k, v in plugins.items():
for p in plugins:
def print_templates(templates):
"""Pretty print templates list"""
print("Available templates:")
for k, v in templates.items():
print("{} - {}".format(k, v))
for t in templates:
def print_notifications(notifications):
"""Pretty print stored notifications"""
print("Stored notifications")
for k, v in notifications.items():
print("{} - {}".format(k, v))
def process_args(parser, args):
......@@ -73,7 +74,7 @@ def process_args(parser, args):
ret = api.get_notifications()
elif == 'plugins':
ret = api.get_plugins()
ret = api.list_plugins()
elif == 'templates':
......@@ -3,12 +3,12 @@ import logging
from .config import config
from .pluginstorage import PluginStorage
from .notificationstorage import NotificationStorage
from .notification import Notification
class Api:
"""Public interface of module"""
def __init__(self, conf=None):
print("Creating instance of API")
if conf: # override default config
......@@ -18,17 +18,15 @@ class Api:
self.notifications = NotificationStorage(
config.get('settings', 'volatile_dir'),
config.get('settings', 'persistent_dir'),
# self.logger.debug("Available notification_types: %s" % self.notifications.notification_types)
def init_logger(self):
"""Init new logger instance"""
self.logger = logging.getLogger("notifylib")
def get_plugins(self):
return self.plugins.get_all()
def list_plugins(self):
"""List of plugin names to client code"""
return self.plugins.get_all().keys()
def get_actions(self, plug_name):
"""Get actions of specified plugin"""
......@@ -39,28 +37,36 @@ class Api:
def get_notification(self, msgid):
"""Show notification of one specific by id"""
if msgid:
return self.notifications.get_notification(msgid)
return self.notifications.get_notification(msgid)
def get_templates(self):
"""Return notification types from plugins"""
return self.notifications.get_notification_types()
return self.plugins.get_notification_types()
# data manipulation
def create(self, **user_opts):
"""Create new notification"""
def store(self, n):
"""Store already created notification"""
def create(self, skel_id, **user_opts):
Create new notification based on selected skeleton
Prefered method for creating notification with minimal knowledge of underlying layers
# get pre-filled skeleton of class Notification
self.logger.debug("Create new notification; user opts entered: %s" % user_opts)
self.logger.debug("Create new notification: chosen skeleton: %s" % skel_id)
self.logger.debug("Create new notification: user opts entered: %s" % user_opts)
notif = self.notifications.get_new_instance(**user_opts)
skel = self.plugin.get_skeleton(skel_id)
notif =, **user_opts)
print("Newly created notification: {}".format(notif))
# print("Newly created notification: {}".format(notif))
self.logger.debug("Stored notifications: %s" % self.notifications.get_all())
# self.logger.debug("Stored notifications: %s" % self.notifications.get_all())
# TODO: refactor
# TODO: rethink/refactor
def call_action(self, mgsid, name, **kwargs):
"""Call action on notification"""
import json
import logging
# TODO: global logging
from datetime import datetime as dt
from .notificationskeleton import NotificationSkeleton
class Notification:
def __init__(self, timestamp, skeleton, persistent=False, **opts):
self.notif_id = timestamp
def __init__(self, notif_id, timestamp, skeleton, persistent=False, **data):
self.notif_id = notif_id
self.timestamp = timestamp
self.skeleton = skeleton
self.opts = opts = data
self.persistent = persistent
# TODO: parse opts into metadata
self.content = opts['message']
self.content =['message']
def from_file(cls, f):
def new(cls, skel, **data):
nid = cls.generate_id()
ts = cls.generate_timestamp()
n = cls(nid, ts, skel, **data)
return n
def from_file(cls, path):
"""Load notification from it's file"""
dict = json.load(f)
return cls(**dict)
except Exception:
# TODO: log failure
with open(path, 'r') as f:
dict = json.load(f)
j_skel = dict['skeleton']
del dict['skeleton']
def valid(self, timestamp):
skel = NotificationSkeleton(j_skel['name'], j_skel['template'], j_skel['actions'])
n =, **dict)
# set attributes of this instance
if 'persistent' in dict:
n.persistent = True
return n
except Exception as e:
# TODO: proper logging
print("Failed to deserialize json file; Error: {}".format(e))
def init_logger(self):
self.logger = logging.getLogger("notifylib")
def valid(self, timestamp=None):
"""If notification is still valid"""
if not timestamp:
t = Notification.generate_timestamp()
# TODO: compare self.timestamp an t
def render(self):
"""Return rendered template as text"""
def serialize_metadata(self):
"""Return serialized data as json"""
return "Content:{}".format(self.content)
def serialize(self):
"""Return serialized data"""
json_data = {
'notif_id': self.notif_id,
'timestamp': self.timestamp,
'persistent': self.persistent,
'message': self.content,
'skeleton': self.skeleton.serialize(),
return json.dumps(json_data)
def generate_id(self):
"""Unique id of message based on timestamp"""
return self.timestamp()
# TODO: append random number for uniqueness
def generate_timestamp(self):
"""Create UTC timestamp"""
return dt.utcnow().timestamp()
def __str__(self):
out = "{\n"
out += "\tbase_type: {}\n".format(self.skeleton)
out += "\tnotif_id: {}\n".format(self.notif_id)
out += "\tskeleton: {}\n".format(self.skeleton)
out += "\ttimestamp: {}\n".format(self.timestamp)
out += "\tpersistent: {}\n".format(self.persistent)
out += "\tmessage: {}\n".format(self.content)
out += "}\n"
return out
import json
class NotificationSkeleton:
def __init__(self, name, template, actions): = name
self.template = template
self.actions = actions
def serialize(self):
json_data = {
'template': self.template,
'actions': self.actions,
return json.dumps(json_data)
def __str__(self):
out = "{\n"
out += "\tname: {}\n".format(
out += "\ttemplate: {}\n".format(self.template)
out += "\tactions: {}\n".format(self.actions)
out += "}\n"
return out
......@@ -7,32 +7,31 @@ from .notification import Notification
class NotificationStorage:
"""In-memory notification storage that serialize and deserialize them"""
def __init__(self, volatile_dir, persistent_dir, notification_types):
print("Constructing new NotifyStorage")
def __init__(self, volatile_dir, persistent_dir):
# print("Constructing new NotifyStorage")
self.storage_dirs = {
'persistent': persistent_dir,
'volatile': volatile_dir,
'fallback': None, # TBD
'fallback': None, # in notification itself
self.notification_types = notification_types # notification data types/templates
self.notifications = []
self.notifications = {}
# self.cached = {}
def init_logger(self):
self.logger = logging.getLogger("notifylib")
def store(self, n):
"""Store in memory and serializate to disk"""
self.notifications[n.notif_id] = n
# n.serialize()
def serialize(self, n):
if n.persistent:
storage_dir = self.storage_dirs['persistent']
......@@ -41,38 +40,36 @@ class NotificationStorage:
# fallback_render_dir = storage_dirs['render_fallback']
# do something to render content
# fallback_content = n.render()
metadata_content = n.serialize_metadata()
fileid = self.generate_id()
# metadata_content = n.serialize_metadata()
content = n.serialize()
# fileid = self.generate_id()
fileid = n.notif_id
# save to disk
regular_file = os.path.join(storage_dir, "{}.json".format(fileid))
# fallback_file = os.path.join(fallback_render_dir, fileid)
# TODO: try/catch
with open(regular_file, 'w') as f:
# with open(fallback_file, 'w') as f:
# f.write(fallback_content)
def load(self, storage_dir):
"""Deserialize from FS"""
self.logger.debug("Deserializing notifications from '%s'" % storage_dir)
for root, dir, files in os.walk(storage_dir):
for f in files:
filepath = os.path.join(storage_dir, f)
self.logger.debug("File %s" % filepath)
# for f in ls storage_dir:
# n = NotificationType.from_file(f)
# if not n.valid():
# delete_from_fs()
# TODO: find better key to identify notification instance in dict
def get_new_instance(self, **opts):
"""Return complete new notification instance based on notification type"""
self.logger.debug("Trying to create from template %s" % opts['template'])
timestamp = dt.utcnow()
return Notification(timestamp, self.notification_types[opts['template']], **opts)
n = Notification.from_file(filepath)
self.notifications[n.notif_id] = n
# TODO: find better key to identify notification instance in dict
# name -> msgid
def get_notification(self, name):
"""Return notification either cached or if missing, cache it and return"""
if not self.cached[name]:
......@@ -84,10 +81,6 @@ class NotificationStorage:
def get_all(self):
return self.notifications
def get_notification_types(self):
"""Return all notification types"""
return self.notification_types
# # TODO: WIP helper fce
# def render_one(self, notif):
# pass
......@@ -96,7 +89,3 @@ class NotificationStorage:
# """Render all notifications"""
# for n in self.notifications:
# self.render_one(n)
def generate_id(self):
"""Unique id of message based on timestamp"""
return dt.utcnow().timestamp()
......@@ -30,7 +30,7 @@ class Plugin:
with open(filename, 'r') as f:
data = yaml.load(f)
# TODO: better filename spliting handling
name = filename.split('.')[0].split('/')[1]
name = filename.split('.')[0].split('/')[-1]
# print("YML data {}".format(data))
......@@ -8,11 +8,13 @@ from .notificationskeleton import NotificationSkeleton
class PluginStorage:
"""Storage for plugins"""
def __init__(self, plugin_dir):
print("Constructing new PluginStorage")
# print("Constructing new PluginStorage")
self.plugin_dir = plugin_dir
self.plugins = {}
self.skeletons = {}
def init_logger(self):
......@@ -38,8 +40,28 @@ class PluginStorage:
"""Return all plugins"""
return self.plugins
def get_skeleton(self, skel_id):
Return notification skeleton based on id
input param in form 'PluginName.skeletonid'
so it need to be parsed to get skel_id
skeleton either exists cached or will be added when needed
# TODO: skeleton storage and lookup
skel_name = skel_id.split('.')[1]
if skel_name not in self.skeletons:
for plug in self.plugins:
notification_types = plug.get_notification_types()
if skel_name in notification_types:
self.skeletons[skel_name] = NotificationSkeleton(**notification_types[skel_name]) # cache it
return self.skeletons[skel_name]
def get_notification_types(self):
ret = {}
ret = []
for name, plugin in self.plugins.items():
self.logger.debug("%s - %s" % (name, plugin))
......@@ -48,8 +70,7 @@ class PluginStorage:
self.logger.debug("Plugin metadata: %s" % args)
for n_name, n_data in args.items():
self.logger.debug("Notif data: %s" % n_data)
ret[n_name] = NotificationSkeleton(**n_data)
type_names = ["{}.{}".format(name, type_name) for type_name in args.keys()]
return ret
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment