flask_imagine.Imagine._handle_adapter() - Code Metrics - Inspection of "Filesystem adapter tests. Logging" - FlaskGuys/Flask-Imagine - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 9f3e8d...05150b )

by Dmitry

created 2016-04-12 22:30 UTC

flask_imagine.Imagine._handle_adapter() A

↳ Parent: Imagine

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	4
dl	0
loc	14
rs	9.2

"""
Flask Imagine extension.
"""
import logging

from flask import current_app, abort, redirect

from .adapters import ImagineFilesystemAdapter
from .filters import AutorotateFilter, ThumbnailFilter
from .filters.interface import ImagineFilterInterface
from .helpers.regex_route import RegexConverter

__all__ = ['Imagine', 'imagine_filter']

LOGGER = logging.getLogger(__file__)


class Imagine(object):
    """
    Flask Imagine extension
    """
    filter_sets = {}
    adapter = None

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

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

        self._set_defaults(app)

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

        self._add_url_rule(app)
        self._add_template_filter(app)

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

        app.config.setdefault('IMAGINE_ADAPTERS', {
            'fs': ImagineFilesystemAdapter,
        })
        app.config.setdefault('IMAGINE_FILTERS', {
            'autorotate': AutorotateFilter,
            'thumbnail': ThumbnailFilter
        })

        app.config.setdefault('IMAGINE_ADAPTER', {
            'name': 'fs',
            'source_folder': '/static/',
            'cache_folder': '/cache/'
        })

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

        return app

    def _handle_adapter(self, app):
        """
        Handle storage adapter configuration
        :param app: Flask application
        :return:
        """
        if 'IMAGINE_ADAPTER' in app.config \
                and 'name' in app.config['IMAGINE_ADAPTER'] \
                and app.config['IMAGINE_ADAPTER']['name'] in app.config['IMAGINE_ADAPTERS'].keys():
            self.adapter = app.config['IMAGINE_ADAPTERS'][app.config['IMAGINE_ADAPTER']['name']](
                **app.config['IMAGINE_ADAPTER']
            )
        else:
            raise ValueError('Unknown adapter type')

    def _handle_filter_sets(self, app):
        """
        Handle filter sets
        :param app: Flask application
        :return:
        """
        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'].iteritems():
                filter_set = []
                if isinstance(filters_settings, dict) and 'filters' in filters_settings:
                    for filter_type, filter_settings in filters_settings['filters'].iteritems():
                        if filter_type in app.config['IMAGINE_FILTERS']:
                            filter_item = app.config['IMAGINE_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 and filters_settings['cached']:
                        filter_config['cached'] = True
                    else:
                        filter_config['cached'] = False

                    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:
        """
        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

    @classmethod
    def _add_template_filter(cls, app):
        """
        Add template filter
        :param app: Flask application
        :return:
        """
        if hasattr(app, 'add_template_filter'):
            app.add_template_filter(imagine_filter, 'imagine_filter')
        else:
            ctx = {
                'imagine_filter': imagine_filter
            }
            app.context_processor(lambda: ctx)

        return app

    def handle_request(self, filter_name, path):
        """
        Handle image request
        :param filter_name: filter_name
        :param path: image_path
        :param kwargs:
        :return:
        """
        if filter_name in self.filter_sets:
            if self.filter_sets[filter_name]['cached']:
                if self.adapter.check_cached_item('%s/%s' % (filter_name, path)):
                    return redirect(
                        '%s/%s/%s' % (
                            self.adapter.source_folder,
                            self.adapter.cache_folder,
                            '%s/%s' % (filter_name, path)
                        )
                    )

            resource = self.adapter.get_item(path)

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

                return redirect(self.adapter.create_cached_item('%s/%s' % (filter_name, path), resource))
            else:
                LOGGER.warning('File "%s" not found.' % path)
                abort(404)
        else:
            LOGGER.warning('Filter "%s" not found.' % filter_name)
            abort(404)


def imagine_filter(path, filter_name):
    """
    Template filter
    :param path: image path
    :param filter_name: filter_name
    :param kwargs:
    :return:
    """
    self = current_app.extensions['imagine']
    return self.build_url(path, filter_name)


1			"""
2			Flask Imagine extension.
3			"""
4			import logging
5
6			from flask import current_app, abort, redirect
7
8			from .adapters import ImagineFilesystemAdapter
9			from .filters import AutorotateFilter, ThumbnailFilter
10			from .filters.interface import ImagineFilterInterface
11			from .helpers.regex_route import RegexConverter
12
13			__all__ = ['Imagine', 'imagine_filter']
14
15			LOGGER = logging.getLogger(__file__)
16
17
18			class Imagine(object):
19			"""
20			Flask Imagine extension
21			"""
22			filter_sets = {}
23			adapter = None
24
25			def __init__(self, app=None):
26			"""
27			:param app: Flask application
28			:return:
29			"""
30			if app is not None:
31			self.init_app(app)
32
33			def init_app(self, app):
34			"""
35			:param app: Flask application
36			:return:
37			"""
38			if not hasattr(app, 'extensions'):
39			app.extensions = {}
40			app.extensions['imagine'] = self
41
42			self._set_defaults(app)
43
44			self._handle_adapter(app)
45			self._handle_filter_sets(app)
46
47			self._add_url_rule(app)
48			self._add_template_filter(app)
49
50			@classmethod
51			def _set_defaults(cls, app):
52			"""
53			Set default configuration parameters
54			:param app: Flask application
55			:return:
56			"""
57			app.config.setdefault('IMAGINE_URL', '/media/cache/resolve')
58			app.config.setdefault('IMAGINE_NAME', 'imagine')
59			app.config.setdefault('IMAGINE_THUMBS_PATH', 'cache/')
60			app.config.setdefault('IMAGINE_CACHE_ENABLED', True)
61
62			app.config.setdefault('IMAGINE_ADAPTERS', {
63			'fs': ImagineFilesystemAdapter,
64			})
65			app.config.setdefault('IMAGINE_FILTERS', {
66			'autorotate': AutorotateFilter,
67			'thumbnail': ThumbnailFilter
68			})
69
70			app.config.setdefault('IMAGINE_ADAPTER', {
71			'name': 'fs',
72			'source_folder': '/static/',
73			'cache_folder': '/cache/'
74			})
75
76			app.config.setdefault('IMAGINE_FILTER_SETS', {})
77
78			return app
79
80			def _handle_adapter(self, app):
81			"""
82			Handle storage adapter configuration
83			:param app: Flask application
84			:return:
85			"""
86			if 'IMAGINE_ADAPTER' in app.config \
87			and 'name' in app.config['IMAGINE_ADAPTER'] \
88			and app.config['IMAGINE_ADAPTER']['name'] in app.config['IMAGINE_ADAPTERS'].keys():
89			self.adapter = app.config['IMAGINE_ADAPTERS'][app.config['IMAGINE_ADAPTER']['name']](
90			**app.config['IMAGINE_ADAPTER']
91			)
92			else:
93			raise ValueError('Unknown adapter type')
94
95			def _handle_filter_sets(self, app):
96			"""
97			Handle filter sets
98			:param app: Flask application
99			:return:
100			"""
101			if 'IMAGINE_FILTER_SETS' in app.config and isinstance(app.config['IMAGINE_FILTER_SETS'], dict):
102			for filter_name, filters_settings in app.config['IMAGINE_FILTER_SETS'].iteritems():
103			filter_set = []
104			if isinstance(filters_settings, dict) and 'filters' in filters_settings:
105			for filter_type, filter_settings in filters_settings['filters'].iteritems():
106			if filter_type in app.config['IMAGINE_FILTERS']:
107			filter_item = app.config['IMAGINE_FILTERS'][filter_type](**filter_settings)
108			if isinstance(filter_item, ImagineFilterInterface):
109			filter_set.append(filter_item)
110			else:
111			raise ValueError('Filter must be implement ImagineFilterInterface')
112			else:
113			raise ValueError('Unknown filter type: %s' % filter_type)
114
115			filter_config = {'filters': filter_set}
116			if 'cached' in filters_settings and filters_settings['cached']:
117			filter_config['cached'] = True
118			else:
119			filter_config['cached'] = False
120
121			self.filter_sets.update({filter_name: filter_config})
122			else:
123			raise ValueError('Wrong settings for filter: %s' % filter_name)
124			else:
125			raise ValueError('Filters configuration does not present')
126
127			def _add_url_rule(self, app):
128			"""
129			Add url rule for get filtered images
130			:param app: Flask application
131			:return:
132			"""
133			app.url_map.converters['regex'] = RegexConverter
134			app.add_url_rule(
135			app.config['IMAGINE_URL'] + '/<regex("[^\/]+"):filter_name>/<path:path>',
136			app.config['IMAGINE_NAME'],
137			self.handle_request
138			)
139
140			return app
141
142			@classmethod
143			def _add_template_filter(cls, app):
144			"""
145			Add template filter
146			:param app: Flask application
147			:return:
148			"""
149			if hasattr(app, 'add_template_filter'):
150			app.add_template_filter(imagine_filter, 'imagine_filter')
151			else:
152			ctx = {
153			'imagine_filter': imagine_filter
154			}
155			app.context_processor(lambda: ctx)
156
157			return app
158
159			def handle_request(self, filter_name, path):
160			"""
161			Handle image request
162			:param filter_name: filter_name
163			:param path: image_path
164			:param kwargs:
165			:return:
166			"""
167			if filter_name in self.filter_sets:
168			if self.filter_sets[filter_name]['cached']:
169			if self.adapter.check_cached_item('%s/%s' % (filter_name, path)):
170			return redirect(
171			'%s/%s/%s' % (
172			self.adapter.source_folder,
173			self.adapter.cache_folder,
174			'%s/%s' % (filter_name, path)
175			)
176			)
177
178			resource = self.adapter.get_item(path)
179
180			if resource:
181			for filter_item in self.filter_sets[filter_name]['filters']:
182			resource = filter_item.apply(resource)
183
184			return redirect(self.adapter.create_cached_item('%s/%s' % (filter_name, path), resource))
185			else:
186			LOGGER.warning('File "%s" not found.' % path)
187			abort(404)
188			else:
189			LOGGER.warning('Filter "%s" not found.' % filter_name)
190			abort(404)
191
192
193			def imagine_filter(path, filter_name):
194			"""
195			Template filter
196			:param path: image path
197			:param filter_name: filter_name
198			:param kwargs:
199			:return:
200			"""
201			self = current_app.extensions['imagine']
202			return self.build_url(path, filter_name)
203

FlaskGuys / Flask-Imagine

GitHub Access Token became invalid

Push — master ( 9f3e8d...05150b )

flask_imagine.Imagine._handle_adapter() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like