Source code for hawat.blueprints.detectors

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# -------------------------------------------------------------------------------
# This file is part of Mentat system (https://mentat.cesnet.cz/).
#
# Copyright (C) since 2011 CESNET, z.s.p.o (http://www.ces.net/)
# Use of this source is governed by the MIT license, see LICENSE file.
# -------------------------------------------------------------------------------


"""
This pluggable module provides access to detectors management features. These
features include:

* general detectors listing
* detailed detector record view
* creating new detector records
* updating existing detector records
* deleting existing detector records
"""

__author__ = "Rajmund Hruška <rajmund.hruska@cesnet.cz>"
__credits__ = "Jan Mach <jan.mach@cesnet.cz>, Pavel Kácha <pavel.kacha@cesnet.cz>, Andrea Kropáčová <andrea.kropacova@cesnet.cz>"

import markupsafe
from flask_babel import gettext, lazy_gettext

from sqlalchemy import or_

from mentat.datatype.sqldb import DetectorModel, ItemChangeLogModel

import hawat.acl
import hawat.menu
from hawat.base import HawatBlueprint
from hawat.view import ItemListView, ItemShowView, ItemCreateView, ItemUpdateView, ItemDeleteView
from hawat.view.mixin import HTMLMixin, SQLAlchemyMixin
from hawat.blueprints.detectors.forms import AdminCreateDetectorForm, AdminUpdateDetectorForm, DetectorSearchForm

BLUEPRINT_NAME = 'detectors'
"""Name of the blueprint as module global constant."""


[docs]class ListView(HTMLMixin, SQLAlchemyMixin, ItemListView): """ General detector record listing. """ methods = ['GET'] authentication = True authorization = [hawat.acl.PERMISSION_POWER]
[docs] @classmethod def get_view_title(cls, **kwargs): return lazy_gettext('Detector management')
@property def dbmodel(self): return DetectorModel
[docs] @classmethod def get_action_menu(cls): action_menu = hawat.menu.Menu() action_menu.add_entry( 'endpoint', 'create', endpoint='detectors.create', resptitle=True ) return action_menu
[docs] @classmethod def get_context_action_menu(cls): action_menu = hawat.menu.Menu() action_menu.add_entry( 'endpoint', 'show', endpoint='detectors.show', hidetitle=True ) action_menu.add_entry( 'endpoint', 'update', endpoint='detectors.update', hidetitle=True ) action_menu.add_entry( 'endpoint', 'delete', endpoint='detectors.delete', hidetitle=True ) return action_menu
[docs] @staticmethod def get_search_form(request_args): """ Must return instance of :py:mod:`flask_wtf.FlaskForm` appropriate for searching given type of items. """ return DetectorSearchForm( request_args, meta={'csrf': False} )
[docs] @staticmethod def build_query(query, model, form_args): # Adjust query based on text search string. if 'search' in form_args and form_args['search']: query = query \ .filter( or_( model.name.like('%{}%'.format(form_args['search'])), model.description.like('%{}%'.format(form_args['search'])), ) ) # Adjust query based on lower time boudary selection. if 'dt_from' in form_args and form_args['dt_from']: query = query.filter(model.createtime >= form_args['dt_from']) # Adjust query based on upper time boudary selection. if 'dt_to' in form_args and form_args['dt_to']: query = query.filter(model.createtime <= form_args['dt_to']) # Adjust query based on record source selection. if 'source' in form_args and form_args['source']: query = query \ .filter(model.source == form_args['source']) if 'sortby' in form_args and form_args['sortby']: sortmap = { 'createtime.desc': lambda x, y: x.order_by(y.createtime.desc()), 'createtime.asc': lambda x, y: x.order_by(y.createtime.asc()), 'name.desc': lambda x, y: x.order_by(y.name.desc()), 'name.asc': lambda x, y: x.order_by(y.name.asc()), 'hits.desc': lambda x, y: x.order_by(y.hits.desc()), 'hits.asc': lambda x, y: x.order_by(y.hits.asc()), 'credibility.desc': lambda x, y: x.order_by(y.credibility.desc()), 'credibility.asc': lambda x, y: x.order_by(y.credibility.asc()) } query = sortmap[form_args['sortby']](query, model) return query
[docs]class ShowView(HTMLMixin, SQLAlchemyMixin, ItemShowView): """ Detailed detector record view. """ methods = ['GET'] authentication = True
[docs] @classmethod def get_menu_legend(cls, **kwargs): return lazy_gettext( 'View details of detector record &quot;%(item)s&quot;', item=markupsafe.escape(kwargs['item'].name) )
[docs] @classmethod def get_view_title(cls, **kwargs): return lazy_gettext('Show detector record details')
@property def dbmodel(self): return DetectorModel
[docs] @classmethod def authorize_item_action(cls, **kwargs): return hawat.acl.PERMISSION_POWER.can()
[docs] @classmethod def get_action_menu(cls): action_menu = hawat.menu.Menu() action_menu.add_entry( 'endpoint', 'update', endpoint='detectors.update' ) action_menu.add_entry( 'endpoint', 'delete', endpoint='detectors.delete' ) return action_menu
[docs] def do_before_response(self, **kwargs): item = self.response_context['item'] if self.can_access_endpoint('detectors.update', item=item) and self.has_endpoint('changelogs.search'): self.response_context.update( context_action_menu_changelogs=self.get_endpoint_class( 'changelogs.search' ).get_context_action_menu() ) item_changelog = self.dbsession.query(ItemChangeLogModel). \ filter(ItemChangeLogModel.model == item.__class__.__name__). \ filter(ItemChangeLogModel.model_id == item.id). \ order_by(ItemChangeLogModel.createtime.desc()). \ limit(100). \ all() self.response_context.update(item_changelog=item_changelog)
[docs]class CreateView(HTMLMixin, SQLAlchemyMixin, ItemCreateView): # pylint: disable=locally-disabled,too-many-ancestors """ View for creating new detector records. """ methods = ['GET', 'POST'] authentication = True
[docs] @classmethod def get_menu_title(cls, **kwargs): return lazy_gettext('Create detector record')
[docs] @classmethod def get_view_title(cls, **kwargs): return lazy_gettext('Create new detector record')
@property def dbmodel(self): return DetectorModel @property def dbchlogmodel(self): return ItemChangeLogModel
[docs] @classmethod def authorize_item_action(cls, **kwargs): return hawat.acl.PERMISSION_POWER.can()
[docs] @staticmethod def get_message_success(**kwargs): return gettext( 'Detector record <strong>%(item_id)s</strong> was successfully created.', item_id=markupsafe.escape(str(kwargs['item'])) )
[docs] @staticmethod def get_message_failure(**kwargs): return gettext('Unable to create new detector record.')
[docs] @staticmethod def get_message_cancel(**kwargs): return gettext('Canceled creating new detector record.')
[docs] @staticmethod def get_item_form(item): return AdminCreateDetectorForm()
[docs]class UpdateView(HTMLMixin, SQLAlchemyMixin, ItemUpdateView): # pylint: disable=locally-disabled,too-many-ancestors """ View for updating existing detector records. """ methods = ['GET', 'POST'] authentication = True
[docs] @classmethod def get_menu_legend(cls, **kwargs): return lazy_gettext( 'Update details of detector record &quot;%(item)s&quot;', item=markupsafe.escape(kwargs['item'].name) )
[docs] @classmethod def get_view_title(cls, **kwargs): return lazy_gettext('Update detector record details')
@property def dbmodel(self): return DetectorModel @property def dbchlogmodel(self): return ItemChangeLogModel
[docs] @classmethod def authorize_item_action(cls, **kwargs): return hawat.acl.PERMISSION_POWER.can()
[docs] @staticmethod def get_message_success(**kwargs): return gettext( 'Detector record <strong>%(item_id)s</strong> was successfully updated.', item_id=markupsafe.escape(str(kwargs['item'])) )
[docs] @staticmethod def get_message_failure(**kwargs): return gettext( 'Unable to update detector record <strong>%(item_id)s</strong>.', item_id=markupsafe.escape(str(kwargs['item'])) )
[docs] @staticmethod def get_message_cancel(**kwargs): return gettext( 'Canceled updating detector record <strong>%(item_id)s</strong>.', item_id=markupsafe.escape(str(kwargs['item'])) )
[docs] @staticmethod def get_item_form(item): return AdminUpdateDetectorForm(db_item_id=item.id, obj=item)
[docs]class DeleteView(HTMLMixin, SQLAlchemyMixin, ItemDeleteView): # pylint: disable=locally-disabled,too-many-ancestors """ View for deleting existing detector records. """ methods = ['GET', 'POST'] authentication = True
[docs] @classmethod def get_menu_legend(cls, **kwargs): return lazy_gettext( 'Delete detector record &quot;%(item)s&quot;', item=markupsafe.escape(kwargs['item'].name) )
@property def dbmodel(self): return DetectorModel @property def dbchlogmodel(self): return ItemChangeLogModel
[docs] @classmethod def authorize_item_action(cls, **kwargs): return hawat.acl.PERMISSION_POWER.can()
[docs] @staticmethod def get_message_success(**kwargs): return gettext( 'Detector record <strong>%(item_id)s</strong> was successfully deleted.', item_id=markupsafe.escape(str(kwargs['item'])) )
[docs] @staticmethod def get_message_failure(**kwargs): return gettext( 'Unable to delete detector record <strong>%(item_id)s</strong>.', item_id=markupsafe.escape(str(kwargs['item'])) )
[docs] @staticmethod def get_message_cancel(**kwargs): return gettext( 'Canceled deleting detector record <strong>%(item_id)s</strong>.', item_id=markupsafe.escape(str(kwargs['item'])) )
# -------------------------------------------------------------------------------
[docs]class DetectorsBlueprint(HawatBlueprint): """Pluggable module - detector management (*detectors*)."""
[docs] @classmethod def get_module_title(cls): return lazy_gettext('Detector management')
[docs] def register_app(self, app): app.menu_main.add_entry( 'view', 'admin.{}'.format(BLUEPRINT_NAME), position=71, view=ListView )
# -------------------------------------------------------------------------------
[docs]def get_blueprint(): """ Mandatory interface for :py:mod:`hawat.Hawat` and factory function. This function must return a valid instance of :py:class:`hawat.app.HawatBlueprint` or :py:class:`flask.Blueprint`. """ hbp = DetectorsBlueprint( BLUEPRINT_NAME, __name__, template_folder='templates', url_prefix='/{}'.format(BLUEPRINT_NAME) ) hbp.register_view_class(ListView, '/list') hbp.register_view_class(CreateView, '/create') hbp.register_view_class(ShowView, '/<int:item_id>/show') hbp.register_view_class(UpdateView, '/<int:item_id>/update') hbp.register_view_class(DeleteView, '/<int:item_id>/delete') return hbp