Imagine.clear_cache() - Code Metrics - Inspection of "Cache clear function." - FlaskGuys/Flask-Imagine - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 598bb5...e05af6 )

by Dmitry

created 2016-04-13 22:15 UTC

Imagine.clear_cache() A

↳ Parent: Imagine

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	6
rs	9.4285

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

from flask import current_app, abort, redirect

from .adapters import ImagineFilesystemAdapter
from .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

    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)

        if isinstance(app.config['IMAGINE_ADAPTERS'], dict):
            self.filters.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_ADAPTERS', {})
        app.config.setdefault('IMAGINE_FILTERS', {})

        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
        """
        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' % unicode(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'].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 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']:
                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 clear_cache(self, path, filter_name=None):
class SomeClass:
    def some_method(self):
        """Do x and return foo."""
        if filter_name:
            self.adapter.remove_cached_item('%s/%s' % (filter_name, path))
        else:
            for filter_name in self.filter_sets.iterkeys():
                self.adapter.remove_cached_item('%s/%s' % (filter_name, path))


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			import logging
5
6			from flask import current_app, abort, redirect
7
8			from .adapters import ImagineFilesystemAdapter
9			from .filters import *
10			from .helpers.regex_route import RegexConverter
11
12			LOGGER = logging.getLogger(__file__)
13
14
15			class Imagine(object):
16			"""
17			Flask Imagine extension
18			"""
19			adapters = {
20			'fs': ImagineFilesystemAdapter
21			}
22			filters = {
23			'autorotate': AutorotateFilter,
24			'crop': CropFilter,
25			'downscale': DownscaleFilter,
26			'relative_resize': RelativeResizeFilter,
27			'rotate': RotateFilter,
28			'thumbnail': ThumbnailFilter,
29			'upscale': UpscaleFilter,
30			'watermark': WatermarkFilter
31			}
32
33			filter_sets = {}
34			adapter = None
35
36			def __init__(self, app=None):
37			"""
38			:param app: Flask application
39			"""
40			if app is not None:
41			self.init_app(app)
42
43			def init_app(self, app):
44			"""
45			:param app: Flask application
46			"""
47			if not hasattr(app, 'extensions'): # pragma: no cover
48			app.extensions = {}
49			app.extensions['imagine'] = self
50
51			self._set_defaults(app)
52
53			if isinstance(app.config['IMAGINE_ADAPTERS'], dict):
54			self.filters.update(app.config['IMAGINE_ADAPTERS'])
55			if isinstance(app.config['IMAGINE_FILTERS'], dict):
56			self.filters.update(app.config['IMAGINE_FILTERS'])
57
58			self._handle_adapter(app)
59			self._handle_filter_sets(app)
60
61			self._add_url_rule(app)
62
63			@classmethod
64			def _set_defaults(cls, app):
65			"""
66			Set default configuration parameters
67			:param app: Flask application
68			:return: Flask application
69			"""
70			app.config.setdefault('IMAGINE_URL', '/media/cache/resolve')
71			app.config.setdefault('IMAGINE_NAME', 'imagine')
72			app.config.setdefault('IMAGINE_CACHE_ENABLED', True)
73
74			app.config.setdefault('IMAGINE_ADAPTERS', {})
75			app.config.setdefault('IMAGINE_FILTERS', {})
76
77			app.config.setdefault('IMAGINE_ADAPTER', {
78			'name': 'fs',
79			'source_folder': 'static',
80			'cache_folder': 'cache'
81			})
82
83			app.config.setdefault('IMAGINE_FILTER_SETS', {})
84
85			return app
86
87			def _handle_adapter(self, app):
88			"""
89			Handle storage adapter configuration
90			:param app: Flask application
91			"""
92			if 'IMAGINE_ADAPTER' in app.config \
93			and 'name' in app.config['IMAGINE_ADAPTER'] \
94			and app.config['IMAGINE_ADAPTER']['name'] in self.adapters.keys():
95			self.adapter = self.adapters[app.config['IMAGINE_ADAPTER']['name']](
96			**app.config['IMAGINE_ADAPTER']
97			)
98			else:
99			raise ValueError('Unknown adapter: %s' % unicode(app.config['IMAGINE_ADAPTER']))
100
101			def _handle_filter_sets(self, app):
102			"""
103			Handle filter sets
104			:param app: Flask application
105			"""
106			if 'IMAGINE_FILTER_SETS' in app.config and isinstance(app.config['IMAGINE_FILTER_SETS'], dict):
107			for filter_name, filters_settings in app.config['IMAGINE_FILTER_SETS'].iteritems():
108			filter_set = []
109			if isinstance(filters_settings, dict) and 'filters' in filters_settings:
110			for filter_type, filter_settings in filters_settings['filters'].iteritems():
111			if filter_type in self.filters:
112			filter_item = self.filters[filter_type](**filter_settings)
113			if isinstance(filter_item, ImagineFilterInterface):
114			filter_set.append(filter_item)
115			else:
116			raise ValueError('Filter must be implement ImagineFilterInterface')
117			else:
118			raise ValueError('Unknown filter type: %s' % filter_type)
119
120			filter_config = {'filters': filter_set}
121			if 'cached' in filters_settings:
122			filter_config['cached'] = filters_settings['cached']
123			else:
124			filter_config['cached'] = app.config['IMAGINE_CACHE_ENABLED']
125
126			self.filter_sets.update({filter_name: filter_config})
127			else:
128			raise ValueError('Wrong settings for filter: %s' % filter_name)
129			else:
130			raise ValueError('Filters configuration does not present')
131
132			def _add_url_rule(self, app):
133			"""
134			Add url rule for get filtered images
135			:param app: Flask application
136			:return: Flask application
137			"""
138			app.url_map.converters['regex'] = RegexConverter
139			app.add_url_rule(
140			app.config['IMAGINE_URL'] + '/<regex("[^\/]+"):filter_name>/<path:path>',
141			app.config['IMAGINE_NAME'],
142			self.handle_request
143			)
144
145			return app
146
147			def handle_request(self, filter_name, path):
148			"""
149			Handle image request
150			:param filter_name: filter_name
151			:param path: image_path
152			:return:
153			"""
154			if filter_name in self.filter_sets:
155			if self.filter_sets[filter_name]['cached']:
156			if self.adapter.check_cached_item('%s/%s' % (filter_name, path)):
157			return redirect(
158			'%s/%s/%s' % (
159			self.adapter.source_folder,
160			self.adapter.cache_folder,
161			'%s/%s' % (filter_name, path)
162			)
163			)
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			return redirect(self.adapter.create_cached_item('%s/%s' % (filter_name, path), resource))
172			else:
173			LOGGER.warning('File "%s" not found.' % path)
174			abort(404)
175			else:
176			LOGGER.warning('Filter "%s" not found.' % filter_name)
177			abort(404)
178
179			def clear_cache(self, path, filter_name=None):
			0 ignored issues – show Coding Style introduced 2016-04-13 22:16 UTC by Report Bug Copy Issue Report This method should have a docstring. The coding style of this project requires that you add a docstring to this code element. Below, you find an example for methods: class SomeClass: def some_method(self): """Do x and return foo.""" If you would like to know more about docstrings, we recommend to read PEP-257: Docstring Conventions. Loading history...
180			if filter_name:
181			self.adapter.remove_cached_item('%s/%s' % (filter_name, path))
182			else:
183			for filter_name in self.filter_sets.iterkeys():
184			self.adapter.remove_cached_item('%s/%s' % (filter_name, path))
185
186
187			def imagine_cache_clear(path, filter_name=None):
188			"""
189			Clear cache for resource path.
190			:param path: str
191			:param filter_name: str or None
192			"""
193			self = current_app.extensions['imagine']
194			self.clear_cache(path, filter_name)
195
196

FlaskGuys / Flask-Imagine

GitHub Access Token became invalid

Push — master ( 598bb5...e05af6 )

Imagine.clear_cache() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like