#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#-------------------------------------------------------------------------------
# Use of this source is governed by the MIT license, see LICENSE file.
#-------------------------------------------------------------------------------
"""
This module contains base classes for all *hawat* application views. They are all
based on :py:class:`flask.views.View`.
"""
import re
import sys
import datetime
import pytz
import traceback
import sqlalchemy
import flask
import flask.app
import flask.views
import flask_login
import flask_principal
import flask_mail
from flask_babel import gettext, force_locale
import hawat.const
import hawat.menu
import hawat.db
import hawat.errors
from hawat.view.mixin import HawatUtils
from hawat.forms import ItemActionConfirmForm
[docs]class DecoratedView:
"""
Wrapper class for classical decorated view functions.
"""
def __init__(self, view_function):
self.view_function = view_function
[docs] def get_view_name(self):
"""Simple adapter method to enable support of classical decorated views."""
return self.view_function.__name__
[docs] def get_view_endpoint(self):
"""Simple adapter method to enable support of classical decorated views."""
return self.get_view_name()
[docs] def get_view_icon(self):
"""Simple adapter method to enable support of classical decorated views."""
return 'view-{}'.format(self.get_view_name())
[docs]class BaseView(flask.views.View):
"""
Base class for all custom hawat application views.
"""
module_ref = None
"""
Weak reference to parent module/blueprint of this view.
"""
module_name = None
"""
Name of the parent module/blueprint. Will be set up during the process
of registering the view into the blueprint in :py:func:`hawat.app.hawatBlueprint.register_view_class`.
"""
authentication = False
"""
Similar to the ``decorators`` mechanism in Flask pluggable views, you may use
this class variable to specify, that the view is protected by authentication.
During the process of registering the view into the blueprint in
:py:func:`hawat.app.hawatBlueprint.register_view_class` the view will be
automatically decorated with :py:func:`flask_login.login_required` decorator.
The advantage of using this in favor of ``decorators`` is that the application
menu can automatically hide/show items inaccessible to current user.
This is a scalar variable that must contain boolean ``True`` or ``False``.
"""
authorization = ()
"""
Similar to the ``decorators`` mechanism in Flask pluggable views, you may use
this class variable to specify, that the view is protected by authorization.
During the process of registering the view into the blueprint in
:py:func:`hawat.app.hawatBlueprint.register_view_class` the view will be
automatically decorated with given authorization decorators.
The advantage of using this in favor of ``decorators`` is that the application
menu can automatically hide/show items inaccessible to current user.
This is a list variable that must contain list of desired decorators.
"""
url_params_unsupported = ()
"""
List of URL parameters, that are not supported by this view and should be removed
before generating the URL.
"""
[docs] @classmethod
def get_view_name(cls):
"""
Return unique name for the view. Name must be unique in the namespace of
parent blueprint/module and should contain only characters ``[a-z0-9]``.
It will be used for generating endpoint name for the view.
*This method does not have any default implementation and must be overridden
by a subclass.*
:return: Name for the view.
:rtype: str
"""
raise NotImplementedError()
[docs] @classmethod
def get_view_endpoint_name(cls):
"""
Return unique name for the view endpoint. Name must be unique in the namespace of
parent blueprint/module and should contain only characters ``[a-z0-9]``.
It will be used for generating endpoint name for the view.
*This method does not have any default implementation and must be overridden
by a subclass.*
:return: Name for the view.
:rtype: str
"""
return cls.get_view_name()
[docs] @classmethod
def get_view_endpoint(cls):
"""
Return name of the routing endpoint for the view within the whole application.
Default implementation generates the endpoint name by concatenating the
module name and view name.
:return: Routing endpoint for the view within the whole application.
:rtype: str
"""
return '{}.{}'.format(
cls.module_name,
cls.get_view_endpoint_name()
)
[docs] @classmethod
def get_view_url(cls, **kwargs):
"""
Return view URL.
:param dict kwargs: Optional parameters.
:return: URL for the view.
:rtype: str
"""
# Filter out unsupported URL parameters.
params = dict(filter(lambda x: x[0] not in cls.url_params_unsupported, kwargs.items()))
return flask.url_for(
cls.get_view_endpoint(),
**params
)
[docs] @classmethod
def get_view_icon(cls):
"""
Return menu entry icon name for the view. Given name will be used as index
to built-in icon registry.
Default implementation generates the icon name by concatenating the prefix
``module-`` with module name.
:return: View icon.
:rtype: str
"""
return 'module-{}'.format(cls.module_name.replace('_', '-'))
[docs] @classmethod
def get_view_title(cls, **kwargs):
"""
Return title for the view, that will be displayed in the ``title`` tag of
HTML ``head`` element and also as the content of page header in ``h2`` tag.
Default implementation returns the return value of :py:func:`hawat.view.BaseView.get_menu_title`
method by default.
:param dict kwargs: Optional parameters.
:return: Title for the view.
:rtype: str
"""
raise NotImplementedError()
#---------------------------------------------------------------------------
[docs] @staticmethod
def has_endpoint(endpoint):
"""
Check if given routing endpoint is available within the application.
:param str endpoint: Application routing endpoint.
:return: ``True`` in case endpoint exists, ``False`` otherwise.
:rtype: bool
"""
return flask.current_app.has_endpoint(endpoint)
[docs] @staticmethod
def get_endpoint_class(endpoint, quiet = False):
"""
Get reference to view class registered to given routing endpoint.
:param str endpoint: Application routing endpoint.
:param bool quiet: Suppress the exception in case given endpoint does not exist.
:return: Reference to view class.
:rtype: class
"""
return flask.current_app.get_endpoint_class(endpoint, quiet)
[docs] @staticmethod
def can_access_endpoint(endpoint, **kwargs):
"""
Check, that the current user can access given endpoint/view.
:param str endpoint: Application routing endpoint.
:param dict kwargs: Optional endpoint parameters.
:return: ``True`` in case user can access the endpoint, ``False`` otherwise.
:rtype: bool
"""
return flask.current_app.can_access_endpoint(endpoint, **kwargs)
[docs] @staticmethod
def get_model(name):
"""
Return reference to class of given model.
:param str name: Name of the model.
"""
return flask.current_app.get_model(name)
[docs] @staticmethod
def get_resource(name):
"""
Return reference to given registered resource.
:param str name: Name of the resource.
"""
return flask.current_app.get_resource(name)
#---------------------------------------------------------------------------
@property
def logger(self):
"""
Return current application`s logger object.
"""
return flask.current_app.logger
[docs]class FileNameView(BaseView):
"""
Base class for direct file access views. These views can be used to access
and serve files from arbitrary filesystem directories (that are accessible to
application process). This can be very usefull for serving files like charts,
that are periodically generated into configurable and changeable location.
"""
[docs] @classmethod
def get_view_icon(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_icon`."""
return 'action-download'
[docs] @classmethod
def get_directory_path(cls):
"""
Return absolute path to the directory, that will be used as a base path
for serving files.
*This method does not have any default implementation and must be overridden
by a subclass.*
:return: Absolute path to the directory for serving files.
:rtype: str
"""
raise NotImplementedError()
[docs] @classmethod
def validate_filename(cls, filename):
"""
Validate given file name to prevent user from accessing restricted files.
In default implementation all files pass the validation.
:param str filename: Name of the file to be validated/filtered.
:return: ``True`` in case file name is allowed, ``False`` otherwise.
:rtype: bool
"""
return bool(filename)
[docs] def dispatch_request(self, filename): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
"""
if not self.validate_filename(filename):
flask.abort(400)
self.logger.info(
"Serving file '{}' from directory '{}'.".format(
filename,
self.get_directory_path()
)
)
return flask.send_from_directory(
self.get_directory_path(),
filename,
as_attachment = True
)
[docs]class FileIdView(BaseView):
"""
Base class for indirrect file access views. These views can be used to access
and serve files from arbitrary filesystem directories (that are accessible to
application process). This can be very usefull for serving files like charts,
that are periodically generated into configurable and changeable location.
The difference between this view class and :py:class:`FileNameView` is,
that is this case some kind of identifier is used to access the file and
provided class method is responsible for translating this identifier into
real file name.
"""
[docs] @classmethod
def get_view_icon(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_icon`."""
return 'action-download'
[docs] @classmethod
def get_directory_path(cls, fileid, filetype):
"""
This method must return absolute path to the directory, that will be
used as a base path for serving files. Parameter ``fileid`` may be used
internally to further customize the base directory, for example when
serving some files places into subdirectories based on some part of the
file name (for example to reduce total number of files in base directory).
*This method does not have any default implementation and must be overridden
by a subclass.*
:param str fileid: Identifier of the requested file.
:param str filetype: Type of the requested file.
:return: Absolute path to the directory for serving files.
:rtype: str
"""
raise NotImplementedError()
[docs] @classmethod
def get_filename(cls, fileid, filetype):
"""
This method must return actual name of the file based on given identifier
and type.
*This method does not have any default implementation and must be overridden
by a subclass.*
:param str fileid: Identifier of the requested file.
:param str filetype: Type of the requested file.
:return: Translated name of the file.
:rtype: str
"""
raise NotImplementedError()
[docs] def dispatch_request(self, fileid, filetype): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
"""
basedirpath = self.get_directory_path(fileid, filetype)
filename = self.get_filename(fileid, filetype)
if not basedirpath or not filename:
flask.abort(400)
self.logger.info(
"Serving file '{}' from directory '{}'.".format(
filename,
basedirpath
)
)
return flask.send_from_directory(
basedirpath,
filename,
as_attachment = True
)
[docs]class RenderableView(BaseView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for all views, that are rendering content based on Jinja2 templates
or returning JSON/XML data.
"""
def __init__(self):
self.response_context = {}
[docs] def mark_time(self, ident, threshold, tag = 'default', label = 'Time mark', log = False):
"""
Mark current time with given identifier and label for further analysis.
This method can be usefull for measuring durations of various operations.
"""
mark = [datetime.datetime.utcnow(), ident, threshold, tag, label]
marks = self.response_context.setdefault('time_marks', [])
marks.append(mark)
if log:
if len(marks) <= 1:
self.logger.info(
'Mark {}:{} ({})'.format(*mark[1:])
)
else:
self.logger.info(
'Mark {}:{}:{} ({});delta={};delta0={}'.format(
*mark[1:],
(str(marks[-1][0]-marks[-2][0])), # Time delta from last mark.
(str(marks[-1][0]-marks[0][0])) # Time delta from first mark.
)
)
[docs] @classmethod
def get_view_template(cls):
"""
Return Jinja2 template file that should be used for rendering the view
content. This default implementation works only in case the view class
was properly registered into the parent blueprint/module with
:py:func:`hawat.app.hawatBlueprint.register_view_class` method.
:return: Jinja2 template file to use to render the view.
:rtype: str
"""
if cls.module_name:
return '{}/{}.html'.format(
cls.module_name,
cls.get_view_endpoint_name()
)
raise RuntimeError("Unable to guess default view template, because module name was not yet set.")
[docs] def do_before_response(self, **kwargs): # pylint: disable=locally-disabled,unused-argument
"""
This method will be called just before generating the response. By providing
some meaningfull implementation you can use it for some simple item and
response context mangling tasks.
:param kwargs: Custom additional arguments.
"""
[docs] def generate_response(self):
"""
Generate the appropriate response from given response context.
:param dict response_context: Response context as a dictionary
"""
raise NotImplementedError()
[docs] @staticmethod
def abort(status_code, message = None):
"""
Abort request processing with HTTP status code.
"""
raise NotImplementedError()
[docs] def flash(self, message, level = 'info'):
"""
Flash information to the user.
"""
raise NotImplementedError()
[docs] def redirect(self, default_url = None, exclude_url = None):
"""
Redirect user to different location.
"""
raise NotImplementedError()
[docs]class SimpleView(RenderableView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for simple views. These are the most, well, simple views, that are
rendering single template file or directly returning some JSON/XML data without
any user parameters.
In most use cases, it should be enough to just enhance the default implementation
of :py:func:`hawat.view.RenderableView.get_response_context` to inject
some additional variables into the template.
"""
[docs] def dispatch_request(self): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
"""
redirect = self.do_before_response() # pylint: disable=assignment-from-no-return
if redirect:
return redirect
return self.generate_response()
[docs]class BaseLoginView(SimpleView):
"""
Base class for login views.
"""
is_sign_in = True
[docs] @classmethod
def get_view_name(cls):
return hawat.const.ACTION_USER_LOGIN
[docs] @classmethod
def get_view_icon(cls):
return hawat.const.ACTION_USER_LOGIN
[docs] @classmethod
def get_view_title(cls, **kwargs):
return gettext('Login')
@property
def dbsession(self):
"""
This property contains the reference to current *SQLAlchemy* database session.
"""
raise NotImplementedError()
@property
def dbmodel(self):
"""
This property must be implemented in each subclass to
return reference to appropriate model class based on *SQLAlchemy* declarative
base.
"""
raise NotImplementedError()
[docs] def fetch(self, item_id, fetch_by = None):
"""
Perform actual search with given query.
"""
raise NotImplementedError()
[docs] def get_user_login(self):
"""
Get login of the user that is being authenticated.
"""
raise NotImplementedError()
[docs] def authenticate_user(self, user): # pylint: disable=locally-disabled,unused-argument
"""
Authenticate given user.
"""
return True
[docs] def dispatch_request(self):
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the *Flask* framework to service the request.
"""
if flask_login.current_user.is_authenticated:
return self.redirect(
default_url = flask.url_for(
flask.current_app.config['ENDPOINT_LOGIN_REDIRECT']
)
)
user_login = self.get_user_login()
if user_login:
try:
user = self.fetch(user_login)
except sqlalchemy.orm.exc.MultipleResultsFound:
self.logger.error(
"Multiple results found for user login '{}'.".format(
user_login
)
)
flask.abort(500)
except sqlalchemy.orm.exc.NoResultFound:
self.flash(
gettext('You have entered wrong login credentials.'),
hawat.const.FLASH_FAILURE
)
self.abort(403)
except Exception: # pylint: disable=locally-disabled,broad-except
self.flash(
flask.Markup(gettext(
"Unable to perform login as <strong>%(user)s</strong>.",
user = flask.escape(str(user_login))
)),
hawat.const.FLASH_FAILURE
)
flask.current_app.log_exception_with_label(
traceback.TracebackException(*sys.exc_info()),
'Unable to perform login.',
)
self.abort(500)
if not user:
self.flash(
gettext('You have entered wrong login credentials.'),
hawat.const.FLASH_FAILURE
)
self.abort(403)
if not user.enabled:
self.flash(
flask.Markup(gettext(
'Your user account <strong>%(login)s (%(name)s)</strong> is currently disabled, you are not permitted to log in.',
login = flask.escape(user.login),
name = flask.escape(user.fullname)
)),
hawat.const.FLASH_FAILURE
)
self.abort(403)
if not self.authenticate_user(user):
self.flash(
gettext('You have used wrong login credentials.'),
hawat.const.FLASH_FAILURE
)
self.abort(403)
flask_login.login_user(user)
# Tell Flask-Principal the identity changed. Access to private method
# _get_current_object is according to the Flask documentation:
# http://flask.pocoo.org/docs/1.0/reqcontext/#notes-on-proxies
flask_principal.identity_changed.send(
flask.current_app._get_current_object(), # pylint: disable=locally-disabled,protected-access
identity = flask_principal.Identity(user.get_id())
)
self.flash(
flask.Markup(gettext(
'You have been successfully logged in as <strong>%(user)s</strong>.',
user = flask.escape(str(user))
)),
hawat.const.FLASH_SUCCESS
)
self.logger.info(
"User '{}' successfully logged in with '{}'.".format(
user.login,
self.module_name
)
)
# Redirect user back to original page.
return self.redirect(
default_url = flask.url_for(
flask.current_app.config['ENDPOINT_LOGIN_REDIRECT']
)
)
self.response_context.update(
next = hawat.forms.get_redirect_target()
)
return self.generate_response()
[docs]class BaseSearchView(RenderableView, HawatUtils):
"""
Base class for search views.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'search'
[docs] @classmethod
def get_view_icon(cls):
"""*Implementation* of :py:func:`mydojo.base.BaseView.get_view_name`."""
return 'action-search'
#---------------------------------------------------------------------------
[docs] @classmethod
def get_quicksearch_by_time(cls):
"""
Get default list of 'by time' quickseach items.
"""
quicksearch_list = []
for item in (
[ '1h', gettext('Search for last hour')],
[ '2h', gettext('Search for last 2 hours')],
[ '3h', gettext('Search for last 3 hours')],
[ '4h', gettext('Search for last 4 hours')],
[ '6h', gettext('Search for last 6 hours')],
['12h', gettext('Search for last 12 hours')],
[ '1d', gettext('Search for last day')],
[ '2d', gettext('Search for last 2 days')],
[ '3d', gettext('Search for last 3 days')],
[ '1w', gettext('Search for last week')],
[ '2w', gettext('Search for last 2 weeks')],
[ '4w', gettext('Search for last 4 weeks')],
['12w', gettext('Search for last 12 weeks')],
[ 'td', gettext('Search for today')],
[ 'tw', gettext('Search for this week')],
[ 'tm', gettext('Search for this month')],
[ 'ty', gettext('Search for this year')],
):
try:
dt_from = cls.get_datetime_window(
item[0],
'current',
moment=datetime.datetime.now(pytz.timezone(flask.session.get('timezone', 'UTC')))
)
dt_to = cls.get_datetime_window(
item[0],
'next',
dt_from
)
quicksearch_list.append(
{
'label': item[1],
'params': {
'dt_from': dt_from.isoformat(
sep = ' '
),
'dt_to': dt_to.isoformat(
sep = ' '
),
'tiid': item[0],
'submit': gettext('Search')
}
}
)
except: # pylint: disable=locally-disabled,bare-except
pass
return quicksearch_list
[docs] @staticmethod
def get_query_parameters(form, request_args):
"""
Get query parameters by comparing contents of processed form data and
original request arguments. Result of this method can be used for generating
modified URLs back to current request. One of the use cases is the result
pager/paginator.
"""
params = {}
for arg in request_args:
if getattr(form, arg, None) and arg in request_args:
# Handle multivalue request arguments separately
# Resources:
# http://flask.pocoo.org/docs/1.0/api/#flask.Request.args
# http://werkzeug.pocoo.org/docs/0.14/datastructures/#werkzeug.datastructures.MultiDict
try:
if form.is_multivalue(arg):
params[arg] = request_args.getlist(arg)
else:
params[arg] = request_args[arg]
except AttributeError:
params[arg] = request_args[arg]
return params
[docs] def search(self, form_args):
"""
Perform actual search with given query.
"""
raise NotImplementedError()
#---------------------------------------------------------------------------
#---------------------------------------------------------------------------
[docs] def do_before_search(self, form_data): # pylint: disable=locally-disabled,unused-argument
"""
This hook method will be called before search attempt.
"""
[docs] def do_after_search(self, items): # pylint: disable=locally-disabled,unused-argument
"""
This hook method will be called after successfull search.
"""
[docs] def dispatch_request(self): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
"""
form = self.get_search_form(flask.request.args)
flask.g.search_form = form
if hawat.const.FORM_ACTION_SUBMIT in flask.request.args:
if form.validate():
form_data = form.data
self.mark_time(
'preprocess',
'begin',
tag = 'search',
label = 'Begin preprocessing for "{}"'.format(flask.request.full_path),
log = True
)
self.do_before_search(form_data)
self.mark_time(
'preprocess',
'end',
tag = 'search',
label = 'Finished preprocessing for "{}"'.format(flask.request.full_path),
log = True
)
try:
self.mark_time(
'search',
'begin',
tag = 'search',
label = 'Begin searching for "{}"'.format(flask.request.full_path),
log = True
)
items = self.search(form_data)
self.mark_time(
'search',
'end',
tag = 'search',
label = 'Finished searching for "{}", items found: {}'.format(flask.request.full_path, len(items)),
log = True
)
self.response_context.update(
searched = True,
items = items,
items_count = len(items),
form_data = form_data
)
# Not all search forms support result paging.
if 'page' in form_data:
self.response_context.update(
pager_index_low = ((form_data['page'] - 1) * form_data['limit']) + 1,
pager_index_high = ((form_data['page'] - 1) * form_data['limit']) + len(items),
pager_index_limit = ((form_data['page'] - 1) * form_data['limit']) + form_data['limit']
)
self.mark_time(
'postprocess',
'begin',
tag = 'search',
label = 'Begin postprocessing for "{}"'.format(flask.request.full_path),
log = True
)
self.do_after_search(items)
self.mark_time(
'postprocess',
'end',
tag = 'search',
label = 'Finished postprocessing for "{}"'.format(flask.request.full_path),
log = True
)
except Exception as err: # pylint: disable=locally-disabled,broad-except
match = re.match('invalid IP4R value: "([^"]+)"', str(err))
if match:
self.flash(
flask.Markup(
gettext(
'Invalid address value <strong>%(address)s</strong> in search form.',
address = flask.escape(str(match.group(1)))
)
),
hawat.const.FLASH_FAILURE
)
else:
raise
else:
self.response_context.update(
form_errors = [(field_name, err) for field_name, error_messages in form.errors.items() for err in error_messages]
)
self.response_context.update(
query_params = self.get_query_parameters(
form,
flask.request.args
),
search_widget_item_limit = 3
)
self.do_before_response()
self.mark_time(
'render',
'begin',
tag = 'render',
label = 'Started rendering for "{}"'.format(flask.request.full_path),
log = True
)
return self.generate_response()
[docs]class CustomSearchView(BaseSearchView):
"""
Base class for multi search views.
"""
[docs] def custom_search(self, form_args):
"""
Perform actual search with given query.
"""
raise NotImplementedError()
[docs] def do_after_search(self, **kwargs): # pylint: disable=locally-disabled,unused-argument
"""
This hook method will be called after successfull search.
"""
[docs] def dispatch_request(self): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
"""
form = self.get_search_form(flask.request.args)
flask.g.search_form = form
if hawat.const.FORM_ACTION_SUBMIT in flask.request.args:
if form.validate():
form_data = form.data
self.mark_time(
'preprocess',
'begin',
tag = 'search',
label = 'Begin preprocessing for "{}"'.format(flask.request.full_path),
log = True
)
self.do_before_search(form_data)
self.mark_time(
'preprocess',
'end',
tag = 'search',
label = 'Finished preprocessing for "{}"'.format(flask.request.full_path),
log = True
)
try:
self.mark_time(
'search',
'begin',
tag = 'search',
label = 'Begin custom-searching for "{}"'.format(flask.request.full_path),
log = True
)
self.custom_search(form_data)
self.mark_time(
'search',
'end',
tag = 'search',
label = 'Finished custom-searching for "{}"'.format(flask.request.full_path),
log = True
)
self.response_context.update(
searched = True,
form_data = form_data
)
self.mark_time(
'postprocess',
'begin',
tag = 'search',
label = 'Begin postprocessing for "{}"'.format(flask.request.full_path),
log = True
)
self.do_after_search()
self.mark_time(
'postprocess',
'end',
tag = 'search',
label = 'Finished postprocessing for "{}"'.format(flask.request.full_path),
log = True
)
except Exception as err: # pylint: disable=locally-disabled,broad-except
match = re.match('invalid IP4R value: "([^"]+)"', str(err))
if match:
self.flash(
flask.Markup(
gettext(
'Invalid address value <strong>%(address)s</strong> in search form.',
address = flask.escape(str(match.group(1)))
)
),
hawat.const.FLASH_FAILURE
)
else:
raise
else:
self.response_context.update(
form_errors = [(field_name, err) for field_name, error_messages in form.errors.items() for err in error_messages]
)
self.response_context.update(
query_params = self.get_query_parameters(
form,
flask.request.args
),
search_widget_item_limit = 3
)
self.do_before_response()
self.mark_time(
'render',
'begin',
tag = 'render',
label = 'Started rendering for "{}"'.format(flask.request.full_path),
log = True
)
return self.generate_response()
[docs]class ItemListView(RenderableView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item *list* views. These views provide quick and simple access
to lists of all objects.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'list'
[docs] @staticmethod
def get_query_parameters(form, request_args):
"""
Get query parameters by comparing contents of processed form data and
original request arguments. Result of this method can be used for generating
modified URLs back to current request. One of the use cases is the result
pager/paginator.
"""
params = {}
for arg in request_args:
if form and getattr(form, arg, None) and arg in request_args:
# Handle multivalue request arguments separately
# Resources:
# http://flask.pocoo.org/docs/1.0/api/#flask.Request.args
# http://werkzeug.pocoo.org/docs/0.14/datastructures/#werkzeug.datastructures.MultiDict
try:
if form.is_multivalue(arg):
params[arg] = request_args.getlist(arg)
else:
params[arg] = request_args[arg]
except AttributeError:
params[arg] = request_args[arg]
return params
[docs] def search(self, form_args):
"""
Perform actual search with given form arguments.
"""
raise NotImplementedError()
[docs] def dispatch_request(self): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
List of all items will be retrieved from database and injected into template
to be displayed to the user.
"""
form = self.get_search_form(flask.request.args)
flask.g.search_form = form
if form:
form_data = form.data
items = []
if form.validate():
form_data = form.data
items = self.search(form_data)
self.response_context.update(
form_data = form_data,
searched = True,
items = items,
items_count = len(items)
)
# Not all search forms support result paging.
if 'page' in form_data:
self.response_context.update(
pager_index_low = ((form_data['page'] - 1) * form_data['limit']) + 1,
pager_index_high = ((form_data['page'] - 1) * form_data['limit']) + len(items),
pager_index_limit = ((form_data['page'] - 1) * form_data['limit']) + form_data['limit']
)
else:
items = self.search({})
self.response_context.update(
items = items
)
self.response_context.update(
query_params = self.get_query_parameters(
form,
flask.request.args
),
search_widget_item_limit = 3,
form = form
)
self.do_before_response()
return self.generate_response()
[docs]class ItemShowView(RenderableView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item *show* views. These views expect unique item identifier
as parameter and are supposed to display specific information about single
item.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'show'
[docs] @classmethod
def get_view_icon(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_icon`."""
return 'action-show'
[docs] @classmethod
def get_view_title(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_title`."""
return gettext('Show')
[docs] @classmethod
def get_view_url(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_url`."""
if 'item' not in kwargs or not kwargs['item']:
raise ValueError(
"Missing item parameter for show URL view '{}'". format(
cls.get_view_endpoint()
)
)
return flask.url_for(
cls.get_view_endpoint(),
item_id = kwargs['item'].get_id()
)
[docs] @classmethod
def authorize_item_action(cls, **kwargs): # pylint: disable=locally-disabled,unused-argument
"""
Perform access authorization for current user to particular item.
"""
return True
[docs] def fetch(self, item_id):
"""
Fetch item with given ID.
"""
raise NotImplementedError()
[docs] def dispatch_request(self, item_id): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
Single item with given unique identifier will be retrieved from database
and injected into template to be displayed to the user.
"""
item = self.fetch(item_id)
if not item:
self.abort(404)
if not self.authorize_item_action(item = item):
self.abort(403)
self.response_context.update(
item_id = item_id,
item = item,
search_widget_item_limit = 100
)
self.do_before_response()
return self.generate_response()
[docs]class ItemActionView(RenderableView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item action views. These views perform various actions
(create/update/delete) with given item class.
"""
[docs] @classmethod
def get_view_icon(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_icon`."""
return 'action-{}'.format(
cls.get_view_name().replace('_', '-')
)
[docs] @classmethod
def get_view_url(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_url`."""
return flask.url_for(
cls.get_view_endpoint(),
item_id = kwargs['item'].get_id()
)
[docs] @classmethod
def get_view_template(cls):
"""*Implementation* of :py:func:`hawat.view.RenderableView.get_view_template`."""
return 'form_{}.html'.format(
cls.get_view_name().replace('-', '_')
)
[docs] @staticmethod
def get_message_success(**kwargs):
"""
*Hook method*. Must return text for flash message in case of action *success*.
The text may contain HTML characters and will be passed to :py:class:`flask.Markup`
before being used, so to certain extend you may emphasize and customize the output.
"""
raise NotImplementedError()
[docs] @staticmethod
def get_message_failure(**kwargs):
"""
*Hook method*. Must return text for flash message in case of action *failure*.
The text may contain HTML characters and will be passed to :py:class:`flask.Markup`
before being used, so to certain extend you may emphasize and customize the output.
"""
raise NotImplementedError()
[docs] @staticmethod
def get_message_cancel(**kwargs):
"""
*Hook method*. Must return text for flash message in case of action *cancel*.
The text may contain HTML characters and will be passed to :py:class:`flask.Markup`
before being used, so to certain extend you may emphasize and customize the output.
"""
raise NotImplementedError()
[docs] def get_url_next(self): # pylint: disable=locally-disabled
"""
*Hook method*. Must return URL for redirection after action *success*. In
most cases there should be call for :py:func:`flask.url_for` function
somewhere in this method.
"""
return None
[docs] def check_action_cancel(self, form, **kwargs):
"""
*Hook method*. Check the form for *cancel* button press and cancel the action.
"""
if hasattr(form, hawat.const.FORM_ACTION_CANCEL):
if getattr(form, hawat.const.FORM_ACTION_CANCEL).data:
self.flash(
flask.Markup(self.get_message_cancel(**kwargs)),
hawat.const.FLASH_INFO
)
return self.redirect(
default_url = self.get_url_next()
)
return None
[docs] def do_before_action(self, item): # pylint: disable=locally-disabled,unused-argument
"""
*Hook method*. Will be called before any action handling tasks.
"""
[docs] def do_after_action(self, item): # pylint: disable=locally-disabled,unused-argument
"""
*Hook method*. Will be called after successfull action handling tasks.
"""
[docs] @classmethod
def authorize_item_action(cls, **kwargs): # pylint: disable=locally-disabled,unused-argument
"""
Perform access authorization for current user to particular item.
"""
return True
@property
def dbsession(self):
"""
This property contains the reference to current *SQLAlchemy* database session.
"""
raise NotImplementedError()
@property
def dbmodel(self):
"""
This property must be implemented in each subclass to
return reference to appropriate model class based on *SQLAlchemy* declarative
base.
"""
raise NotImplementedError()
@property
def dbchlogmodel(self):
"""
This property must be implemented in each subclass to
return reference to appropriate model class based on *SQLAlchemy* declarative
base.
"""
raise NotImplementedError()
[docs] def fetch(self, item_id, fetch_by = None):
"""
Perform actual search with given query.
"""
raise NotImplementedError()
[docs] def changelog_log(self, item, json_state_before = '', json_state_after = ''):
"""
Log item action into changelog. One of the method arguments is permitted
to be left out. This enables logging create and delete actions.
:param hawat.db.MODEL item: Item that is being changed.
:param str json_state_before: JSON representation of item state before action.
:param str json_state_after: JSON representation of item state after action.
"""
if not json_state_before and not json_state_after:
raise ValueError("Invalid use of changelog_log() method, both of the item states are null.")
# 'Author' may be an instance of 'AnonymousUserMixin' for actions performed by
# anonymous users. In that case store 'Null' into database.
author = flask_login.current_user._get_current_object() # pylint: disable=locally-disabled,protected-access
if not isinstance(author, self.get_model(hawat.const.MODEL_USER)):
author = None
chlog = self.dbchlogmodel(
author = author,
model = item.__class__.__name__,
model_id = item.id,
endpoint = self.get_view_endpoint(),
module = self.module_name,
operation = self.get_view_name(),
before = json_state_before,
after = json_state_after
)
chlog.calculate_diff()
self.dbsession.add(chlog)
self.dbsession.commit()
[docs] def get_affected_items(self, item, form):
"""
Return dict of hawat.db.MODEL items affected by the change of the item
as keys and their JSON representation before change as value.
(e.g. dict of groups which lost a member after a user item was deleted)
:param hawat.db.MODEL item: Item that is being changed.
:param flask_wtf.FlaskForm form: Form representing the change.
"""
changed = set()
for attribute in ["memberships", "memberships_wanted", "managements", "members", "members_wanted", "managers"]:
if hasattr(item, attribute):
changed.update(getattr(item, attribute))
if hasattr(form, attribute):
changed.update(getattr(form, attribute).data)
return { obj: obj.to_json() for obj in changed }
[docs] def handle_error(self, **kwargs):
"""
Handle and log the error, rollback all database changes
and show the failure message to user.
"""
self.dbsession.rollback()
self.flash(
flask.Markup(self.get_message_failure(**kwargs)),
hawat.const.FLASH_FAILURE
)
flask.current_app.log_exception_with_label(
traceback.TracebackException(*sys.exc_info()),
self.get_message_failure(**kwargs)
)
return self.redirect(default_url = self.get_url_next())
[docs]class ItemCreateView(ItemActionView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item *create* action views. These views create new items in
database.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'create'
[docs] @classmethod
def get_view_template(cls):
"""
Return Jinja2 template file that should be used for rendering the view
content. This default implementation works only in case the view class
was properly registered into the parent blueprint/module with
:py:func:`hawat.app.hawatBlueprint.register_view_class` method.
:return: Title for the view.
:rtype: str
"""
if cls.module_name:
return '{}/creatupdate.html'.format(cls.module_name)
raise RuntimeError("Unable to guess default view template, because module name was not yet set.")
[docs] @classmethod
def get_view_title(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_title`."""
return gettext('Create')
[docs] @classmethod
def get_view_url(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_url`."""
return flask.url_for(cls.get_view_endpoint())
[docs] def get_item(self):
"""
*Hook method*. Must return instance for given item class.
"""
return self.dbmodel()
[docs] @staticmethod
def get_message_duplicate(**kwargs):
"""
*Hook method*. Must return text for flash message in case of action *failure*.
The text may contain HTML characters and will be passed to :py:class:`flask.Markup`
before being used, so to certain extend you may emphasize and customize the output.
"""
return gettext(
'Item "%(item)s" already exists',
item = flask.escape(str(kwargs['item']))
)
[docs] def dispatch_request(self): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
This method will attempt to validate the submitted form and create new
instance of appropriate item from form data and finally store the item
into the database.
"""
if not self.authorize_item_action():
self.abort(403)
item = self.get_item()
self.response_context.update(item = item)
form = self.get_item_form(item)
self.response_context.update(form = form)
cancel_response = self.check_action_cancel(form)
if cancel_response:
return cancel_response
if form.validate_on_submit():
affected_items = self.get_affected_items(item, form)
form_data = form.data
self.response_context.update(form_data = form_data)
form.populate_obj(item)
try:
self.do_before_action(item)
except Exception:
return self.handle_error(item = item)
if form_data[hawat.const.FORM_ACTION_SUBMIT]:
try:
self.dbsession.add(item)
self.dbsession.commit()
self.do_after_action(item)
# Log the item creation into changelog.
self.changelog_log(item, '', item.to_json())
# Log changes of all affected items into changelog.
for affected_item, json_before in affected_items.items():
self.changelog_log(affected_item, json_before, affected_item.to_json())
self.flash(
flask.Markup(
self.get_message_success(item = item)
),
hawat.const.FLASH_SUCCESS
)
return self.redirect(default_url = self.get_url_next())
except sqlalchemy.exc.IntegrityError:
self.dbsession.rollback()
self.flash(
flask.Markup(
self.get_message_duplicate(item = item)
),
hawat.const.FLASH_FAILURE
)
return self.redirect(default_url = self.get_url_next())
except Exception: # pylint: disable=locally-disabled,broad-except
return self.handle_error(item = item)
self.response_context.update(
action_name = gettext('Create'),
form_url = flask.url_for(self.get_view_endpoint()),
item_action = hawat.const.ACTION_ITEM_CREATE,
item_type = self.dbmodel.__name__.lower()
)
self.do_before_response()
return self.generate_response()
[docs]class ItemCreateForView(ItemActionView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item *createfor* action views. These views differ a little bit
from *create* action views. They are used to create new items within database,
but only for particular defined parent item. One example use case is creating
network records for particular abuse group.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'createfor'
[docs] @classmethod
def get_view_icon(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_icon`."""
return 'module-{}'.format(cls.module_name)
[docs] @classmethod
def get_view_url(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_url`."""
return flask.url_for(
cls.get_view_endpoint(),
parent_id = kwargs['parent'].id
)
[docs] @classmethod
def get_view_template(cls):
"""
Return Jinja2 template file that should be used for rendering the view
content. This default implementation works only in case the view class
was properly registered into the parent blueprint/module with
:py:func:`hawat.app.hawatBlueprint.register_view_class` method.
:return: Title for the view.
:rtype: str
"""
if cls.module_name:
return '{}/creatupdate.html'.format(cls.module_name)
raise RuntimeError("Unable to guess default view template, because module name was not yet set.")
@property
def dbmodel_par(self):
"""
*Hook property*. This property must be implemented in each subclass to
return reference to appropriate model class for parent objects and that
is based on *SQLAlchemy* declarative base.
"""
raise NotImplementedError()
@property
def dbquery_par(self):
"""
This property contains the reference to *SQLAlchemy* query object appropriate
for particular ``dbmodel_par`` property.
"""
return self.dbsession.query(self.dbmodel_par)
[docs] def get_item(self):
"""
*Hook method*. Must return instance for given item class.
"""
return self.dbmodel()
[docs] @staticmethod
def add_parent_to_item(item, parent):
"""
*Hook method*. Use given parent object for given item object. The actual
operation to realize this relationship is highly dependent on current
circumstance. It is up to the developer to perform correct set of actions
to implement parent - child relationship for particular object types.
"""
raise NotImplementedError()
[docs] @staticmethod
def get_message_duplicate(**kwargs):
"""
*Hook method*. Must return text for flash message in case of action *failure*.
The text may contain HTML characters and will be passed to :py:class:`flask.Markup`
before being used, so to certain extend you may emphasize and customize the output.
"""
return gettext(
'Item "%(item)s" already exists',
item = flask.escape(str(kwargs['item']))
)
[docs] def dispatch_request(self, parent_id): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
This method will attempt to validate the submitted form and create new
instance of appropriate item from form data and finally store the item
into the database.
"""
parent = self.dbquery_par.filter(self.dbmodel_par.id == parent_id).one_or_none()
if not parent:
self.abort(404)
if not self.authorize_item_action(item = parent):
self.abort(403)
self.response_context.update(
parent_id = parent_id,
parent = parent
)
item = self.get_item()
form = self.get_item_form(item)
cancel_response = self.check_action_cancel(form, parent = parent)
if cancel_response:
return cancel_response
if form.validate_on_submit():
affected_items = self.get_affected_items(item, form)
form_data = form.data
self.response_context.update(form_data = form_data)
form.populate_obj(item)
self.add_parent_to_item(item, parent)
try:
self.do_before_action(item)
except Exception:
return self.handle_error(parent = parent)
if form_data[hawat.const.FORM_ACTION_SUBMIT]:
try:
self.dbsession.add(item)
self.dbsession.commit()
self.do_after_action(item)
# Log the item creation into changelog.
self.changelog_log(item, '', item.to_json())
# Log changes of all affected items into changelog.
for affected_item, json_before in affected_items.items():
self.changelog_log(affected_item, json_before, affected_item.to_json())
self.flash(
flask.Markup(self.get_message_success(item = item, parent = parent)),
hawat.const.FLASH_SUCCESS
)
return self.redirect(default_url = self.get_url_next())
except sqlalchemy.exc.IntegrityError:
self.dbsession.rollback()
self.flash(
flask.Markup(self.get_message_duplicate(item = item)),
hawat.const.FLASH_FAILURE
)
return self.redirect(default_url = self.get_url_next())
except Exception: # pylint: disable=locally-disabled,broad-except
return self.handle_error(parent=parent)
self.response_context.update(
action_name = gettext('Create'),
form_url = flask.url_for(self.get_view_endpoint(), parent_id = parent_id),
form = form,
item_action = hawat.const.ACTION_ITEM_CREATEFOR,
item_type = self.dbmodel.__name__.lower(),
item = item,
parent_type = self.dbmodel_par.__name__.lower(),
parent = parent
)
self.do_before_response()
return self.generate_response()
[docs]class BaseRegisterView(ItemCreateView): # pylint: disable=locally-disabled,abstract-method
"""
View responsible for registering new user account into application.
"""
methods = ['GET', 'POST']
is_sign_up = True
[docs] @classmethod
def get_view_name(cls):
return hawat.const.ACTION_USER_REGISTER
[docs] @classmethod
def get_view_icon(cls):
return hawat.const.ACTION_USER_REGISTER
[docs] @classmethod
def get_view_title(cls, **kwargs):
return gettext('User account registration')
[docs] @classmethod
def get_view_template(cls):
return '{}/registration.html'.format(cls.module_name)
[docs] @staticmethod
def get_message_success(**kwargs):
return gettext(
'User account <strong>%(login)s (%(name)s)</strong> was successfully registered.',
login = kwargs['item'].login,
name = kwargs['item'].fullname
)
[docs] @staticmethod
def get_message_failure(**kwargs):
return gettext('Unable to register new user account.')
[docs] @staticmethod
def get_message_cancel(**kwargs):
return gettext('Account registration canceled.')
[docs] @staticmethod
def get_message_duplicate(**kwargs):
return gettext('Please use different login, the "%(item)s" is already taken.', item = str(kwargs['item']))
[docs] def do_before_action(self, item): # pylint: disable=locally-disabled,unused-argument
item.roles = [hawat.const.ROLE_USER]
item.enabled = False
[docs] def do_after_action(self, item): # pylint: disable=locally-disabled,unused-argument
self.inform_admins(item, self.response_context['form_data'])
self.inform_managers(item, self.response_context['form_data'])
self.inform_user(item, self.response_context['form_data'])
[docs]class ItemUpdateView(ItemActionView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item *update* action views. These views update existing items
in database.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'update'
[docs] @classmethod
def get_view_template(cls):
"""
Return Jinja2 template file that should be used for rendering the view
content. This default implementation works only in case the view class
was properly registered into the parent blueprint/module with
:py:func:`hawat.app.hawatBlueprint.register_view_class` method.
:return: Title for the view.
:rtype: str
"""
if cls.module_name:
return '{}/creatupdate.html'.format(cls.module_name)
raise RuntimeError("Unable to guess default view template, because module name was not yet set.")
[docs] @classmethod
def get_view_title(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_title`."""
return gettext('Update')
[docs] def get_affected_items(self, item, form):
"""
Return dict of hawat.db.MODEL items affected by the update of the item
as keys and their JSON representation before update as value.
(e.g. dict of groups which lost a member after a user item was updated)
Overrides get_affected_items(self, item, form) in ItemActionView.
:param hawat.db.MODEL item: Item that is being changed.
:param flask_wtf.FlaskForm form: Form representing the change.
"""
changed = set()
for attribute in ["memberships", "memberships_wanted", "managements", "members", "members_wanted", "managers"]:
if hasattr(item, attribute) and hasattr(form, attribute):
changed.update(set(getattr(item, attribute)).symmetric_difference(set(getattr(form, attribute).data)))
return { obj: obj.to_json() for obj in changed }
[docs] def dispatch_request(self, item_id): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
This method will attempt to validate the submitted form and update the
instance of appropriate item from form data and finally store the item
back into the database.
"""
item = self.fetch(item_id)
if not item:
self.abort(404)
if not self.authorize_item_action(item = item):
self.abort(403)
self.dbsession.add(item)
form = self.get_item_form(item)
cancel_response = self.check_action_cancel(form, item = item)
if cancel_response:
return cancel_response
item_json_before = item.to_json()
if form.validate_on_submit():
affected_items = self.get_affected_items(item, form)
form_data = form.data
self.response_context.update(form_data = form_data)
form.populate_obj(item)
try:
self.do_before_action(item)
except Exception:
return self.handle_error(item = item)
if form_data[hawat.const.FORM_ACTION_SUBMIT]:
try:
if item not in self.dbsession.dirty:
self.flash(
gettext('No changes detected, no update needed.'),
hawat.const.FLASH_INFO
)
return self.redirect(default_url = self.get_url_next())
self.dbsession.commit()
self.do_after_action(item)
# Log the item update into changelog.
self.changelog_log(item, item_json_before, item.to_json())
# Log changes of all affected items into changelog.
for affected_item, json_before in affected_items.items():
self.changelog_log(affected_item, json_before, affected_item.to_json())
self.flash(
flask.Markup(self.get_message_success(item = item)),
hawat.const.FLASH_SUCCESS
)
return self.redirect(default_url = self.get_url_next())
except Exception: # pylint: disable=locally-disabled,broad-except
return self.handle_error(item = item)
self.response_context.update(
action_name = gettext('Update'),
form_url = flask.url_for(self.get_view_endpoint(), item_id = item_id),
form = form,
item_action = hawat.const.ACTION_ITEM_UPDATE,
item_type = self.dbmodel.__name__.lower(),
item_id = item_id,
item = item
)
self.do_before_response()
return self.generate_response()
[docs]class ItemDeleteView(ItemActionView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item *delete* action views. These views delete existing items
from database.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'delete'
[docs] @classmethod
def get_view_title(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_title`."""
return gettext('Delete')
[docs] def dispatch_request(self, item_id): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
This method will attempt to validate the submitted form and delete the
instance of appropriate item from database in case user agreed to the
item removal action.
"""
item = self.fetch(item_id)
if not item:
self.abort(404)
if not self.authorize_item_action(item = item):
self.abort(403)
form = ItemActionConfirmForm()
cancel_response = self.check_action_cancel(form, item = item)
if cancel_response:
return cancel_response
item_json_before = item.to_json()
if form.validate_on_submit():
affected_items = self.get_affected_items(item, form)
form_data = form.data
self.response_context.update(form_data = form_data)
try:
self.do_before_action(item)
except Exception:
return self.handle_error(item=item)
if form_data[hawat.const.FORM_ACTION_SUBMIT]:
try:
self.dbsession.delete(item)
self.dbsession.commit()
self.do_after_action(item)
# Log the item deletion into changelog.
self.changelog_log(item, item_json_before, '')
# Log changes of all affected items into changelog.
for affected_item, json_before in affected_items.items():
self.changelog_log(affected_item, json_before, affected_item.to_json())
self.flash(
flask.Markup(self.get_message_success(item = item)),
hawat.const.FLASH_SUCCESS
)
return self.redirect(
default_url = self.get_url_next(),
exclude_url = flask.url_for(
'{}.{}'.format(self.module_name, 'show'),
item_id = item.id
)
)
except Exception: # pylint: disable=locally-disabled,broad-except
return self.handle_error(item = item)
self.response_context.update(
confirm_form = form,
confirm_url = flask.url_for(self.get_view_endpoint(), item_id = item_id),
item_name = str(item),
item_id = item_id,
item = item
)
self.do_before_response()
return self.generate_response()
[docs]class ItemChangeView(ItemActionView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for single item change views, that are doing some simple modification
of item attribute, like enable/disable item, etc.
"""
[docs] @classmethod
def validate_item_change(cls, **kwargs): # pylint: disable=locally-disabled,unused-argument
"""
Perform validation of particular change to given item.
"""
return True
[docs] @classmethod
def change_item(cls, **kwargs):
"""
*Hook method*: Change given item in any desired way.
:param item: Item to be changed/modified.
"""
raise NotImplementedError()
[docs] def dispatch_request(self, item_id): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
This method will attempt to validate the submitted form, then perform
arbitrary mangling action with the item and submit the changes to the
database.
"""
item = self.fetch(item_id)
if not item:
self.abort(404)
if not self.authorize_item_action(item = item):
self.abort(403)
if not self.validate_item_change(item = item):
self.abort(400)
form = ItemActionConfirmForm()
cancel_response = self.check_action_cancel(form, item = item)
if cancel_response:
return cancel_response
item_json_before = item.to_json()
if form.validate_on_submit():
form_data = form.data
self.response_context.update(form_data = form_data)
try:
self.do_before_action(item)
except Exception:
return self.handle_error(item=item)
if form_data[hawat.const.FORM_ACTION_SUBMIT]:
try:
self.change_item(item = item)
if item not in self.dbsession.dirty:
self.flash(
gettext('No changes detected, no update needed.'),
hawat.const.FLASH_INFO
)
return self.redirect(default_url = self.get_url_next())
self.dbsession.commit()
self.do_after_action(item)
# Log the item change into changelog.
self.changelog_log(item, item_json_before, item.to_json())
self.flash(
flask.Markup(self.get_message_success(item = item)),
hawat.const.FLASH_SUCCESS
)
return self.redirect(
default_url = self.get_url_next()
)
except Exception: # pylint: disable=locally-disabled,broad-except
return self.handle_error(item = item)
self.response_context.update(
confirm_form = form,
confirm_url = flask.url_for(self.get_view_endpoint(), item_id = item_id),
item_name = str(item),
item_id = item_id,
item = item
)
self.do_before_response()
return self.generate_response()
[docs]class ItemDisableView(ItemChangeView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item disabling views.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'disable'
[docs] @classmethod
def get_view_title(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_title`."""
return gettext('Disable')
[docs] @classmethod
def validate_item_change(cls, **kwargs): # pylint: disable=locally-disabled,unused-argument
"""*Implementation* of :py:func:`hawat.view.ItemChangeView.validate_item_change`."""
# Reject item change in case given item is already disabled.
if not kwargs['item'].enabled:
return False
return True
[docs] @classmethod
def change_item(cls, **kwargs):
"""
*Implementation* of :py:func:`hawat.view.ItemChangeView.change_item`.
"""
kwargs['item'].enabled = False
[docs]class ItemEnableView(ItemChangeView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item enabling views.
"""
[docs] @classmethod
def get_view_name(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_name`."""
return 'enable'
[docs] @classmethod
def get_view_title(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_title`."""
return gettext('Enable')
[docs] @classmethod
def validate_item_change(cls, **kwargs): # pylint: disable=locally-disabled,unused-argument
"""*Implementation* of :py:func:`hawat.view.ItemChangeView.validate_item_change`."""
# Reject item change in case given item is already enabled.
if kwargs['item'].enabled:
return False
return True
[docs] @classmethod
def change_item(cls, **kwargs):
"""
*Implementation* of :py:func:`hawat.view.ItemChangeView.change_item`.
"""
kwargs['item'].enabled = True
[docs]class ItemObjectRelationView(ItemChangeView): # pylint: disable=locally-disabled,abstract-method
"""
Base class for item object relation action views.
"""
[docs] @classmethod
def get_view_icon(cls):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_icon`."""
return 'module-{}'.format(cls.module_name)
[docs] @classmethod
def get_view_template(cls):
"""
Return Jinja2 template file that should be used for rendering the view
content. This default implementation works only in case the view class
was properly registered into the parent blueprint/module with
:py:func:`hawat.app.hawatBlueprint.register_view_class` method.
:return: Title for the view.
:rtype: str
"""
if cls.module_name:
return '{}/{}.html'.format(cls.module_name, cls.get_view_endpoint_name())
raise RuntimeError("Unable to guess default view template, because module name was not yet set.")
[docs] @classmethod
def get_view_url(cls, **kwargs):
"""*Implementation* of :py:func:`hawat.view.BaseView.get_view_url`."""
return flask.url_for(
cls.get_view_endpoint(),
item_id = kwargs['item'].get_id(),
other_id = kwargs['other'].get_id()
)
@property
def dbmodel_other(self):
"""
*Hook property*. This property must be implemented in each subclass to
return reference to appropriate model class for other objects and that
is based on *SQLAlchemy* declarative base.
"""
raise NotImplementedError()
@property
def dbquery_other(self):
"""
This property contains the reference to *SQLAlchemy* query object appropriate
for particular ``dbmodel_other`` property.
"""
return self.dbsession.query(self.dbmodel_other)
[docs] def dispatch_request(self, item_id, other_id): # pylint: disable=locally-disabled,arguments-differ
"""
Mandatory interface required by the :py:func:`flask.views.View.dispatch_request`.
Will be called by the **Flask** framework to service the request.
This method will attempt to validate the submitted form and create new
instance of appropriate item from form data and finally store the item
into the database.
"""
item = self.fetch(item_id)
if not item:
self.abort(404)
other = self.dbquery_other.filter(self.dbmodel_other.id == other_id).first()
if not other:
self.abort(404)
if not self.authorize_item_action(item = item, other = other):
self.abort(403)
if not self.validate_item_change(item = item, other = other):
self.abort(400)
form = ItemActionConfirmForm()
cancel_response = self.check_action_cancel(form, item = item, other = other)
if cancel_response:
return cancel_response
item_json_before = item.to_json()
other_json_before = other.to_json()
if form.validate_on_submit():
form_data = form.data
self.response_context.update(form_data = form_data)
self.do_before_action(item)
if form_data[hawat.const.FORM_ACTION_SUBMIT]:
try:
self.change_item(item = item, other = other)
if item not in self.dbsession.dirty:
self.flash(
gettext('No changes detected, no update needed.'),
hawat.const.FLASH_INFO
)
return self.redirect(default_url = self.get_url_next())
self.dbsession.commit()
self.do_after_action(item)
# Log changes of 'item' and 'other' into changelog.
self.changelog_log(item, item_json_before, item.to_json())
self.changelog_log(other, other_json_before, other.to_json())
self.flash(
flask.Markup(
self.get_message_success(
item = item,
other = other
)
),
hawat.const.FLASH_SUCCESS
)
return self.redirect(
default_url = self.get_url_next()
)
except Exception: # pylint: disable=locally-disabled,broad-except
return self.handle_error(item = item, other = other)
self.response_context.update(
confirm_form = form,
confirm_url = flask.url_for(
'{}.{}'.format(
self.module_name,
self.get_view_name()
),
item_id = item_id,
other_id = other_id
),
item_name = str(item),
item_id = item_id,
item = item,
other_name = str(other),
other_id = other_id,
other = other
)
self.do_before_response()
return self.generate_response()