Imagine._add_url_rule() - Code Metrics - Inspection of "Python 3 support. Directly configuration for filte..." - FlaskGuys/Flask-Imagine - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( cc1653...dedf0b )

by Dmitry

created 2016-04-25 22:14 UTC

Imagine._add_url_rule() A

↳ Parent: Imagine

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	1
c	2
b	0
f	0
dl	0
loc	14
rs	9.4285

"""
Flask Imagine extension.
"""
from __future__ import unicode_literals
import logging

from io import BytesIO
from flask import current_app, abort, redirect

from flask.ext.imagine.adapters import *
from flask.ext.imagine.filters import *

from .helpers.regex_route import RegexConverter

LOGGER = logging.getLogger(__file__)


class Imagine(object):
    """
    Flask Imagine extension
    """
    _adapters = {
        'fs': ImagineFilesystemAdapter
    }
    _filters = {
        'autorotate': AutorotateFilter,
        'crop': CropFilter,
        'downscale': DownscaleFilter,
        'relative_resize': RelativeResizeFilter,
        'rotate': RotateFilter,
        'thumbnail': ThumbnailFilter,
        'upscale': UpscaleFilter,
        'watermark': WatermarkFilter
    }

    _filter_sets = {}
    _adapter = None
    _redirect_code = 302

    def __init__(self, app=None):
        """
        :param app: Flask application
        """
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        """
        :param app: Flask application
        """
        if not hasattr(app, 'extensions'):  # pragma: no cover
            app.extensions = {}
        app.extensions['imagine'] = self

        self._set_defaults(app)

        self._redirect_code = app.config['IMAGINE_CACHE_REDIRECT_CODE']

        if isinstance(app.config['IMAGINE_ADAPTERS'], dict):
            self._adapters.update(app.config['IMAGINE_ADAPTERS'])
        if isinstance(app.config['IMAGINE_FILTERS'], dict):
            self._filters.update(app.config['IMAGINE_FILTERS'])

        self._handle_adapter(app)
        self._handle_filter_sets(app)

        self._add_url_rule(app)

    @classmethod
    def _set_defaults(cls, app):
        """
        Set default configuration parameters
        :param app: Flask application
        :return: Flask application
        """
        app.config.setdefault('IMAGINE_URL', '/media/cache/resolve')
        app.config.setdefault('IMAGINE_NAME', 'imagine')
        app.config.setdefault('IMAGINE_CACHE_ENABLED', True)
        app.config.setdefault('IMAGINE_CACHE_REDIRECT_CODE', 302)

        app.config.setdefault('IMAGINE_ADAPTERS', {})
        app.config.setdefault('IMAGINE_FILTERS', {})

        app.config.setdefault('IMAGINE_ADAPTER', {
            'name': 'fs',
        })

        app.config.setdefault('IMAGINE_FILTER_SETS', {})

        return app

    def _handle_adapter(self, app):
        """
        Handle storage _adapter configuration
        :param app: Flask application
        """
        if 'IMAGINE_ADAPTER' in app.config \
                and 'name' in app.config['IMAGINE_ADAPTER'] \
                and app.config['IMAGINE_ADAPTER']['name'] in self._adapters.keys():
            self._adapter = self._adapters[app.config['IMAGINE_ADAPTER']['name']](
                **app.config['IMAGINE_ADAPTER']
            )
        else:
            raise ValueError('Unknown _adapter: %s' % str(app.config['IMAGINE_ADAPTER']))

    def _handle_filter_sets(self, app):
        """
        Handle filter sets
        :param app: Flask application
        """
        if 'IMAGINE_FILTER_SETS' in app.config and isinstance(app.config['IMAGINE_FILTER_SETS'], dict):
            for filter_name, filters_settings in app.config['IMAGINE_FILTER_SETS'].items():
                filter_set = []
                if isinstance(filters_settings, dict) and '_filters' in filters_settings:
                    for filter_type, filter_settings in filters_settings['_filters'].items():
                        if filter_type in self._filters:
                            filter_item = self._filters[filter_type](**filter_settings)
                            if isinstance(filter_item, ImagineFilterInterface):
                                filter_set.append(filter_item)
                            else:
                                raise ValueError('Filter must be implement ImagineFilterInterface')
                        else:
                            raise ValueError('Unknown filter type: %s' % filter_type)

                    filter_config = {'_filters': filter_set}
                    if 'cached' in filters_settings:
                        filter_config['cached'] = filters_settings['cached']
                    else:
                        filter_config['cached'] = app.config['IMAGINE_CACHE_ENABLED']

                    self._filter_sets.update({filter_name: filter_config})
                else:
                    raise ValueError('Wrong settings for filter: %s' % filter_name)
        else:
            raise ValueError('Filters configuration does not present')

    def _add_url_rule(self, app):
        """
        Add url rule for get filtered images
        :param app: Flask application
        :return: Flask application
        """
        app.url_map.converters['regex'] = RegexConverter
        app.add_url_rule(
            app.config['IMAGINE_URL'] + '/<regex("[^\/]+"):filter_name>/<path:path>',
            app.config['IMAGINE_NAME'],
            self.handle_request
        )

        return app

    def handle_request(self, filter_name, path):
        """
        Handle image request
        :param filter_name: filter_name
        :param path: image_path
        :return:
        """
        if filter_name in self._filter_sets:
            if self._filter_sets[filter_name]['cached']:
                cached_item_path = self._adapter.check_cached_item('%s/%s' % (filter_name, path))
                if cached_item_path:
                    return redirect(cached_item_path, self._redirect_code)

            resource = self._adapter.get_item(path)

            if resource:
                for filter_item in self._filter_sets[filter_name]['_filters']:
                    resource = filter_item.apply(resource)

                if self._filter_sets[filter_name]['cached']:
                    return redirect(
                        self._adapter.create_cached_item('%s/%s' % (filter_name, path), resource),
                        self._redirect_code
                    )
                else:
                    output = BytesIO()
                    resource.save(output, format=str(resource.format))
                    return output.getvalue()
            else:
                LOGGER.warning('File "%s" not found.' % path)
                abort(404)
        else:
            LOGGER.warning('Filter "%s" not found.' % filter_name)
            abort(404)

    def clear_cache(self, path, filter_name=None):
        """
        Clear cache for resource path.
        :param path: str
        :param filter_name: str or None
        """
        if filter_name:
            self._adapter.remove_cached_item('%s/%s' % (filter_name, path))
        else:
            for filter_name in self._filter_sets:
                self._adapter.remove_cached_item('%s/%s' % (filter_name, path))

    def add_filter_set(self, filter_name, filter_set, cached=True):
        """
        Manual addition of filter set
        :param filter_name: str
        :param filter_set: list
        :param cached: bool
        """
        try:
            hash(filter_name)
        except TypeError as e:

            raise ValueError('Filter set name must be as instance of hashable type: %s' % str(e))

        if not isinstance(filter_set, list):
            raise ValueError('Filters must be a list.')

        if len(filter_set) == 0:
            raise ValueError('Filters count must be greater than 0.')

        for filter_instance in filter_set:
            if not isinstance(filter_instance, ImagineFilterInterface):
                raise ValueError('All filters must implement of ImagineFilterInterface.')

        if not isinstance(cached, bool):
            raise ValueError('Cached parameter must be a bool.')

        filter_config = {
            '_filters': filter_set,
            'cached': cached
        }

        if filter_name not in self._filter_sets:
            self._filter_sets.update({filter_name: filter_config})
        else:
            raise ValueError('Duplicate filter set name.')


def imagine_cache_clear(path, filter_name=None):
    """
    Clear cache for resource path.
    :param path: str
    :param filter_name: str or None
    """
    self = current_app.extensions['imagine']
    self.clear_cache(path, filter_name)


1			"""
2			Flask Imagine extension.
3			"""
4			from __future__ import unicode_literals
5			import logging
6
7			from io import BytesIO
8			from flask import current_app, abort, redirect
9
10			from flask.ext.imagine.adapters import *
11			from flask.ext.imagine.filters import *
12
13			from .helpers.regex_route import RegexConverter
14
15			LOGGER = logging.getLogger(__file__)
16
17
18			class Imagine(object):
19			"""
20			Flask Imagine extension
21			"""
22			_adapters = {
23			'fs': ImagineFilesystemAdapter
24			}
25			_filters = {
26			'autorotate': AutorotateFilter,
27			'crop': CropFilter,
28			'downscale': DownscaleFilter,
29			'relative_resize': RelativeResizeFilter,
30			'rotate': RotateFilter,
31			'thumbnail': ThumbnailFilter,
32			'upscale': UpscaleFilter,
33			'watermark': WatermarkFilter
34			}
35
36			_filter_sets = {}
37			_adapter = None
38			_redirect_code = 302
39
40			def __init__(self, app=None):
41			"""
42			:param app: Flask application
43			"""
44			if app is not None:
45			self.init_app(app)
46
47			def init_app(self, app):
48			"""
49			:param app: Flask application
50			"""
51			if not hasattr(app, 'extensions'): # pragma: no cover
52			app.extensions = {}
53			app.extensions['imagine'] = self
54
55			self._set_defaults(app)
56
57			self._redirect_code = app.config['IMAGINE_CACHE_REDIRECT_CODE']
58
59			if isinstance(app.config['IMAGINE_ADAPTERS'], dict):
60			self._adapters.update(app.config['IMAGINE_ADAPTERS'])
61			if isinstance(app.config['IMAGINE_FILTERS'], dict):
62			self._filters.update(app.config['IMAGINE_FILTERS'])
63
64			self._handle_adapter(app)
65			self._handle_filter_sets(app)
66
67			self._add_url_rule(app)
68
69			@classmethod
70			def _set_defaults(cls, app):
71			"""
72			Set default configuration parameters
73			:param app: Flask application
74			:return: Flask application
75			"""
76			app.config.setdefault('IMAGINE_URL', '/media/cache/resolve')
77			app.config.setdefault('IMAGINE_NAME', 'imagine')
78			app.config.setdefault('IMAGINE_CACHE_ENABLED', True)
79			app.config.setdefault('IMAGINE_CACHE_REDIRECT_CODE', 302)
80
81			app.config.setdefault('IMAGINE_ADAPTERS', {})
82			app.config.setdefault('IMAGINE_FILTERS', {})
83
84			app.config.setdefault('IMAGINE_ADAPTER', {
85			'name': 'fs',
86			})
87
88			app.config.setdefault('IMAGINE_FILTER_SETS', {})
89
90			return app
91
92			def _handle_adapter(self, app):
93			"""
94			Handle storage _adapter configuration
95			:param app: Flask application
96			"""
97			if 'IMAGINE_ADAPTER' in app.config \
98			and 'name' in app.config['IMAGINE_ADAPTER'] \
99			and app.config['IMAGINE_ADAPTER']['name'] in self._adapters.keys():
100			self._adapter = self._adapters[app.config['IMAGINE_ADAPTER']['name']](
101			**app.config['IMAGINE_ADAPTER']
102			)
103			else:
104			raise ValueError('Unknown _adapter: %s' % str(app.config['IMAGINE_ADAPTER']))
105
106			def _handle_filter_sets(self, app):
107			"""
108			Handle filter sets
109			:param app: Flask application
110			"""
111			if 'IMAGINE_FILTER_SETS' in app.config and isinstance(app.config['IMAGINE_FILTER_SETS'], dict):
112			for filter_name, filters_settings in app.config['IMAGINE_FILTER_SETS'].items():
113			filter_set = []
114			if isinstance(filters_settings, dict) and '_filters' in filters_settings:
115			for filter_type, filter_settings in filters_settings['_filters'].items():
116			if filter_type in self._filters:
117			filter_item = self._filters[filter_type](**filter_settings)
118			if isinstance(filter_item, ImagineFilterInterface):
119			filter_set.append(filter_item)
120			else:
121			raise ValueError('Filter must be implement ImagineFilterInterface')
122			else:
123			raise ValueError('Unknown filter type: %s' % filter_type)
124
125			filter_config = {'_filters': filter_set}
126			if 'cached' in filters_settings:
127			filter_config['cached'] = filters_settings['cached']
128			else:
129			filter_config['cached'] = app.config['IMAGINE_CACHE_ENABLED']
130
131			self._filter_sets.update({filter_name: filter_config})
132			else:
133			raise ValueError('Wrong settings for filter: %s' % filter_name)
134			else:
135			raise ValueError('Filters configuration does not present')
136
137			def _add_url_rule(self, app):
138			"""
139			Add url rule for get filtered images
140			:param app: Flask application
141			:return: Flask application
142			"""
143			app.url_map.converters['regex'] = RegexConverter
144			app.add_url_rule(
145			app.config['IMAGINE_URL'] + '/<regex("[^\/]+"):filter_name>/<path:path>',
146			app.config['IMAGINE_NAME'],
147			self.handle_request
148			)
149
150			return app
151
152			def handle_request(self, filter_name, path):
153			"""
154			Handle image request
155			:param filter_name: filter_name
156			:param path: image_path
157			:return:
158			"""
159			if filter_name in self._filter_sets:
160			if self._filter_sets[filter_name]['cached']:
161			cached_item_path = self._adapter.check_cached_item('%s/%s' % (filter_name, path))
162			if cached_item_path:
163			return redirect(cached_item_path, self._redirect_code)
164
165			resource = self._adapter.get_item(path)
166
167			if resource:
168			for filter_item in self._filter_sets[filter_name]['_filters']:
169			resource = filter_item.apply(resource)
170
171			if self._filter_sets[filter_name]['cached']:
172			return redirect(
173			self._adapter.create_cached_item('%s/%s' % (filter_name, path), resource),
174			self._redirect_code
175			)
176			else:
177			output = BytesIO()
178			resource.save(output, format=str(resource.format))
179			return output.getvalue()
180			else:
181			LOGGER.warning('File "%s" not found.' % path)
182			abort(404)
183			else:
184			LOGGER.warning('Filter "%s" not found.' % filter_name)
185			abort(404)
186
187			def clear_cache(self, path, filter_name=None):
188			"""
189			Clear cache for resource path.
190			:param path: str
191			:param filter_name: str or None
192			"""
193			if filter_name:
194			self._adapter.remove_cached_item('%s/%s' % (filter_name, path))
195			else:
196			for filter_name in self._filter_sets:
197			self._adapter.remove_cached_item('%s/%s' % (filter_name, path))
198
199			def add_filter_set(self, filter_name, filter_set, cached=True):
200			"""
201			Manual addition of filter set
202			:param filter_name: str
203			:param filter_set: list
204			:param cached: bool
205			"""
206			try:
207			hash(filter_name)
208			except TypeError as e:
			0 ignored issues – show Coding Style Naming introduced 2016-04-25 22:15 UTC by Report Bug Copy Issue Report The name `e` does not conform to the variable naming conventions (`[a-z_][a-z0-9_]{1,30}$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
209			raise ValueError('Filter set name must be as instance of hashable type: %s' % str(e))
210
211			if not isinstance(filter_set, list):
212			raise ValueError('Filters must be a list.')
213
214			if len(filter_set) == 0:
215			raise ValueError('Filters count must be greater than 0.')
216
217			for filter_instance in filter_set:
218			if not isinstance(filter_instance, ImagineFilterInterface):
219			raise ValueError('All filters must implement of ImagineFilterInterface.')
220
221			if not isinstance(cached, bool):
222			raise ValueError('Cached parameter must be a bool.')
223
224			filter_config = {
225			'_filters': filter_set,
226			'cached': cached
227			}
228
229			if filter_name not in self._filter_sets:
230			self._filter_sets.update({filter_name: filter_config})
231			else:
232			raise ValueError('Duplicate filter set name.')
233
234
235			def imagine_cache_clear(path, filter_name=None):
236			"""
237			Clear cache for resource path.
238			:param path: str
239			:param filter_name: str or None
240			"""
241			self = current_app.extensions['imagine']
242			self.clear_cache(path, filter_name)
243

FlaskGuys / Flask-Imagine

GitHub Access Token became invalid

Push — master ( cc1653...dedf0b )

Imagine._add_url_rule() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like