auth() - Code Metrics - Inspection of "Refactor auth checks:" - christabor/flask_jsondash - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 2193e1...87074b )

by Chris

created 2016-10-19 18:23 UTC

auth() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	4
c	1
b	0
f	0
dl	0
loc	17
rs	9.2

# -*- coding: utf-8 -*-

"""
flask_jsondash.charts_builder
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

The chart blueprint that houses all functionality.
"""

import json
import os
import uuid
from collections import namedtuple
from datetime import datetime as dt

from flask_jsondash import static, templates

from flask import (
    Blueprint,
    current_app,
    flash,
    redirect,
    render_template,
    request,
    send_from_directory,
    url_for,
)
import jinja2

import db_adapters as adapter
from settings import (
    CHARTS_CONFIG,
)

template_dir = os.path.dirname(templates.__file__)
static_dir = os.path.dirname(static.__file__)

Paginator = namedtuple('Paginator',
                       'count per_page curr_page num_pages limit skip')

charts = Blueprint(
    'jsondash',
    __name__,
    template_folder=template_dir,
    static_url_path=static_dir,
    static_folder=static_dir,
)
default_config = dict(
    JSONDASH_FILTERUSERS=False,
    JSONDASH_GLOBALDASH=False,
    JSONDASH_GLOBAL_USER='global',
    JSONDASH_PERPAGE=25,
)


def auth(**kwargs):
    """Check if general auth functions have been specified.

    Checks for either a global auth (if authtype is None), or
    an action specific auth (specified by authtype).
    """
    authtype = kwargs.pop('authtype')
    if 'JSONDASH' not in current_app.config:
        return True
    if 'auth' not in current_app.config['JSONDASH']:
        return True
    auth_conf = current_app.config.get('JSONDASH').get('auth')
    if all([authtype is not None, authtype in auth_conf]):
        # Only perform the user-supplied check
        # if the authtype is actually enabled.
        return current_app.config['JSONDASH']['auth'][authtype](**kwargs)
    return False


def metadata(key=None):
    """An abstraction around misc. metadata.

    This allows loose coupling for enabling and setting
    metadata for each chart.
    """
    _metadata = dict()
    conf = current_app.config
    conf_metadata = conf.get('JSONDASH', {}).get('metadata', None)
    # Also useful for getting arbitrary configuration keys.
    if key is not None:
        if key in conf_metadata:
            return conf_metadata[key]()
        else:
            return None
    # Update all metadata values if the function exists.
    for k, func in conf_metadata.items():
        _metadata[k] = conf_metadata[k]()
    return _metadata


def setting(name, default=None):
    """A simplified getter for namespaced flask config values."""
    if default is None:
        default = default_config.get(name)
    return current_app.config.get(name, default)


@charts.context_processor
def _ctx():
    """Inject any context needed for this blueprint."""
    filter_user = setting('JSONDASH_FILTERUSERS')
    return dict(
        charts_config=CHARTS_CONFIG,
        page_title='dashboards',
        global_dashuser=setting('JSONDASH_GLOBAL_USER'),
        global_dashboards=setting('JSONDASH_GLOBALDASH'),
        username=metadata(key='username') if filter_user else None,
        filter_dashboards=filter_user,
    )


@jinja2.contextfilter
@charts.app_template_filter('jsonstring')
def jsonstring(ctx, data):
    """Format view json module data for template use.

    It's automatically converted to unicode key/value pairs,
    which is undesirable for the template.
    """
    if 'date' in data:
        data['date'] = str(data['date'])
    return json.dumps(data)


@charts.route('/jsondash/<path:filename>')
def _static(filename):
    """Send static files directly for this blueprint."""
    return send_from_directory(static_dir, filename)


def paginator(count=None):
    """Get pagination calculations in a compact format."""
    if count is None:
        count = adapter.count()
    per_page = setting('JSONDASH_PERPAGE')
    # Allow query parameter overrides.
    per_page = int(request.args.get('per_page', 0)) or per_page
    per_page = per_page if per_page > 2 else 2  # Prevent division errors etc
    curr_page = int(request.args.get('page', 1)) - 1
    num_pages = count // per_page
    rem = count % per_page
    extra_pages = 2 if rem else 1
    pages = range(1, num_pages + extra_pages)
    return Paginator(
        limit=per_page,
        per_page=per_page,
        curr_page=curr_page,
        skip=curr_page * per_page,
        num_pages=pages,
        count=count,
    )


@charts.route('/charts/', methods=['GET'])
def dashboard():
    """Load all views."""
    opts = dict()
    views = []
    if setting('JSONDASH_FILTERUSERS'):
        opts.update(filter=dict(created_by=metadata(key='username')))
        views = list(adapter.read(**opts))
        if setting('JSONDASH_GLOBALDASH'):
            opts.update(
                filter=dict(created_by=setting('JSONDASH_GLOBAL_USER')))
            views += list(adapter.read(**opts))
    else:
        views = list(adapter.read(**opts))
    if views:
        pagination = paginator(count=len(views))
        opts.update(limit=pagination.limit, skip=pagination.skip)
    else:
        pagination = None
    kwargs = dict(
        views=views,
        view=None,
        paginator=pagination,
        total_modules=sum([len(view['modules']) for view in views]),
    )
    return render_template('pages/charts_index.html', **kwargs)


@charts.route('/charts/<id>', methods=['GET'])
def view(id):
    """Load a json view config from the DB."""
    if not auth(authtype='view'):
        flash('You do not have access to view this dashboard.', 'error')
        return redirect(url_for('jsondash.dashboard'))
    viewjson = adapter.read(c_id=id)
    if not viewjson:
        flash('Could not find view: {}'.format(id), 'error')
        return redirect(url_for('jsondash.dashboard'))
    # Remove _id, it's not JSON serializeable.
    viewjson.pop('_id')
    # Chart family is encoded in chart type value for lookup.
    active_charts = [v.get('family') for
                     v in viewjson['modules'] if v.get('family')]
    kwargs = dict(id=id, view=viewjson, active_charts=active_charts)
    return render_template('pages/chart_detail.html', **kwargs)


@charts.route('/charts/<c_id>/delete', methods=['POST'])
def delete(c_id):
    """Delete a json dashboard config."""
    dash_url = url_for('jsondash.dashboard')
    if not auth(authtype='delete'):
        flash('You do not have access to delete dashboards.', 'error')
        return redirect(dash_url)
    adapter.delete(c_id)
    flash('Deleted dashboard {}'.format(c_id))
    return redirect(dash_url)


@charts.route('/charts/update', methods=['POST'])
def update():
    """Normalize the form POST and setup the json view config object."""
    if not auth(authtype='update'):
        flash('You do not have access to update dashboards.', 'error')
        return redirect(url_for('jsondash.dashboard'))
    data = request.form
    c_id = data['id']
    view_url = url_for('jsondash.view', id=c_id)
    if 'edit-raw' in request.form:
        try:
            data = json.loads(request.form.get('config'))
            data = adapter.reformat_data(data, c_id)
            # Update db
            adapter.update(c_id, data=data, fmt_modules=False)
        except (TypeError, ValueError):
            flash('Invalid JSON config.', 'error')
            return redirect(view_url)
    else:
        # Update db
        d = dict(
            name=data['name'],
            modules=adapter._format_modules(data),
            date=dt.now(),
            id=data['id'],
        )
        d.update(**metadata())
        adapter.update(c_id, data=d)
    flash('Updated view "{}"'.format(c_id))
    return redirect(view_url)


@charts.route('/charts/create', methods=['POST'])

def create():
    """Normalize the form POST and setup the json view config object."""
    if not auth(authtype='create'):
        flash('You do not have access to create dashboards.', 'error')
        return redirect(url_for('jsondash.dashboard'))
    data = request.form
    new_id = str(uuid.uuid1())
    d = dict(
        name=data['name'],
        modules=adapter._format_modules(data),
        date=dt.now(),
        id=new_id,
    )
    d.update(**metadata())
    # Add to DB
    adapter.create(data=d)
    flash('Created new view "{}"'.format(data['name']))
    return redirect(url_for('jsondash.view', id=new_id))


@charts.route('/charts/clone/<c_id>', methods=['POST'])

def clone(c_id):
    """Clone a json view config from the DB."""
    if not auth(authtype='clone'):
        flash('You do not have access to clone dashboards.', 'error')
        return redirect(url_for('jsondash.dashboard'))
    viewjson = adapter.read(c_id=c_id)
    if not viewjson:
        flash('Could not find view: {}'.format(id), 'error')
        return redirect(url_for('jsondash.dashboard'))
    # Update some fields.
    data = dict(
        name='Clone of {}'.format(viewjson['name']),
        modules=viewjson['modules'],
        date=dt.now(),
        id=str(uuid.uuid1()),
    )
    data.update(**metadata())
    # Add to DB
    adapter.create(data=data)
    return redirect(url_for('jsondash.view', id=data['id']))


1		# -- coding: utf-8 --
2
3		"""
4		flask_jsondash.charts_builder
5		~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
6
7		The chart blueprint that houses all functionality.
8		"""
9
10		import json
11		import os
12		import uuid
13		from collections import namedtuple
14		from datetime import datetime as dt
15
16		from flask_jsondash import static, templates
17
18		from flask import (
19		Blueprint,
20		current_app,
21		flash,
22		redirect,
23		render_template,
24		request,
25		send_from_directory,
26		url_for,
27		)
28		import jinja2
29
30		import db_adapters as adapter
31		from settings import (
32		CHARTS_CONFIG,
33		)
34
35		template_dir = os.path.dirname(templates.__file__)
36		static_dir = os.path.dirname(static.__file__)
37
38		Paginator = namedtuple('Paginator',
39		'count per_page curr_page num_pages limit skip')
40
41		charts = Blueprint(
42		'jsondash',
43		__name__,
44		template_folder=template_dir,
45		static_url_path=static_dir,
46		static_folder=static_dir,
47		)
48		default_config = dict(
49		JSONDASH_FILTERUSERS=False,
50		JSONDASH_GLOBALDASH=False,
51		JSONDASH_GLOBAL_USER='global',
52		JSONDASH_PERPAGE=25,
53		)
54
55
56		def auth(**kwargs):
57		"""Check if general auth functions have been specified.
58
59		Checks for either a global auth (if authtype is None), or
60		an action specific auth (specified by authtype).
61		"""
62		authtype = kwargs.pop('authtype')
63		if 'JSONDASH' not in current_app.config:
64		return True
65		if 'auth' not in current_app.config['JSONDASH']:
66		return True
67		auth_conf = current_app.config.get('JSONDASH').get('auth')
68		if all([authtype is not None, authtype in auth_conf]):
69		# Only perform the user-supplied check
70		# if the authtype is actually enabled.
71		return current_app.config['JSONDASH']['auth'][authtype](**kwargs)
72		return False
73
74
75		def metadata(key=None):
76		"""An abstraction around misc. metadata.
77
78		This allows loose coupling for enabling and setting
79		metadata for each chart.
80		"""
81		_metadata = dict()
82		conf = current_app.config
83		conf_metadata = conf.get('JSONDASH', {}).get('metadata', None)
84		# Also useful for getting arbitrary configuration keys.
85		if key is not None:
86		if key in conf_metadata:
87		return conf_metadata[key]()
88		else:
89		return None
90		# Update all metadata values if the function exists.
91		for k, func in conf_metadata.items():
92		_metadata[k] = conf_metadata[k]()
93		return _metadata
94
95
96		def setting(name, default=None):
97		"""A simplified getter for namespaced flask config values."""
98		if default is None:
99		default = default_config.get(name)
100		return current_app.config.get(name, default)
101
102
103		@charts.context_processor
104		def _ctx():
105		"""Inject any context needed for this blueprint."""
106		filter_user = setting('JSONDASH_FILTERUSERS')
107		return dict(
108		charts_config=CHARTS_CONFIG,
109		page_title='dashboards',
110		global_dashuser=setting('JSONDASH_GLOBAL_USER'),
111		global_dashboards=setting('JSONDASH_GLOBALDASH'),
112		username=metadata(key='username') if filter_user else None,
113		filter_dashboards=filter_user,
114		)
115
116
117		@jinja2.contextfilter
118		@charts.app_template_filter('jsonstring')
119		def jsonstring(ctx, data):
120		"""Format view json module data for template use.
121
122		It's automatically converted to unicode key/value pairs,
123		which is undesirable for the template.
124		"""
125		if 'date' in data:
126		data['date'] = str(data['date'])
127		return json.dumps(data)
128
129
130		@charts.route('/jsondash/<path:filename>')
131		def _static(filename):
132		"""Send static files directly for this blueprint."""
133		return send_from_directory(static_dir, filename)
134
135
136		def paginator(count=None):
137		"""Get pagination calculations in a compact format."""
138		if count is None:
139		count = adapter.count()
140		per_page = setting('JSONDASH_PERPAGE')
141		# Allow query parameter overrides.
142		per_page = int(request.args.get('per_page', 0)) or per_page
143		per_page = per_page if per_page > 2 else 2 # Prevent division errors etc
144		curr_page = int(request.args.get('page', 1)) - 1
145		num_pages = count // per_page
146		rem = count % per_page
147		extra_pages = 2 if rem else 1
148		pages = range(1, num_pages + extra_pages)
149		return Paginator(
150		limit=per_page,
151		per_page=per_page,
152		curr_page=curr_page,
153		skip=curr_page * per_page,
154		num_pages=pages,
155		count=count,
156		)
157
158
159		@charts.route('/charts/', methods=['GET'])
160		def dashboard():
161		"""Load all views."""
162		opts = dict()
163		views = []
164		if setting('JSONDASH_FILTERUSERS'):
165		opts.update(filter=dict(created_by=metadata(key='username')))
166		views = list(adapter.read(**opts))
167		if setting('JSONDASH_GLOBALDASH'):
168		opts.update(
169		filter=dict(created_by=setting('JSONDASH_GLOBAL_USER')))
170		views += list(adapter.read(**opts))
171		else:
172		views = list(adapter.read(**opts))
173		if views:
174		pagination = paginator(count=len(views))
175		opts.update(limit=pagination.limit, skip=pagination.skip)
176		else:
177		pagination = None
178		kwargs = dict(
179		views=views,
180		view=None,
181		paginator=pagination,
182		total_modules=sum([len(view['modules']) for view in views]),
183		)
184		return render_template('pages/charts_index.html', **kwargs)
185
186
187		@charts.route('/charts/<id>', methods=['GET'])
188		def view(id):
189		"""Load a json view config from the DB."""
190		if not auth(authtype='view'):
191		flash('You do not have access to view this dashboard.', 'error')
192		return redirect(url_for('jsondash.dashboard'))
193		viewjson = adapter.read(c_id=id)
194		if not viewjson:
195		flash('Could not find view: {}'.format(id), 'error')
196		return redirect(url_for('jsondash.dashboard'))
197		# Remove _id, it's not JSON serializeable.
198		viewjson.pop('_id')
199		# Chart family is encoded in chart type value for lookup.
200		active_charts = [v.get('family') for
201		v in viewjson['modules'] if v.get('family')]
202		kwargs = dict(id=id, view=viewjson, active_charts=active_charts)
203		return render_template('pages/chart_detail.html', **kwargs)
204
205
206		@charts.route('/charts/<c_id>/delete', methods=['POST'])
207		def delete(c_id):
208		"""Delete a json dashboard config."""
209		dash_url = url_for('jsondash.dashboard')
210		if not auth(authtype='delete'):
211		flash('You do not have access to delete dashboards.', 'error')
212		return redirect(dash_url)
213		adapter.delete(c_id)
214		flash('Deleted dashboard {}'.format(c_id))
215		return redirect(dash_url)
216
217
218		@charts.route('/charts/update', methods=['POST'])
219		def update():
220		"""Normalize the form POST and setup the json view config object."""
221		if not auth(authtype='update'):
222		flash('You do not have access to update dashboards.', 'error')
223		return redirect(url_for('jsondash.dashboard'))
224		data = request.form
225		c_id = data['id']
226		view_url = url_for('jsondash.view', id=c_id)
227		if 'edit-raw' in request.form:
228		try:
229		data = json.loads(request.form.get('config'))
230		data = adapter.reformat_data(data, c_id)
231		# Update db
232		adapter.update(c_id, data=data, fmt_modules=False)
233		except (TypeError, ValueError):
234		flash('Invalid JSON config.', 'error')
235		return redirect(view_url)
236		else:
237		# Update db
238		d = dict(
239		name=data['name'],
240		modules=adapter._format_modules(data),
241		date=dt.now(),
242		id=data['id'],
243		)
244		d.update(**metadata())
245		adapter.update(c_id, data=d)
246		flash('Updated view "{}"'.format(c_id))
247		return redirect(view_url)
248
249
250	View Code Duplication	@charts.route('/charts/create', methods=['POST'])
		0 ignored issues – show Duplication introduced 2016-10-01 05:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
251		def create():
252		"""Normalize the form POST and setup the json view config object."""
253		if not auth(authtype='create'):
254		flash('You do not have access to create dashboards.', 'error')
255		return redirect(url_for('jsondash.dashboard'))
256		data = request.form
257		new_id = str(uuid.uuid1())
258		d = dict(
259		name=data['name'],
260		modules=adapter._format_modules(data),
261		date=dt.now(),
262		id=new_id,
263		)
264		d.update(**metadata())
265		# Add to DB
266		adapter.create(data=d)
267		flash('Created new view "{}"'.format(data['name']))
268		return redirect(url_for('jsondash.view', id=new_id))
269
270
271	View Code Duplication	@charts.route('/charts/clone/<c_id>', methods=['POST'])
		0 ignored issues – show Duplication introduced 2016-10-19 18:24 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
272		def clone(c_id):
273		"""Clone a json view config from the DB."""
274		if not auth(authtype='clone'):
275		flash('You do not have access to clone dashboards.', 'error')
276		return redirect(url_for('jsondash.dashboard'))
277		viewjson = adapter.read(c_id=c_id)
278		if not viewjson:
279		flash('Could not find view: {}'.format(id), 'error')
280		return redirect(url_for('jsondash.dashboard'))
281		# Update some fields.
282		data = dict(
283		name='Clone of {}'.format(viewjson['name']),
284		modules=viewjson['modules'],
285		date=dt.now(),
286		id=str(uuid.uuid1()),
287		)
288		data.update(**metadata())
289		# Add to DB
290		adapter.create(data=data)
291		return redirect(url_for('jsondash.view', id=data['id']))
292

christabor / flask_jsondash

Push — master ( 2193e1...87074b )

auth() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like