EventSource.clear() - 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

EventSource.clear() A

↳ Parent: EventSource

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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


import threading
import collections
import warnings
import os.path

import watchdog.observers
import watchdog.observers.polling
import watchdog.events


class Event(list):
    '''
    Event subscription list.

    A list of callable objects. Calling an instance of this will cause a
    call to each item in the list in ascending order by index.

    Usage:
    >>> def f(x):
    ...     print 'f(%s)' % x
    >>> def g(x):
    ...     print 'g(%s)' % x
    >>> e = Event()
    >>> e()
    >>> e.append(f)
    >>> e(123)
    f(123)
    >>> e.remove(f)
    >>> e()
    >>> e += (f, g)
    >>> e(10)
    f(10)
    g(10)
    >>> del e[0]
    >>> e(2)
    g(2)
    '''

    lock_class = threading.Lock
    queue_class = collections.deque

    def __init__(self, iterable=()):
        self._lock = self.lock_class()
        self._queue = self.queue_class()
        super(Event, self).__init__(iterable)

    def __call__(self, *args, **kwargs):
        self._queue.append((args, kwargs))
        while self._queue:
            if self._lock.acquire(False):
                try:
                    args, kwargs = self._queue.popleft()
                    for f in self:
                        f(*args, **kwargs)
                finally:
                    self._lock.release()
            else:
                break

    def __repr__(self):
        return "Event(%s)" % list.__repr__(self)


class EventManager(collections.defaultdict):
    '''
    Dict-like of :class:`Event` objects.

    Usage:
    >>> def f(x):
    ...     print 'f(%s)' % x
    >>> def g(x):
    ...     print 'g(%s)' % x
    >>> m = EventManager()
    >>> 'e' in m
    False
    >>> m.e.append(f)
    >>> 'e' in m
    True
    >>> m['e'](123)
    f(123)
    >>> m['e'].remove(f)
    >>> m['e']()
    >>> m['e'] += (f, g)
    >>> m['e'](10)
    f(10)
    g(10)
    >>> del m['e'][0]
    >>> m['e'](2)
    g(2)
    '''
    def __init__(self, app=None):
        self.app = app
        super(EventManager, self).__init__(Event)


class EventSource(object):
    '''
    Base class for event source classes.

    This serves as both abstract base and public interface reference.
    '''
    def __init__(self, manager, app=None):
        '''
        :param manager: event manager
        :type manager: browsepy.manager.EventManager
        :type app: optional application object, their
                   :attr:`flask.Flask.config` will be honored.
        :type app: flask.Flask.config
        '''
        self.manager = manager
        self.app = app
        self.reload()

    def reload(self):
        '''
        Clean and reload event source.

        This method should respond to :attr:`app` config changes.
        '''
        self.clear()

    def clear(self):
        '''
        Clean event source internal state.

        Event sources must not trigger any event after clear.
        '''
        pass

    @classmethod
    def check(cls, app):
        '''
        Get wether this source should be added to
        :class:`browsepy.manager.EventManager` or not.

        :param app: application object
        :type app: flask.Flask
        :returns: False if should not be added, True otherwise
        :rtype: bool
        '''
        return False


class WatchdogEventSource(EventSource):
    observer_class = watchdog.observers.Observer
    unsupported_observer_classes = {
        watchdog.observers.polling.PollingObserver,
        watchdog.observers.polling.PollingObserverVFS
        }
    event_class = collections.namedtuple(
        'FSEvent', ('type', 'path', 'source', 'is_directory')
        )
    event_map = {
        watchdog.events.EVENT_TYPE_MOVED: 'fs_move',
        watchdog.events.EVENT_TYPE_DELETED: 'fs_remove',
        watchdog.events.EVENT_TYPE_CREATED: 'fs_create',
        watchdog.events.EVENT_TYPE_MODIFIED: 'fs_modify'
        }
    _observer = None

    def __init__(self, manager, app=None):
        self.manager = manager
        self.app = app
        self.reload()

    def dispatch(self, wevent):
        '''
        Handler for watchdog's observer.
        '''
        event = self.event_class(
            self.event_map[wevent.event_type],
            wevent.dest_path if type == 'fs_move' else wevent.src_path,
            wevent.src_path,
            wevent.is_directory
            )
        event_type_specific = '%s_%s' % (
            event.type,
            'directory' if event.is_directory else 'file'
            )
        self.manager['fs_any'](event)
        self.manager[event.type](event)
        self.manager[event_type_specific](event)

    def reload(self):
        '''
        Reload config, create an observer for path specified on
        **base_directory** :attr:`app` config.
        '''
        self.clear()
        path = self.app.config.get('base_directory') if self.app else None
        if not path:
            return
        if not os.path.isdir(path):
            warnings.warn(
                'Path {0!r} is not observable.'.format(path),
                category=RuntimeWarning,
                stacklevel=2
                )
            return
        observer = self.observer_class()
        observer.schedule(self, path, recursive=True)
        observer.start()
        self._observer = observer

    def clear(self):
        '''
        Stop current observer, so no event will be triggered after calling
        this method.
        '''
        observer = self._observer
        if observer:
            observer.unschedule_all()
            observer.stop()
            observer.join()
            self._observer = None

    @classmethod
    def check(cls, app):
        '''
        Get wether this source should be added to
        :class:`browsepy.manager.EventManager` or not.

        :class:`WatchdogEventSource` should not be added if **cache_enable**
        if either False on app config or :class:`watchdog.observers.Observer`
        points to a polling observer class (usually because current OS is not
        supported).

        :param app: application object
        :type app: flask.Flask
        :returns: False if should not be added, True otherwise
        :rtype: bool
        '''
        if app and app.config.get('cache_enable'):
            return cls.observer_class not in cls.unsupported_observer_classes
        return False


1
2			import threading
3			import collections
4			import warnings
5			import os.path
6
7			import watchdog.observers
8			import watchdog.observers.polling
9			import watchdog.events
10
11
12			class Event(list):
13			'''
14			Event subscription list.
15
16			A list of callable objects. Calling an instance of this will cause a
17			call to each item in the list in ascending order by index.
18
19			Usage:
20			>>> def f(x):
21			... print 'f(%s)' % x
22			>>> def g(x):
23			... print 'g(%s)' % x
24			>>> e = Event()
25			>>> e()
26			>>> e.append(f)
27			>>> e(123)
28			f(123)
29			>>> e.remove(f)
30			>>> e()
31			>>> e += (f, g)
32			>>> e(10)
33			f(10)
34			g(10)
35			>>> del e[0]
36			>>> e(2)
37			g(2)
38			'''
39
40			lock_class = threading.Lock
41			queue_class = collections.deque
42
43			def __init__(self, iterable=()):
44			self._lock = self.lock_class()
45			self._queue = self.queue_class()
46			super(Event, self).__init__(iterable)
47
48			def __call__(self, args, *kwargs):
49			self._queue.append((args, kwargs))
50			while self._queue:
51			if self._lock.acquire(False):
52			try:
53			args, kwargs = self._queue.popleft()
54			for f in self:
55			f(args, *kwargs)
56			finally:
57			self._lock.release()
58			else:
59			break
60
61			def __repr__(self):
62			return "Event(%s)" % list.__repr__(self)
63
64
65			class EventManager(collections.defaultdict):
66			'''
67			Dict-like of :class:`Event` objects.
68
69			Usage:
70			>>> def f(x):
71			... print 'f(%s)' % x
72			>>> def g(x):
73			... print 'g(%s)' % x
74			>>> m = EventManager()
75			>>> 'e' in m
76			False
77			>>> m.e.append(f)
78			>>> 'e' in m
79			True
80			>>> m['e'](123)
81			f(123)
82			>>> m['e'].remove(f)
83			>>> m['e']()
84			>>> m['e'] += (f, g)
85			>>> m['e'](10)
86			f(10)
87			g(10)
88			>>> del m['e'][0]
89			>>> m['e'](2)
90			g(2)
91			'''
92			def __init__(self, app=None):
93			self.app = app
94			super(EventManager, self).__init__(Event)
95
96
97			class EventSource(object):
98			'''
99			Base class for event source classes.
100
101			This serves as both abstract base and public interface reference.
102			'''
103			def __init__(self, manager, app=None):
104			'''
105			:param manager: event manager
106			:type manager: browsepy.manager.EventManager
107			:type app: optional application object, their
108			:attr:`flask.Flask.config` will be honored.
109			:type app: flask.Flask.config
110			'''
111			self.manager = manager
112			self.app = app
113			self.reload()
114
115			def reload(self):
116			'''
117			Clean and reload event source.
118
119			This method should respond to :attr:`app` config changes.
120			'''
121			self.clear()
122
123			def clear(self):
124			'''
125			Clean event source internal state.
126
127			Event sources must not trigger any event after clear.
128			'''
129			pass
130
131			@classmethod
132			def check(cls, app):
133			'''
134			Get wether this source should be added to
135			:class:`browsepy.manager.EventManager` or not.
136
137			:param app: application object
138			:type app: flask.Flask
139			:returns: False if should not be added, True otherwise
140			:rtype: bool
141			'''
142			return False
143
144
145			class WatchdogEventSource(EventSource):
146			observer_class = watchdog.observers.Observer
147			unsupported_observer_classes = {
148			watchdog.observers.polling.PollingObserver,
149			watchdog.observers.polling.PollingObserverVFS
150			}
151			event_class = collections.namedtuple(
152			'FSEvent', ('type', 'path', 'source', 'is_directory')
153			)
154			event_map = {
155			watchdog.events.EVENT_TYPE_MOVED: 'fs_move',
156			watchdog.events.EVENT_TYPE_DELETED: 'fs_remove',
157			watchdog.events.EVENT_TYPE_CREATED: 'fs_create',
158			watchdog.events.EVENT_TYPE_MODIFIED: 'fs_modify'
159			}
160			_observer = None
161
162			def __init__(self, manager, app=None):
163			self.manager = manager
164			self.app = app
165			self.reload()
166
167			def dispatch(self, wevent):
168			'''
169			Handler for watchdog's observer.
170			'''
171			event = self.event_class(
172			self.event_map[wevent.event_type],
173			wevent.dest_path if type == 'fs_move' else wevent.src_path,
174			wevent.src_path,
175			wevent.is_directory
176			)
177			event_type_specific = '%s_%s' % (
178			event.type,
179			'directory' if event.is_directory else 'file'
180			)
181			self.manager['fs_any'](event)
182			self.manager[event.type](event)
183			self.manager[event_type_specific](event)
184
185			def reload(self):
186			'''
187			Reload config, create an observer for path specified on
188			base_directory :attr:`app` config.
189			'''
190			self.clear()
191			path = self.app.config.get('base_directory') if self.app else None
192			if not path:
193			return
194			if not os.path.isdir(path):
195			warnings.warn(
196			'Path {0!r} is not observable.'.format(path),
197			category=RuntimeWarning,
198			stacklevel=2
199			)
200			return
201			observer = self.observer_class()
202			observer.schedule(self, path, recursive=True)
203			observer.start()
204			self._observer = observer
205
206			def clear(self):
207			'''
208			Stop current observer, so no event will be triggered after calling
209			this method.
210			'''
211			observer = self._observer
212			if observer:
213			observer.unschedule_all()
214			observer.stop()
215			observer.join()
216			self._observer = None
217
218			@classmethod
219			def check(cls, app):
220			'''
221			Get wether this source should be added to
222			:class:`browsepy.manager.EventManager` or not.
223
224			:class:`WatchdogEventSource` should not be added if cache_enable
225			if either False on app config or :class:`watchdog.observers.Observer`
226			points to a polling observer class (usually because current OS is not
227			supported).
228
229			:param app: application object
230			:type app: flask.Flask
231			:returns: False if should not be added, True otherwise
232			:rtype: bool
233			'''
234			if app and app.config.get('cache_enable'):
235			return cls.observer_class not in cls.unsupported_observer_classes
236			return False
237

ergoithz / browsepy

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

EventSource.clear() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like