browse() - Code Metrics - Inspection of "implement browse endpoint cache" - ergoithz/browsepy - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — 5.2-unstable ( ee49a9...946858 )

by Felipe A.

created 2016-12-07 14:50 UTC

browse() C

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	1

Metric	Value
cc	7
dl	0
loc	33
rs	5.5
c	3
b	0
f	1

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

import logging
import os
import os.path
import json
import base64
import io

from flask import Flask, Response, request, render_template, redirect, \
                  url_for, send_from_directory, stream_with_context, \
                  make_response, current_app
from werkzeug.exceptions import NotFound

from .__meta__ import __app__, __version__, __license__, __author__  # noqa
from .file import Node, OutsideRemovableBase, OutsideDirectoryBase, \
                  secure_filename
from . import event
from . import compat
from . import manager

__basedir__ = os.path.abspath(os.path.dirname(compat.fsdecode(__file__)))

logger = logging.getLogger(__name__)

app = Flask(
    __name__,
    static_url_path='/static',
    static_folder=os.path.join(__basedir__, 'static'),
    template_folder=os.path.join(__basedir__, 'templates')
    )
app.config.update(
    directory_base=compat.getcwd(),
    directory_start=compat.getcwd(),
    directory_remove=None,
    directory_upload=None,
    directory_tar_buffsize=262144,
    directory_downloadable=True,
    use_binary_multiples=True,
    plugin_modules=[],
    plugin_namespaces=(
        'browsepy.plugin',
        'browsepy_',
        '',
        ),
    cache_enable=True,
    cache_class='browsepy.cache:SimpleLRUCache',
    cache_kwargs={},
    cache_browse_key='view/browse<{sort}>/{path}',
    browse_sort_properties=['text', 'type', 'modified', 'size']
    )
app.jinja_env.add_extension('browsepy.extensions.HTMLCompress')

if "BROWSEPY_SETTINGS" in os.environ:
    app.config.from_envvar("BROWSEPY_SETTINGS")

plugin_manager = manager.PluginManager(app)


def iter_cookie_browse_sorting():
    '''
    Get sorting-cookie data of current request.

    :yields: tuple of path and sorting property
    :ytype: 2-tuple of strings
    '''
    try:
        data = request.cookies.get('browse-sorting', 'e30=').encode('ascii')
        valid = current_app.config.get('browse_sort_properties', ())
        for path, prop in json.loads(base64.b64decode(data).decode('utf-8')):
            if prop.startswith('-'):
                if prop[1:] in valid:
                    yield path, prop
            elif prop in valid:
                yield path, prop
    except (ValueError, TypeError, KeyError) as e:
        logger.exception(e)


def get_cookie_browse_sorting(path, default):
    '''
    Get sorting-cookie data for path of current request.

    :returns: sorting property
    :rtype: string
    '''
    for cpath, cprop in iter_cookie_browse_sorting():
        if path == cpath:
            return cprop
    return default


def browse_sortkey_reverse(prop):
    '''
    Get sorting function for browse

    :returns: tuple with sorting gunction and reverse bool
    :rtype: tuple of a dict and a bool
    '''
    if prop.startswith('-'):
        prop = prop[1:]
        reverse = True
    else:
        reverse = False

    if prop == 'text':
        return (
            lambda x: (
                x.is_directory == reverse,
                x.link.text.lower() if x.link and x.link.text else x.name
                ),
            reverse
            )
    if prop == 'size':
        return (
            lambda x: (
                x.is_directory == reverse,
                x.stats.st_size
                ),
            reverse
            )
    return (
        lambda x: (
            x.is_directory == reverse,
            getattr(x, prop, None)
            ),
        reverse
        )


def template_generator(template_name, **context):
    '''
    Template streaming response.

    :param template_name: template
    :param **context: parameters for templates.
    :yields: HTML strings
    '''
    app.update_template_context(context)
    template = app.jinja_env.get_template(template_name)
    return template.generate(context)


def caching_template_generator(cache_key, template_name, **context):
    '''
    Caching template streaming response.

    :param template_name: template
    :param **context: parameters for templates.
    :yields: HTML strings
    '''
    cache = current_app.extensions['plugin_manager'].cache
    buffer = io.TextIOBase()
    app.update_template_context(context)
    template = app.jinja_env.get_template(template_name)
    for chunk in template.generate(context):
        yield chunk
        buffer.write(chunk)
    cache[cache_key] = buffer.read(), request.url_root


def stream_template(template_name, **context):
    '''
    Some templates can be huge, this function returns an streaming response,
    sending the content in chunks and preventing from timeout.

    :param template_name: template
    :param **context: parameters for templates.
    :yields: HTML strings
    '''
    cache_key = context.pop('cache_key', None)
    if cache_key:
        stream = caching_template_generator(
            cache_key, template_name, **context)
    else:
        stream = template_generator(template_name, **context)
    return Response(stream_with_context(stream))


@app.context_processor
def template_globals():
    return {
        'manager': app.extensions['plugin_manager'],
        'len': len,
        }


@app.route('/sort/<string:property>', defaults={"path": ""})
@app.route('/sort/<string:property>/<path:path>')
def sort(property, path):
    try:
        directory = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()

    if not directory.is_directory:
        return NotFound()

    data = [
        (cpath, cprop)
        for cpath, cprop in iter_cookie_browse_sorting()
        if cpath != path
        ]
    data.append((path, property))
    raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))

    # prevent cookie becoming too large
    while len(raw_data) > 3975:  # 4000 - len('browse-sorting=""; Path=/')
        data.pop(0)
        raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))

    response = redirect(url_for(".browse", path=directory.urlpath))
    response.set_cookie('browse-sorting', raw_data)
    return response


@app.route("/browse", defaults={"path": ""})
@app.route('/browse/<path:path>')
def browse(path):
    sort_property = get_cookie_browse_sorting(path, 'text')

    cache_manager = current_app.extensions.get('plugin_manager')
    cache_key = current_app.config.get('cache_browse_key', '').format(
        sort=sort_property,
        path=path
        )
    if cache_manager:
        data, url_root = cache_manager.cache.get(cache_key) or (None, None)
        if data and url_root == request.url_root:
            return Response(data)
        elif not cache_manager.has_event_source(event.WatchdogEventSource):
            cache_key = None

    sort_fnc, sort_reverse = browse_sortkey_reverse(sort_property)

    try:
        directory = Node.from_urlpath(path)
        if directory.is_directory:
            return stream_template(
                'browse.html',
                cache_key=cache_key,
                file=directory,
                sort_property=sort_property,
                sort_fnc=sort_fnc,
                sort_reverse=sort_reverse
                )
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route('/open/<path:path>', endpoint="open")
def open_file(path):
    try:
        file = Node.from_urlpath(path)
        if file.is_file:
            return send_from_directory(file.parent.path, file.name)
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/download/file/<path:path>")
def download_file(path):
    try:
        file = Node.from_urlpath(path)
        if file.is_file:
            return file.download()
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/download/directory/<path:path>.tgz")
def download_directory(path):
    try:
        directory = Node.from_urlpath(path)
        if directory.is_directory:
            return directory.download()
    except OutsideDirectoryBase:
        pass
    return NotFound()


@app.route("/remove/<path:path>", methods=("GET", "POST"))
def remove(path):
    try:
        file = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()
    if request.method == 'GET':
        if not file.can_remove:
            return NotFound()
        return render_template('remove.html', file=file)
    parent = file.parent
    if parent is None:
        # base is not removable
        return NotFound()

    try:
        file.remove()
    except OutsideRemovableBase:
        return NotFound()

    return redirect(url_for(".browse", path=parent.urlpath))


@app.route("/upload", defaults={'path': ''}, methods=("POST",))
@app.route("/upload/<path:path>", methods=("POST",))
def upload(path):
    try:
        directory = Node.from_urlpath(path)
    except OutsideDirectoryBase:
        return NotFound()

    if not directory.is_directory or not directory.can_upload:
        return NotFound()

    for v in request.files.listvalues():
        for f in v:
            filename = secure_filename(f.filename)
            if filename:
                filename = directory.choose_filename(filename)
                filepath = os.path.join(directory.path, filename)
                f.save(filepath)
    return redirect(url_for(".browse", path=directory.urlpath))


@app.route("/")
def index():
    path = app.config["directory_start"] or app.config["directory_base"]
    try:
        urlpath = Node(path).urlpath
    except OutsideDirectoryBase:
        return NotFound()
    return browse(urlpath)


@app.after_request
def page_not_found(response):
    if response.status_code == 404:
        return make_response((render_template('404.html'), 404))
    return response


@app.errorhandler(404)
def page_not_found_error(e):
    return render_template('404.html'), 404


@app.errorhandler(500)
def internal_server_error(e):  # pragma: no cover
    logger.exception(e)
    return getattr(e, 'message', 'Internal server error'), 500


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			import logging
5			import os
6			import os.path
7			import json
8			import base64
9			import io
10
11			from flask import Flask, Response, request, render_template, redirect, \
12			url_for, send_from_directory, stream_with_context, \
13			make_response, current_app
14			from werkzeug.exceptions import NotFound
15
16			from .__meta__ import __app__, __version__, __license__, __author__ # noqa
17			from .file import Node, OutsideRemovableBase, OutsideDirectoryBase, \
18			secure_filename
19			from . import event
20			from . import compat
21			from . import manager
22
23			__basedir__ = os.path.abspath(os.path.dirname(compat.fsdecode(__file__)))
24
25			logger = logging.getLogger(__name__)
26
27			app = Flask(
28			__name__,
29			static_url_path='/static',
30			static_folder=os.path.join(__basedir__, 'static'),
31			template_folder=os.path.join(__basedir__, 'templates')
32			)
33			app.config.update(
34			directory_base=compat.getcwd(),
35			directory_start=compat.getcwd(),
36			directory_remove=None,
37			directory_upload=None,
38			directory_tar_buffsize=262144,
39			directory_downloadable=True,
40			use_binary_multiples=True,
41			plugin_modules=[],
42			plugin_namespaces=(
43			'browsepy.plugin',
44			'browsepy_',
45			'',
46			),
47			cache_enable=True,
48			cache_class='browsepy.cache:SimpleLRUCache',
49			cache_kwargs={},
50			cache_browse_key='view/browse<{sort}>/{path}',
51			browse_sort_properties=['text', 'type', 'modified', 'size']
52			)
53			app.jinja_env.add_extension('browsepy.extensions.HTMLCompress')
54
55			if "BROWSEPY_SETTINGS" in os.environ:
56			app.config.from_envvar("BROWSEPY_SETTINGS")
57
58			plugin_manager = manager.PluginManager(app)
59
60
61			def iter_cookie_browse_sorting():
62			'''
63			Get sorting-cookie data of current request.
64
65			:yields: tuple of path and sorting property
66			:ytype: 2-tuple of strings
67			'''
68			try:
69			data = request.cookies.get('browse-sorting', 'e30=').encode('ascii')
70			valid = current_app.config.get('browse_sort_properties', ())
71			for path, prop in json.loads(base64.b64decode(data).decode('utf-8')):
72			if prop.startswith('-'):
73			if prop[1:] in valid:
74			yield path, prop
75			elif prop in valid:
76			yield path, prop
77			except (ValueError, TypeError, KeyError) as e:
78			logger.exception(e)
79
80
81			def get_cookie_browse_sorting(path, default):
82			'''
83			Get sorting-cookie data for path of current request.
84
85			:returns: sorting property
86			:rtype: string
87			'''
88			for cpath, cprop in iter_cookie_browse_sorting():
89			if path == cpath:
90			return cprop
91			return default
92
93
94			def browse_sortkey_reverse(prop):
95			'''
96			Get sorting function for browse
97
98			:returns: tuple with sorting gunction and reverse bool
99			:rtype: tuple of a dict and a bool
100			'''
101			if prop.startswith('-'):
102			prop = prop[1:]
103			reverse = True
104			else:
105			reverse = False
106
107			if prop == 'text':
108			return (
109			lambda x: (
110			x.is_directory == reverse,
111			x.link.text.lower() if x.link and x.link.text else x.name
112			),
113			reverse
114			)
115			if prop == 'size':
116			return (
117			lambda x: (
118			x.is_directory == reverse,
119			x.stats.st_size
120			),
121			reverse
122			)
123			return (
124			lambda x: (
125			x.is_directory == reverse,
126			getattr(x, prop, None)
127			),
128			reverse
129			)
130
131
132			def template_generator(template_name, **context):
133			'''
134			Template streaming response.
135
136			:param template_name: template
137			:param **context: parameters for templates.
138			:yields: HTML strings
139			'''
140			app.update_template_context(context)
141			template = app.jinja_env.get_template(template_name)
142			return template.generate(context)
143
144
145			def caching_template_generator(cache_key, template_name, **context):
146			'''
147			Caching template streaming response.
148
149			:param template_name: template
150			:param **context: parameters for templates.
151			:yields: HTML strings
152			'''
153			cache = current_app.extensions['plugin_manager'].cache
154			buffer = io.TextIOBase()
155			app.update_template_context(context)
156			template = app.jinja_env.get_template(template_name)
157			for chunk in template.generate(context):
158			yield chunk
159			buffer.write(chunk)
160			cache[cache_key] = buffer.read(), request.url_root
161
162
163			def stream_template(template_name, **context):
164			'''
165			Some templates can be huge, this function returns an streaming response,
166			sending the content in chunks and preventing from timeout.
167
168			:param template_name: template
169			:param **context: parameters for templates.
170			:yields: HTML strings
171			'''
172			cache_key = context.pop('cache_key', None)
173			if cache_key:
174			stream = caching_template_generator(
175			cache_key, template_name, **context)
176			else:
177			stream = template_generator(template_name, **context)
178			return Response(stream_with_context(stream))
179
180
181			@app.context_processor
182			def template_globals():
183			return {
184			'manager': app.extensions['plugin_manager'],
185			'len': len,
186			}
187
188
189			@app.route('/sort/<string:property>', defaults={"path": ""})
190			@app.route('/sort/<string:property>/<path:path>')
191			def sort(property, path):
192			try:
193			directory = Node.from_urlpath(path)
194			except OutsideDirectoryBase:
195			return NotFound()
196
197			if not directory.is_directory:
198			return NotFound()
199
200			data = [
201			(cpath, cprop)
202			for cpath, cprop in iter_cookie_browse_sorting()
203			if cpath != path
204			]
205			data.append((path, property))
206			raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))
207
208			# prevent cookie becoming too large
209			while len(raw_data) > 3975: # 4000 - len('browse-sorting=""; Path=/')
210			data.pop(0)
211			raw_data = base64.b64encode(json.dumps(data).encode('utf-8'))
212
213			response = redirect(url_for(".browse", path=directory.urlpath))
214			response.set_cookie('browse-sorting', raw_data)
215			return response
216
217
218			@app.route("/browse", defaults={"path": ""})
219			@app.route('/browse/<path:path>')
220			def browse(path):
221			sort_property = get_cookie_browse_sorting(path, 'text')
222
223			cache_manager = current_app.extensions.get('plugin_manager')
224			cache_key = current_app.config.get('cache_browse_key', '').format(
225			sort=sort_property,
226			path=path
227			)
228			if cache_manager:
229			data, url_root = cache_manager.cache.get(cache_key) or (None, None)
230			if data and url_root == request.url_root:
231			return Response(data)
232			elif not cache_manager.has_event_source(event.WatchdogEventSource):
233			cache_key = None
234
235			sort_fnc, sort_reverse = browse_sortkey_reverse(sort_property)
236
237			try:
238			directory = Node.from_urlpath(path)
239			if directory.is_directory:
240			return stream_template(
241			'browse.html',
242			cache_key=cache_key,
243			file=directory,
244			sort_property=sort_property,
245			sort_fnc=sort_fnc,
246			sort_reverse=sort_reverse
247			)
248			except OutsideDirectoryBase:
249			pass
250			return NotFound()
251
252
253			@app.route('/open/<path:path>', endpoint="open")
254			def open_file(path):
255			try:
256			file = Node.from_urlpath(path)
257			if file.is_file:
258			return send_from_directory(file.parent.path, file.name)
259			except OutsideDirectoryBase:
260			pass
261			return NotFound()
262
263
264			@app.route("/download/file/<path:path>")
265			def download_file(path):
266			try:
267			file = Node.from_urlpath(path)
268			if file.is_file:
269			return file.download()
270			except OutsideDirectoryBase:
271			pass
272			return NotFound()
273
274
275			@app.route("/download/directory/<path:path>.tgz")
276			def download_directory(path):
277			try:
278			directory = Node.from_urlpath(path)
279			if directory.is_directory:
280			return directory.download()
281			except OutsideDirectoryBase:
282			pass
283			return NotFound()
284
285
286			@app.route("/remove/<path:path>", methods=("GET", "POST"))
287			def remove(path):
288			try:
289			file = Node.from_urlpath(path)
290			except OutsideDirectoryBase:
291			return NotFound()
292			if request.method == 'GET':
293			if not file.can_remove:
294			return NotFound()
295			return render_template('remove.html', file=file)
296			parent = file.parent
297			if parent is None:
298			# base is not removable
299			return NotFound()
300
301			try:
302			file.remove()
303			except OutsideRemovableBase:
304			return NotFound()
305
306			return redirect(url_for(".browse", path=parent.urlpath))
307
308
309			@app.route("/upload", defaults={'path': ''}, methods=("POST",))
310			@app.route("/upload/<path:path>", methods=("POST",))
311			def upload(path):
312			try:
313			directory = Node.from_urlpath(path)
314			except OutsideDirectoryBase:
315			return NotFound()
316
317			if not directory.is_directory or not directory.can_upload:
318			return NotFound()
319
320			for v in request.files.listvalues():
321			for f in v:
322			filename = secure_filename(f.filename)
323			if filename:
324			filename = directory.choose_filename(filename)
325			filepath = os.path.join(directory.path, filename)
326			f.save(filepath)
327			return redirect(url_for(".browse", path=directory.urlpath))
328
329
330			@app.route("/")
331			def index():
332			path = app.config["directory_start"] or app.config["directory_base"]
333			try:
334			urlpath = Node(path).urlpath
335			except OutsideDirectoryBase:
336			return NotFound()
337			return browse(urlpath)
338
339
340			@app.after_request
341			def page_not_found(response):
342			if response.status_code == 404:
343			return make_response((render_template('404.html'), 404))
344			return response
345
346
347			@app.errorhandler(404)
348			def page_not_found_error(e):
349			return render_template('404.html'), 404
350
351
352			@app.errorhandler(500)
353			def internal_server_error(e): # pragma: no cover
354			logger.exception(e)
355			return getattr(e, 'message', 'Internal server error'), 500
356

ergoithz / browsepy

Push — 5.2-unstable ( ee49a9...946858 )

browse() C

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like